Coverage for libs/sm/drivers/udevSR.py: 35%

1# Copyright (C) Citrix Systems Inc. 

2# 

3# This program is free software; you can redistribute it and/or modify 

4# it under the terms of the GNU Lesser General Public License as published 

5# by the Free Software Foundation; version 2.1 only. 

6# 

7# This program is distributed in the hope that it will be useful, 

8# but WITHOUT ANY WARRANTY; without even the implied warranty of 

9# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

10# GNU Lesser General Public License for more details. 

11# 

12# You should have received a copy of the GNU Lesser General Public License 

13# along with this program; if not, write to the Free Software Foundation, Inc., 

14# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA 

15# 

16# udevSR: represents VDIs which are hotplugged into dom0 via udev e.g. 

17# USB CDROM/disk devices 

18# matches with drivers/udevSR 

19 

20from sm import SR 

21from sm import VDI 

22from sm.core import util 

23import os 

24import time 

25import stat 

26from sm.core import xs_errors 

27from sm import sysdevice 

28 

29CAPABILITIES = ["VDI_INTRODUCE", "VDI_ATTACH", "VDI_DETACH", "VDI_UPDATE", "SR_UPDATE"] 

30 

31CONFIGURATION = \ 

32 [['location', 'path to mount (required) (e.g. server:/path)']] 

33 

34DRIVER_INFO = { 

35 'name': 'udev', 

36 'description': 'SR plugin which represents devices plugged in via udev as VDIs', 

37 'vendor': 'Citrix Systems Inc', 

38 'copyright': '(C) 2008 Citrix Systems Inc', 

39 'driver_version': '1.0', 

40 'required_api_version': '1.1', 

41 'capabilities': CAPABILITIES, 

42 'configuration': CONFIGURATION 

43 } 

44 

45TYPE = 'udev' 

46 

47 

48class udevSR(SR.SR): 

49 """udev-driven storage repository""" 

50 

51 def handles(type): 

52 if type == TYPE: 

53 return True 

54 return False 

55 handles = staticmethod(handles) 

56 

57 def content_type(self, sr_uuid): 

58 return super(udevSR, self).content_type(sr_uuid) 

59 

60 def vdi(self, uuid): 

61 util.SMlog("params = %s" % (self.srcmd.params.keys())) 

62 

63 if 'vdi_location' in self.srcmd.params: 63 ↛ 64line 63 didn't jump to line 64, because the condition on line 63 was never true

64 vdi_location = self.srcmd.params['vdi_location'] 

65 else: 

66 vdi_location = self.get_vdi_location(uuid) 

67 

68 return udevVDI(self, vdi_location) 

69 

70 def get_vdi_location(self, uuid): 

71 vdi = self.session.xenapi.VDI 

72 vdi_ref = vdi.get_by_uuid(uuid) 

73 return vdi.get_location(vdi_ref) 

74 

75 def load(self, sr_uuid): 

76 # First of all, check we've got the correct keys in dconf 

77 if 'location' not in self.dconf: 

78 raise xs_errors.XenError('ConfigLocationMissing') 

79 self.sr_vditype = 'phy' 

80 # Cache the sm_config 

81 self.sm_config = self.session.xenapi.SR.get_sm_config(self.sr_ref) 

82 

83 def update(self, sr_uuid): 

84 # Return as much information as we have 

85 sr_root = self.dconf['location'] 

86 

87 if util.pathexists(sr_root): 

88 for filename in os.listdir(sr_root): 

89 path = os.path.join(sr_root, filename) 

90 x = udevVDI(self, path) 

91 self.vdis[path] = x 

92 

93 the_sum = 0 

94 for vdi in self.vdis.values(): 

95 the_sum = the_sum + vdi.size 

96 

97 self.physical_size = the_sum 

98 self.physical_utilisation = the_sum 

99 self.virtual_allocation = the_sum 

100 

101 self._db_update() 

102 

103 def scan(self, sr_uuid): 

104 self.update(sr_uuid) 

105 

106 # base class scan does all the work: 

107 return super(udevSR, self).scan(sr_uuid) 

108 

109 def create(self, sr_uuid, size): 

110 pass 

111 

112 def delete(self, sr_uuid): 

113 pass 

114 

115 def attach(self, sr_uuid): 

116 pass 

117 

118 def detach(self, sr_uuid): 

119 pass 

120 

121 

122def read_whole_file(filename): 

123 f = open(filename, 'r') 

124 try: 

125 return f.readlines() 

126 finally: 

127 f.close() 

128 

129 

130class udevVDI(VDI.VDI): 

131 def __init__(self, sr, location): 

132 self.location = location 

133 VDI.VDI.__init__(self, sr, None) 

134 

135 def load(self, location): 

136 self.path = self.location 

137 self.size = 0 

138 self.utilisation = 0 

139 self.label = self.path 

140 self.sm_config = {} 

141 try: 

142 s = os.stat(self.path) 

143 self.deleted = False 

144 

145 # Use the CTIME of the symlink to mean "time it was hotplugged" 

146 iso8601 = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(s[stat.ST_CTIME])) 

147 self.sm_config['hotplugged_at'] = iso8601 

148 

149 self.path = os.path.realpath(self.path) 

150 

151 dev = os.path.basename(self.path) 

152 info = sysdevice.stat(dev) 

153 if "size" in info.keys(): 

154 self.size = info["size"] 

155 self.utilisation = self.size 

156 

157 self.label = "%s %s" % (info["bus"], info["bus_path"]) 

158 self.description = info["hwinfo"] 

159 

160 # XXX: what other information can we recover? 

161 if 'type' in self.sr.sm_config: 

162 self.read_only = self.sr.sm_config['type'] == "cd" 

163 

164 usb_path = info.get("usb_path") 

165 if usb_path: 

166 self.sm_config["usb_path"] = info["usb_path"] 

167 pusbs = self.session.xenapi.PUSB.get_all_records() 

168 for pusb in pusbs.values(): 

169 if usb_path == pusb.get("path"): 

170 if pusb.get("passthrough_enabled"): 

171 raise xs_errors.XenError('VDIUnavailable') 

172 break 

173 

174 except OSError as e: 

175 self.deleted = True 

176 

177 def introduce(self, sr_uuid, vdi_uuid): 

178 self.uuid = vdi_uuid 

179 self.location = self.sr.srcmd.params['vdi_location'] 

180 self._db_introduce() 

181 # Update the physical_utilisation etc 

182 self.sr.update(sr_uuid) 

183 return super(udevVDI, self).get_params() 

184 

185 def update(self, sr_uuid, vdi_location): 

186 self.load(vdi_location) 

187 # _db_update requires self.uuid to be set 

188 self.uuid = self.sr.srcmd.params['vdi_uuid'] 

189 self._db_update() 

190 # also reset the name-label and description since we're a bit of 

191 # a special SR 

192 # this would lead to an infinite loop as VDI.set_name_label now 

193 # calls VDI.update 

194 # temporarily commenting this to pass quicktest 

195 #vdi = self.sr.session.xenapi.VDI.get_by_uuid(self.uuid) 

196 #self.sr.session.xenapi.VDI.set_name_label(vdi, self.label) 

197 #self.sr.session.xenapi.VDI.set_name_description(vdi, self.description) 

198 

199 def attach(self, sr_uuid, vdi_uuid): 

200 if self.deleted: 

201 raise xs_errors.XenError('VDIUnavailable') 

202 

203 return super(udevVDI, self).attach(sr_uuid, vdi_uuid) 

204 

205 def detach(self, sr_uuid, vdi_uuid): 

206 pass 

207 

208# SR registration at import 

209SR.registerSR(udevSR)