On 02/10/2014 08:46 PM, Rodrigo Trujillo wrote:
This patch implements the mockmodel class to simulate lookup, add, remove, update of devices in a guest vm. Also, it adds test cases to test the rest API.
Signed-off-by: Rodrigo Trujillo <rodrigo.trujillo@linux.vnet.ibm.com> --- src/kimchi/mockmodel.py | 63 ++++++++++++++++++++++++++++++++++++++++++ tests/test_rest.py | 73 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+)
diff --git a/src/kimchi/mockmodel.py b/src/kimchi/mockmodel.py index f6400a4..3161eb0 100644 --- a/src/kimchi/mockmodel.py +++ b/src/kimchi/mockmodel.py @@ -560,6 +560,48 @@ class MockModel(object): def networks_get_list(self): return sorted(self._mock_networks.keys())
+ def storages_create(self, vm_name, params): + path = params.get('path') + if path.startswith('/') and not os.path.exists(path): + msg = "Path specified for device is not valide" + raise InvalidParameter(msg) + + dom = self._get_vm(vm_name) + if params['dev'] in self.storages_get_list(vm_name):
Ops... it will fail if I don't pass a 'dev' to create the cdrom It should be. dev = param.get('dev', None) if dev is not None and dev in self.storages_get_list(vm_name): raise
+ return OperationFailed('Device name already in use.') + vmdev = MockVMStorageDevice(params)
+ dom.storagedevices[params['dev']] = vmdev + return params['dev']
You need to choose a 'dev' if the user does not provide one
+ + def storages_get_list(self, vm_name): + dom = self._get_vm(vm_name) + return dom.storagedevices.keys() + + def storage_lookup(self, vm_name, dev_name): + dom = self._get_vm(vm_name) + if dev_name not in self.storages_get_list(vm_name): + msg = 'The storage device "%s" does not exist in the guest "%s"' \ + % (dev_name,vm_name) + raise NotFoundError(msg) + return dom.storagedevices.get('dev_name').info + + def storage_delete(self, vm_name, dev_name): + dom = self._get_vm(vm_name) + if dev_name not in self.storages_get_list(vm_name): + msg = 'The storage device "%s" does not exist in the guest "%s"' \ + % (dev_name,vm_name) + raise NotFoundError(msg) + dom.storagedevices.pop(dev_name) + + def storage_update(self, vm_name, dev_name, params): + try: + dom = self._get_vm(vm_name) + dom.storagedevices[dev_name].info.update(params) + except Exception as e: + msg = 'Was not possible to update storage device: %s' % e.message + raise OperationFailed(e.message) + return dev_name + def vmifaces_create(self, vm, params): if (params["type"] == "network" and params["network"] not in self.networks_get_list()): @@ -738,6 +780,24 @@ class MockVMTemplate(VMTemplate): return disk_paths
+class MockVMStorageDevice(object): + def __init__(self, params): + # Defaults + if params['dev'] == 'hda': + self.info = {'dev': params.get('dev'), + 'type': 'disk', + 'path': '/tmp/myimage.img'} + elif params['dev'] == 'hdc': + self.info = {'dev': params.get('dev'), + 'type': 'cdrom', + 'path': ''} + # New devices + else: + self.info = {'dev': params.get('dev'), + 'type': params.get('type'), + 'path': params.get('path')} + + class MockVMIface(object): counter = 0
@@ -763,6 +823,9 @@ class MockVM(object): self.disk_paths = [] self.networks = template_info['networks'] ifaces = [MockVMIface(net) for net in self.networks] + default_devices = [{'dev':'hda'}, {'dev':'hdc'}] + self.storagedevices = dict([(dev['dev'], MockVMStorageDevice(dev)) \ + for dev in default_devices]) self.ifaces = dict([(iface.info['mac'], iface) for iface in ifaces]) self.info = {'state': 'shutoff', 'stats': "{'cpu_utilization': 20, 'net_throughput' : 35, \ diff --git a/tests/test_rest.py b/tests/test_rest.py index 8b033ae..3cf87c2 100644 --- a/tests/test_rest.py +++ b/tests/test_rest.py @@ -353,6 +353,79 @@ class RestTests(unittest.TestCase): resp = self.request('/templates/test', '{}', 'DELETE') self.assertEquals(204, resp.status)
+ def test_vm_storage_devices(self): + + with RollbackContext() as rollback: + # Create a template as a base for our VMs + req = json.dumps({'name': 'test', 'cdrom': '/nonexistent.iso'}) + resp = self.request('/templates', req, 'POST') + self.assertEquals(201, resp.status) + # Delete the template + rollback.prependDefer(self.request, + '/templates/test', '{}', 'DELETE') + + # Create a VM with default args + req = json.dumps({'name': 'test-vm', + 'template': '/templates/test'}) + resp = self.request('/vms', req, 'POST') + self.assertEquals(201, resp.status) + # Delete the VM + rollback.prependDefer(self.request, + '/vms/test-vm', '{}', 'DELETE') + + # Attach a storage disk + req = json.dumps({'dev': 'hdx', + 'type': 'disk', + 'path': '/tmp'})
Add a test to create a cdrom without passing the "dev" parameter
+ resp = self.request('/vms/test-vm/storages', req, 'POST') + self.assertEquals(201, resp.status) + # Delete the disk + rollback.prependDefer(self.request, + '/vms/test-vm/storages/hdx', '{}', 'DELETE') + + # Detach storage disk + resp = self.request('/vms/test-vm/storages/hdx', '{}', 'DELETE') + self.assertEquals(204, resp.status) + + # Detach storage disk that does not exist + resp = self.request('/vms/test-vm/storages/hdx', '{}', 'DELETE') + self.assertEquals(404, resp.status) + + # Attach cdrom with nonexistent iso + req = json.dumps({'dev': 'hdx', + 'type': 'cdrom', + 'path': '/tmp/nonexistent.iso'}) + resp = self.request('/vms/test-vm/storages', req, 'POST') + self.assertEquals(400, resp.status) + + open('/tmp/existent.iso', 'w').close() + # Attach a cdrom with existent dev name + req = json.dumps({'dev': 'hdx', + 'type': 'cdrom', + 'path': '/tmp/existent.iso'}) + resp = self.request('/vms/test-vm/storages', req, 'POST') + self.assertEquals(201, resp.status) + # Delete the cdrom + rollback.prependDefer(self.request, + '/vms/test-vm/storages/hdx', '{}', 'DELETE') + + # Change path of storage cdrom + req = json.dumps({'path': 'http://myserver.com/myiso.iso'}) + resp = self.request('/vms/test-vm/storages/hdx', req, 'PUT') + self.assertEquals(200, resp.status) + + # Test GET + devs = json.loads(self.request('/vms/test-vm/storages').read()) + self.assertEquals(3, len(devs)) + + # Detach storage cdrom + resp = self.request('/vms/test-vm/storages/hdx', '{}', 'DELETE') + self.assertEquals(204, resp.status) + + # Test GET + devs = json.loads(self.request('/vms/test-vm/storages').read()) + self.assertEquals(2, len(devs)) + def test_vm_iface(self):
with RollbackContext() as rollback: