Coverage for libs/sm/drivers/DummySR.py: 18%
193 statements
« prev ^ index » next coverage.py v7.3.2, created at 2026-07-10 07:59 +0000
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# DummySR: an example dummy SR for the SDK
17# matches with drivers/DummySR
19import time
21from sm import SR
22from sm import VDI
23from sm.core import util
24from sm.core import xs_errors
26CAPABILITIES = ["SR_PROBE", "VDI_CREATE", "VDI_DELETE", "VDI_ATTACH", "VDI_DETACH",
27 "VDI_ACTIVATE", "VDI_DEACTIVATE", "VDI_CLONE", "VDI_SNAPSHOT", "VDI_RESIZE",
28 "VDI_INTRODUCE", "VDI_MIRROR"]
30CONFIGURATION = []
32DRIVER_INFO = {
33 'name': 'dummy',
34 'description': 'SR plugin which manages fake data',
35 'vendor': 'Citrix Systems Inc',
36 'copyright': '(C) 2008 Citrix Systems Inc',
37 'driver_version': '1.0',
38 'required_api_version': '1.1',
39 'capabilities': CAPABILITIES,
40 'configuration': CONFIGURATION
41 }
43TYPE = 'dummy'
46class DummySR(SR.SR):
47 """dummy storage repository"""
49 def handles(type):
50 if type == TYPE:
51 return True
52 return False
53 handles = staticmethod(handles)
55 def load(self, sr_uuid):
56 self.sr_vditype = 'phy'
58 def content_type(self, sr_uuid):
59 return super(DummySR, self).content_type(sr_uuid)
61 def create(self, sr_uuid, size):
62 self._assertValues(['sr_uuid', 'args', 'host_ref', 'session_ref', 'device_config', 'command', 'sr_ref'])
63 assert(len(self.srcmd.params['args']) == 1)
65 def delete(self, sr_uuid):
66 self._assertValues(['sr_uuid', 'args', 'host_ref', 'session_ref', 'device_config', 'command', 'sr_ref'])
67 assert(len(self.srcmd.params['args']) == 0)
69 def attach(self, sr_uuid):
70 self._assertValues(['sr_uuid', 'args', 'host_ref', 'session_ref', 'device_config', 'command', 'sr_ref'])
71 assert(len(self.srcmd.params['args']) == 0)
73 def detach(self, sr_uuid):
74 self._assertValues(['sr_uuid', 'args', 'host_ref', 'session_ref', 'device_config', 'command', 'sr_ref'])
75 assert(len(self.srcmd.params['args']) == 0)
77 def probe(self):
78 # N.B. There are no SR references
79 self._assertValues(['args', 'host_ref', 'session_ref', 'device_config', 'command'])
80 assert(len(self.srcmd.params['args']) == 0)
82 # Create some Dummy SR records
83 entry = {}
84 entry['size'] = 1024
85 SRlist = {}
86 SRlist[util.gen_uuid()] = entry
88 # Return the Probe XML
89 return util.SRtoXML(SRlist)
91 def vdi(self, uuid):
92 return DummyVDI(self, uuid)
94 def scan(self, sr_uuid):
95 self._assertValues(['sr_uuid', 'args', 'host_ref', 'session_ref', 'device_config', 'command', 'sr_ref'])
96 assert(len(self.srcmd.params['args']) == 0)
98 # The list of VDIs comes from the XenAPI - we have no state
99 for v in self._getallVDIrecords():
100 x = DummyVDI(self, v['uuid'])
101 x.size = v['virtual_size']
102 x.utilisation = v['physical_utilisation']
103 self.vdis[x.uuid] = x
105 self.physical_size = 2000000000000
106 self.physical_utilisation = 0
107 self.virtual_allocation = 0
108 return super(DummySR, self).scan(sr_uuid)
110 def _assertValues(self, vals):
111 for attr in vals:
112 assert(attr in self.srcmd.params)
113 util.SMlog("%s param %s: [%s]" % (self.cmd, attr, self.srcmd.params[attr]))
115 # Iterate through the device_config dictionary
116 for key in self.dconf.keys():
117 util.SMlog("\tdevice_config: [%s:%s]" % (key, self.dconf[key]))
119 # Query the sm_config; parameters can be set at Create time. Iterate through keys
120 self.sm_config = self.session.xenapi.SR.get_sm_config(self.sr_ref)
121 for key in self.sm_config.keys():
122 util.SMlog("\tsm_config: [%s:%s]" % (key, self.sm_config[key]))
124 def _getallVDIrecords(self):
125 """Helper function which returns a list of all VDI records for this SR
126 stored in the XenAPI server"""
127 # Returns a list of (reference, record) pairs: we only need the records
128 vdis = list(self.session.VDI.get_all_records(self.session_ref)['Value'].values())
129 # We only need the VDIs corresponding to this SR
130 return [v for v in vdis if v['SR'] == self.sr_ref]
133class DummyVDI(VDI.VDI):
134 def load(self, vdi_uuid):
135 self.path = "/dev/null" # returned on attach
136 self.uuid = vdi_uuid
137 self.size = 0
138 self.utilisation = 0
139 self.location = vdi_uuid
140 self.sm_config = {}
142 def create(self, sr_uuid, vdi_uuid, size):
143 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_sm_config'])
144 assert(len(self.sr.srcmd.params['args']) == 8)
146 self.vdi_sm_config = self.sr.srcmd.params['vdi_sm_config']
147 for key in self.vdi_sm_config.keys():
148 util.SMlog("\tvdi_sm_config: [%s:%s]" % (key, self.vdi_sm_config[key]))
150 for v in self.sr._getallVDIrecords():
151 if v['uuid'] == vdi_uuid:
152 raise xs_errors.XenError('VDIExists')
154 self.size = size
155 self.utilisation = size
156 self.sm_config['samplekey'] = "This is a dummy SR VDI"
157 self._db_introduce()
158 self.run_corner_cases_tests()
159 return self.get_params()
161 def delete(self, sr_uuid, vdi_uuid):
162 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_ref', 'vdi_location', 'vdi_uuid'])
163 assert(len(self.sr.srcmd.params['args']) == 0)
165 # Assert that the VDI record exists
166 self.session.VDI.get_record(self.sr.session_ref)
167 self.run_corner_cases_tests()
168 self._db_forget()
170 def introduce(self, sr_uuid, vdi_uuid):
171 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_sm_config', 'new_uuid'])
172 assert(len(self.sr.srcmd.params['args']) == 0)
173 self.vdi_sm_config = self.sr.srcmd.params['vdi_sm_config']
174 for key in self.vdi_sm_config.keys():
175 util.SMlog("\tvdi_sm_config: [%s:%s]" % (key, self.vdi_sm_config[key]))
177 for v in self.sr._getallVDIrecords():
178 if v['uuid'] == vdi_uuid:
179 raise xs_errors.XenError('VDIExists')
180 self.uuid = vdi_uuid
181 self.location = self.sr.srcmd.params['vdi_location']
182 self._db_introduce()
183 self.run_corner_cases_tests()
184 return super(DummyVDI, self).get_params()
186 def attach(self, sr_uuid, vdi_uuid):
187 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_ref', 'vdi_location', 'vdi_uuid'])
188 assert(len(self.sr.srcmd.params['args']) == 1)
189 vdi = super(DummyVDI, self).attach(sr_uuid, vdi_uuid)
190 self.run_corner_cases_tests()
191 return vdi
193 def detach(self, sr_uuid, vdi_uuid):
194 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_ref', 'vdi_location', 'vdi_uuid'])
195 self.run_corner_cases_tests()
196 assert(len(self.sr.srcmd.params['args']) == 0)
198 def activate(self, sr_uuid, vdi_uuid):
199 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_ref', 'vdi_location', 'vdi_uuid'])
200 assert(len(self.sr.srcmd.params['args']) == 1)
201 self.vdi_ref = self.sr.srcmd.params['vdi_ref']
202 self.other_config = self.session.xenapi.VDI.get_other_config(self.vdi_ref)
203 self.run_corner_cases_tests()
204 for key in self.other_config.keys():
205 util.SMlog("\tvdi_other_config: [%s:%s]" % (key, self.other_config[key]))
207 def deactivate(self, sr_uuid, vdi_uuid):
208 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_ref', 'vdi_location', 'vdi_uuid'])
209 self.run_corner_cases_tests()
210 assert(len(self.sr.srcmd.params['args']) == 0)
212 def resize(self, sr_uuid, vdi_uuid, size):
213 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref', 'vdi_ref', 'vdi_location', 'vdi_uuid'])
214 assert(len(self.sr.srcmd.params['args']) == 1)
216 self.size = size
217 self.utilisation = size
218 self._db_update()
219 self.run_corner_cases_tests()
220 return super(DummyVDI, self).get_params()
222 def snapshot(self, sr_uuid, vdi_uuid):
223 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref'])
224 assert(len(self.sr.srcmd.params['args']) == 0)
226 dest = util.gen_uuid()
227 vdi = VDI.VDI(self.sr, dest)
228 vdi.read_only = True
229 vdi.location = dest
230 vdi.size = 0
231 vdi.utilisation = 0
232 vdi._db_introduce()
233 self.run_corner_cases_tests()
234 return vdi.get_params()
236 def clone(self, sr_uuid, vdi_uuid):
237 self.sr._assertValues(['sr_uuid', 'args', 'host_ref', 'device_config', 'command', 'sr_ref'])
238 assert(len(self.sr.srcmd.params['args']) == 0)
240 dest = util.gen_uuid()
241 vdi = VDI.VDI(self.sr, dest)
242 vdi.read_only = False
243 vdi.location = dest
244 vdi.size = 0
245 vdi.utilisation = 0
246 vdi._db_introduce()
247 self.run_corner_cases_tests()
248 return vdi.get_params()
250 def check_no_other_vdi_operation_in_progress(self):
251 vdis = util.list_VDI_records_in_sr(self.sr)
252 vdi_ref = self.session.xenapi.VDI.get_by_uuid(self.uuid)
253 del vdis[vdi_ref]
254 active_vdis = [v for v in list(vdis.values()) if v['current_operations'] != {}]
255 if len(active_vdis) != 0:
256 msg = "LVHDRT: found other operations in progress for VDI: %s" % active_vdis[0]['uuid']
257 util.SMlog(msg)
258 raise xs_errors.XenError('VDIInUse')
260 def get_attached_vbds(self):
261 vdi_ref = self.session.xenapi.VDI.get_by_uuid(self.uuid)
262 vbds = self.session.xenapi.VBD.get_all_records_where("field \"VDI\" = \"%s\"" % vdi_ref)
263 return [v for v in list(vbds.values()) if v['currently_attached'] == "true"]
265 def check_vbd_list_is_stable(self, attached_vbds):
266 newly_attached_vbds = self.get_attached_vbds()
267 old_set = set(attached_vbds)
268 new_set = set(newly_attached_vbds)
269 diff_set = old_set.difference(new_set) | new_set.difference(old_set)
270 if len(diff_set) != 0:
271 msg = "LVHDRT: found a non-stable VBD: %s" % (diff_set.pop())
272 util.SMlog(msg)
273 raise xs_errors.XenError('VBDListNotStable')
275 def run_corner_cases_tests(self):
277 def fn():
278 attached_vbds = self.get_attached_vbds()
279 for i in range(0, 10):
280 time.sleep(2)
281 self.check_no_other_vdi_operation_in_progress()
282 self.check_vbd_list_is_stable(attached_vbds)
284 util.fistpoint.activate_custom_fn("LVHDRT_xapiSM_serialization_tests", fn)
286# SR registration at import
287SR.registerSR(DummySR)