[Kimchi-devel] [PATCH V2 4/4] CDROM Management: Guest vm storage devices mockmodel and rest api test cases
Aline Manera
alinefm at linux.vnet.ibm.com
Tue Feb 11 13:46:41 UTC 2014
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 at 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:
More information about the Kimchi-devel
mailing list