Re: [Kimchi-devel] [PATCH 2/2] Template tests

Create a test_template.py file to handle all the template tests. There is no need to differ model and mockmodel in this case as the Template code is the same in both models. Signed-off-by: Aline Manera <alinefm(a)linux.vnet.ibm.com&gt; --- tests/test_mockmodel.py | 70 +--------- tests/test_model.py | 238 --------------------------------- tests/test_rest.py | 337 ---------------------------------------------- tests/test_template.py | 348 ++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 349 insertions(+), 644 deletions(-) create mode 100644 tests/test_template.py diff --git a/tests/test_mockmodel.py b/tests/test_mockmodel.py index 29354aa..542f845 100644 --- a/tests/test_mockmodel.py +++ b/tests/test_mockmodel.py @@ -1,7 +1,7 @@ # # Project Kimchi # -# Copyright IBM, Corp. 2013-2014 +# Copyright IBM, Corp. 2013-2015 # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -96,74 +96,6 @@ class MockModelTests(unittest.TestCase): else: self.fail("Expected exception not raised") - def test_template_cpu_info(self): - template = self._create_default_template() - # GET of cpu_info will be {} - cpu_info = template['cpu_info'] - self.assertEquals(cpu_info, {}) - self.assertEquals(cpu_info.get('topology'), None) - - # Update topology - # GET of cpu_info will contain 'topology' - cpu_info_data = {'cpu_info': {'topology': {'sockets': 1, - 'cores': 1, - 'threads': 1}}} - _, resp_code = self._send_url_request('PUT', '/templates/test', - cpu_info_data) - self.assertEquals(200, resp_code) - - updated_template, resp_code = \ - self._send_url_request('GET', '/templates/test') - self.assertEquals(200, resp_code) - self.assertEquals(updated_template['cpu_info'], - cpu_info_data['cpu_info']) - - def test_template_update_disk_type(self): - def _get_default_disk_data(disk_type): - return {'disks': [{'index': 0, 'format': disk_type, 'size': 10}]} - - template = self._create_default_template() - # Default template is created with 1 disk without any declared - # type. - disk_data = template['disks'] - self.assertEquals(disk_data, [{'index': 0, 'size': 10}]) - - # For all supported types, edit the template and check if - # the change was made. - disk_types = ['bochs', 'cloop', 'cow', 'dmg', 'qcow', 'qcow2', - 'qed', 'raw', 'vmdk', 'vpc'] - for disk_type in disk_types: - disk_data = _get_default_disk_data(disk_type) - _, resp_code = self._send_url_request('PUT', '/templates/test', - disk_data) - self.assertEquals(200, resp_code) - - updated_template, resp_code = \ - self._send_url_request('GET', '/templates/test') - self.assertEquals(200, resp_code) - self.assertEquals(updated_template['disks'], disk_data['disks']) - - # Check Bad Request when type is invalid - bad_disk_data = _get_default_disk_data('invalid_disk_type') - _, resp_code = self._send_url_request('PUT', '/templates/test', - bad_disk_data) - self.assertEquals(400, resp_code) - - def _create_default_template(self): - params = {'name': 'test', 'cdrom': fake_iso} - template, resp_code = self._send_url_request('POST', '/templates', - params) - self.assertEquals(201, resp_code) - return template - - def _send_url_request(self, method, url, data=None): - req = None - if data: - req = json.dumps(data) - resp = request(host, ssl_port, url, req, method) - rsp_body = resp.read() - return json.loads(rsp_body), resp.status - def test_screenshot_refresh(self): # Create a VM req = json.dumps({'name': 'test', 'cdrom': fake_iso}) diff --git a/tests/test_model.py b/tests/test_model.py index f80f1c9..f37a33c 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -558,244 +558,6 @@ class ModelTests(unittest.TestCase): self.assertTrue(os.access(disk_path, os.F_OK)) self.assertFalse(os.access(disk_path, os.F_OK)) - @unittest.skipUnless(utils.running_as_root(), 'Must be run as root') - def test_template_storage_customise(self): - inst = model.Model(objstore_loc=self.tmp_store) - - with RollbackContext() as rollback: - path = '/tmp/kimchi-images' - pool = 'test-pool' - if not os.path.exists(path): - os.mkdir(path) - - params = {'name': 'test', 'disks': [{'size': 1}], - 'cdrom': self.kimchi_iso} - inst.templates_create(params) - rollback.prependDefer(inst.template_delete, 'test') - - params = {'storagepool': '/storagepools/test-pool'} - self.assertRaises(InvalidParameter, inst.template_update, - 'test', params) - - # For all supported formats, edit the template and check if - # the change was made. - disk_formats = ['bochs', 'cloop', 'cow', 'dmg', 'qcow', 'qcow2', - 'qed', 'raw', 'vmdk', 'vpc'] - for disk_format in disk_formats: - disk_data = {'disks': [{'index': 0, 'format': disk_format, - 'size': 1}]} - inst.template_update('test', disk_data) - updated_template = inst.template_lookup('test') - self.assertEquals(updated_template['disks'], - disk_data['disks']) - # Restore disk data to default value - inst.template_update('test', {'disks': [{'index': 0, 'size': 1}]}) - - args = {'name': pool, - 'path': path, - 'type': 'dir'} - inst.storagepools_create(args) - rollback.prependDefer(shutil.rmtree, args['path']) - rollback.prependDefer(inst.storagepool_delete, pool) - - inst.template_update('test', params) - - params = {'name': 'test-vm-1', 'template': '/templates/test'} - self.assertRaises(InvalidParameter, inst.vms_create, params) - - inst.storagepool_activate(pool) - rollback.prependDefer(inst.storagepool_deactivate, pool) - - inst.vms_create(params) - rollback.prependDefer(inst.vm_delete, 'test-vm-1') - vm_info = inst.vm_lookup(params['name']) - disk_path = '/tmp/kimchi-images/%s-0.img' % vm_info['uuid'] - self.assertTrue(os.access(disk_path, os.F_OK)) - vol = '%s-0.img' % vm_info['uuid'] - volinfo = inst.storagevolume_lookup(pool, vol) - self.assertEquals(1, volinfo['ref_cnt']) - - # reset template to default storage pool - # so we can remove the storage pool created 'test-pool' - params = {'storagepool': '/storagepools/default'} - inst.template_update('test', params) - - @unittest.skipUnless(utils.running_as_root(), 'Must be run as root') - def test_template_create(self): - inst = model.Model('test:///default', - objstore_loc=self.tmp_store) - # Test non-exist path raises InvalidParameter - params = {'name': 'test', - 'cdrom': '/non-exsitent.iso'} - self.assertRaises(InvalidParameter, inst.templates_create, params) - - # Test non-iso path raises InvalidParameter - params['cdrom'] = os.path.abspath(__file__) - self.assertRaises(InvalidParameter, inst.templates_create, params) - - with RollbackContext() as rollback: - net_name = 'test-network' - net_args = {'name': net_name, - 'connection': 'nat', - 'subnet': '127.0.100.0/24'} - inst.networks_create(net_args) - rollback.prependDefer(inst.network_delete, net_name) - - params = {'name': 'test', 'memory': 1024, 'cpus': 1, - 'cdrom': self.kimchi_iso} - inst.templates_create(params) - rollback.prependDefer(inst.template_delete, 'test') - info = inst.template_lookup('test') - for key in params.keys(): - self.assertEquals(params[key], info[key]) - self.assertEquals("default", info["networks"][0]) - - # create template with non-existent network - params['name'] = 'new-test' - params['networks'] = ["no-exist"] - self.assertRaises(InvalidParameter, inst.templates_create, params) - - params['networks'] = ['default', 'test-network'] - inst.templates_create(params) - rollback.prependDefer(inst.template_delete, params['name']) - info = inst.template_lookup(params['name']) - for key in params.keys(): - self.assertEquals(params[key], info[key]) - - @unittest.skipUnless(utils.running_as_root(), 'Must be run as root') - def test_template_integrity(self): - inst = model.Model('test:///default', - objstore_loc=self.tmp_store) - - with RollbackContext() as rollback: - net_name = 'test-network' - net_args = {'name': net_name, - 'connection': 'nat', - 'subnet': '127.0.100.0/24'} - inst.networks_create(net_args) - - path = '/tmp/kimchi-iso/' - if not os.path.exists(path): - os.makedirs(path) - iso = path + 'ubuntu12.04.iso' - iso_gen.construct_fake_iso(iso, True, '12.04', 'ubuntu') - - args = {'name': 'test-pool', - 'path': '/tmp/kimchi-images', - 'type': 'dir'} - inst.storagepools_create(args) - rollback.prependDefer(inst.storagepool_delete, 'test-pool') - - params = {'name': 'test', 'memory': 1024, 'cpus': 1, - 'networks': ['test-network'], 'cdrom': iso, - 'storagepool': '/storagepools/test-pool'} - inst.templates_create(params) - rollback.prependDefer(inst.template_delete, 'test') - - # Try to delete network - # It should fail as it is associated to a template - self.assertRaises(InvalidOperation, inst.network_delete, net_name) - # Update template to release network and then delete it - params = {'networks': []} - inst.template_update('test', params) - inst.network_delete(net_name) - - shutil.rmtree(path) - info = inst.template_lookup('test') - self.assertEquals(info['invalid']['cdrom'], [iso]) - - @unittest.skipUnless(utils.running_as_root(), 'Must be run as root') - def test_template_clone(self): - inst = model.Model(None, - objstore_loc=self.tmp_store) - with RollbackContext() as rollback: - orig_params = {'name': 'test-template', 'memory': 1024, - 'cpus': 1, 'cdrom': self.kimchi_iso} - inst.templates_create(orig_params) - rollback.prependDefer(inst.template_delete, 'test-template') - orig_temp = inst.template_lookup(orig_params['name']) - - ident = inst.template_clone('test-template') - rollback.prependDefer(inst.template_delete, ident) - clone_temp = inst.template_lookup(ident) - - clone_temp['name'] = orig_temp['name'] - for key in clone_temp.keys(): - self.assertEquals(clone_temp[key], orig_temp[key]) - - @unittest.skipUnless(utils.running_as_root(), 'Must be run as root') - def test_template_update(self): - inst = model.Model(None, - objstore_loc=self.tmp_store) - with RollbackContext() as rollback: - net_name = 'test-network' - net_args = {'name': net_name, - 'connection': 'nat', - 'subnet': '127.0.100.0/24'} - inst.networks_create(net_args) - rollback.prependDefer(inst.network_delete, net_name) - inst.network_activate(net_name) - rollback.prependDefer(inst.network_deactivate, net_name) - - net_name = u'kīмсhī-пet' - net_args = {'name': net_name, - 'connection': 'nat', - 'subnet': '127.0.20.0/24'} - inst.networks_create(net_args) - rollback.prependDefer(inst.network_delete, net_name) - inst.network_activate(net_name) - rollback.prependDefer(inst.network_deactivate, net_name) - - orig_params = {'name': 'test', 'memory': 1024, 'cpus': 1, - 'cdrom': self.kimchi_iso} - inst.templates_create(orig_params) - - params = {'name': 'new-test'} - self.assertEquals('new-test', inst.template_update('test', params)) - self.assertRaises(NotFoundError, inst.template_delete, 'test') - - params = {'name': 'new-test', 'memory': 512, 'cpus': 2} - inst.template_update('new-test', params) - rollback.prependDefer(inst.template_delete, 'new-test') - - info = inst.template_lookup('new-test') - for key in params.keys(): - self.assertEquals(params[key], info[key]) - self.assertEquals("default", info["networks"][0]) - - params = {'name': 'new-test', 'memory': 1024, 'cpus': 1, - 'networks': ['default', 'test-network', u'kīмсhī-пet']} - inst.template_update('new-test', params) - info = inst.template_lookup('new-test') - for key in params.keys(): - self.assertEquals(params[key], info[key]) - - # test cpu_info - # new-test has 1 cpu, so this should fail: - params['cpu_info'] = {"topology": - {"sockets": 1, "cores": 1, "threads": 2}} - self.assertRaises(InvalidParameter, inst.template_update, - 'new-test', params) - - params['cpus'] = 2 - inst.template_update('new-test', params) - info = inst.template_lookup('new-test') - for key in params.keys(): - self.assertEquals(params[key], info[key]) - - # test update with non-existent network - params = {'networks': ["no-exist"]} - self.assertRaises(InvalidParameter, inst.template_update, - 'new-test', params) - - params = {'name': 'some-vm', 'template': '/templates/new-test'} - self.assertEquals('some-vm', inst.vms_create(params)) - rollback.prependDefer(inst.vm_delete, 'some-vm') - - iface_args = {'type': 'network', 'network': u'kīмсhī-пet'} - mac = inst.vmifaces_create('some-vm', iface_args) - self.assertEquals(17, len(mac)) - def test_vm_edit(self): config.set("authentication", "method", "pam") inst = model.Model(None, diff --git a/tests/test_rest.py b/tests/test_rest.py index 953cff7..a54e98e 100644 --- a/tests/test_rest.py +++ b/tests/test_rest.py @@ -23,7 +23,6 @@ import json import os import re import requests -import shutil import time import unittest import urllib2 @@ -851,115 +850,6 @@ class RestTests(unittest.TestCase): resp = self.request('/vms/test-vm', '{}', 'DELETE') self.assertEquals(204, resp.status) - def test_template_customise_storage(self): - req = json.dumps({'name': 'test', 'cdrom': fake_iso, - 'disks': [{'size': 1}]}) - resp = self.request('/templates', req, 'POST') - self.assertEquals(201, resp.status) - - # Update a Template with non-existent pool fails with 400 - req = json.dumps({'storagepool': '/storagepools/alt'}) - resp = self.request('/templates/test', req, 'PUT') - self.assertEquals(400, resp.status) - - # Create alternate storage - req = json.dumps({'name': 'alt', - 'capacity': 1024, - 'allocated': 512, - 'path': '/tmp', - 'type': 'dir'}) - resp = self.request('/storagepools', req, 'POST') - self.assertEquals(201, resp.status) - - req = json.dumps({'storagepool': '/storagepools/alt'}) - resp = self.request('/templates/test', req, 'PUT') - self.assertEquals(200, resp.status) - - # Create a VM on inactive pool fails with 400 - req = json.dumps({'name': 'test-vm', 'template': '/templates/test'}) - resp = self.request('/vms', req, 'POST') - self.assertEquals(400, resp.status) - - resp = self.request('/storagepools/alt/activate', req, 'POST') - self.assertEquals(200, resp.status) - - # Create a VM - req = json.dumps({'name': 'test-vm', 'template': '/templates/test'}) - resp = self.request('/vms', req, 'POST') - vm = json.loads(resp.read()) - self.assertEquals(201, resp.status) - - # Verify the volume was created - vol_uri = '/storagepools/alt/storagevolumes/%s-0.img' % vm['uuid'] - resp = self.request(vol_uri) - vol = json.loads(resp.read()) - self.assertEquals(1073741824, vol['capacity']) - - # Delete the VM - resp = self.request('/vms/test-vm', '{}', 'DELETE') - self.assertEquals(204, resp.status) - - # Verify the volume was deleted - self.assertHTTPStatus(404, vol_uri) - - def test_template_customise_network(self): - with RollbackContext() as rollback: - tmpl = {'name': 'test', 'cdrom': fake_iso, - 'disks': [{'size': 1}]} - req = json.dumps(tmpl) - resp = self.request('/templates', req, 'POST') - self.assertEquals(201, resp.status) - # Delete the template - rollback.prependDefer(self.request, - '/templates/test', '{}', 'DELETE') - tmpl_res = json.loads(resp.read()) - self.assertTrue(type(tmpl_res['networks']) is list) - self.assertEquals("default", tmpl_res['networks'][0]) - - tmpl['name'] = "failed_tmpl" - # Create a Template with non-array network fails with 400 - tmpl['networks'] = "test-network" - req = json.dumps(tmpl) - resp = self.request('/templates', req, 'POST') - self.assertEquals(400, resp.status) - - # Create a Template with non-existent network fails with 400 - tmpl['networks'] = ["test-network"] - req = json.dumps(tmpl) - resp = self.request('/templates', req, 'POST') - self.assertEquals(400, resp.status) - - # Create a network - req = json.dumps({'name': 'test-network', - 'connection': 'nat', - 'net': '127.0.1.0/24'}) - resp = self.request('/networks', req, 'POST') - self.assertEquals(201, resp.status) - # Delete the network - rollback.prependDefer(self.request, - '/networks/test-network', '{}', 'DELETE') - - tmpl['name'] = "test" - # Update a Template with non-array network fails with 400 - tmpl['networks'] = "bad-network" - req = json.dumps(tmpl) - resp = self.request('/templates/test', req, 'PUT') - self.assertEquals(400, resp.status) - # Update a Template with non-existent network fails with 400 - tmpl['networks'] = ["bad-network"] - req = json.dumps(tmpl) - resp = self.request('/templates/test', req, 'PUT') - self.assertEquals(400, resp.status) - - # Update a Template with existent network, successful - tmpl['networks'] = ["default", "test-network"] - req = json.dumps(tmpl) - resp = self.request('/templates/test', req, 'PUT') - self.assertEquals(200, resp.status) - tmpl_res = json.loads(resp.read()) - self.assertTrue(type(tmpl_res['networks']) is list) - self.assertEquals(tmpl['networks'], tmpl_res['networks']) - def test_unnamed_vms(self): # Create a Template req = json.dumps({'name': 'test', 'cdrom': fake_iso}) @@ -1029,233 +919,6 @@ class RestTests(unittest.TestCase): resp = self.request('/storagepools/%s' % name, '{}', 'DELETE') self.assertEquals(204, resp.status) - def test_templates(self): - def verify_template(t, res): - for field in ('name', 'os_distro', 'os_version', 'memory', - 'cpus', 'storagepool', 'graphics'): - if field in t: - self.assertEquals(t[field], res[field]) - - resp = self.request('/templates') - self.assertEquals(200, resp.status) - self.assertEquals(0, len(json.loads(resp.read()))) - - # Create a template without cdrom and disk specified fails with 400 - t = {'name': 'test', 'os_distro': 'ImagineOS', - 'os_version': '1.0', 'memory': 1024, 'cpus': 1, - 'storagepool': '/storagepools/alt'} - req = json.dumps(t) - resp = self.request('/templates', req, 'POST') - self.assertEquals(400, resp.status) - - # Create an image based template - open('/tmp/mock.img', 'w').close() - t = {'name': 'test_img_template', 'os_distro': 'ImagineOS', - 'os_version': '1.0', 'memory': 1024, 'cpus': 1, - 'storagepool': '/storagepools/default-pool', - 'disks': [{'base': '/tmp/mock.img'}]} - req = json.dumps(t) - resp = self.request('/templates', req, 'POST') - self.assertEquals(201, resp.status) - os.remove('/tmp/mock.img') - - # Create a template - open('/tmp/mock.iso', 'w').close() - graphics = {'type': 'spice', 'listen': '127.0.0.1'} - t = {'name': 'test', 'os_distro': 'ImagineOS', - 'os_version': '1.0', 'memory': 1024, 'cpus': 1, - 'storagepool': '/storagepools/default-pool', - 'cdrom': '/tmp/mock.iso', 'graphics': graphics} - req = json.dumps(t) - resp = self.request('/templates', req, 'POST') - self.assertEquals(201, resp.status) - os.remove('/tmp/mock.iso') - - # Verify the template - res = json.loads(self.request('/templates/test').read()) - verify_template(t, res) - - # clone a template - resp = self.request('/templates/%s/clone' % t['name'], '{}', 'POST') - self.assertEquals(303, resp.status) - - # Verify the clone template - res = json.loads(self.request('/templates/%s-clone1' % - t['name']).read()) - old_temp = t['name'] - t['name'] = res['name'] - verify_template(t, res) - # Delete the clone template - resp = self.request('/templates/%s' % t['name'], '{}', 'DELETE') - self.assertEquals(204, resp.status) - t['name'] = old_temp - - # Create a template with same name fails with 400 - t = {'name': 'test', 'os_distro': 'ImagineOS', - 'os_version': '1.0', 'memory': 1024, 'cpus': 1, - 'storagepool': '/storagepools/default-pool', - 'cdrom': fake_iso} - req = json.dumps(t) - resp = self.request('/templates', req, 'POST') - self.assertEquals(400, resp.status) - - # Update the template - t['os_distro'] = 'Linux.ISO' - t['os_version'] = '1.1' - t['graphics'] = {'type': 'vnc', 'listen': '127.0.0.1'} - req = json.dumps(t) - resp = self.request('/templates/%s' % t['name'], req, 'PUT') - self.assertEquals(200, resp.status) - - # Verify the template - res = json.loads(self.request('/templates/test').read()) - verify_template(t, res) - - # Update the template with ipv6 address as listen - t['graphics'] = {'type': 'vnc', 'listen': 'fe00::0'} - req = json.dumps(t) - resp = self.request('/templates/%s' % t['name'], req, 'PUT') - self.assertEquals(200, resp.status) - - # Verify the template - res = json.loads(self.request('/templates/test').read()) - verify_template(t, res) - - # Update the template with integer values - t['memory'] = 512 - t['cpus'] = 2 - req = json.dumps(t) - resp = self.request('/templates/%s' % t['name'], req, 'PUT') - self.assertEquals(200, resp.status) - - # Verify the template - res = json.loads(self.request('/templates/%s' % t['name']).read()) - verify_template(t, res) - - # Update the template name - oldname = t['name'] - t['name'] = "test1" - req = json.dumps(t) - resp = self.request('/templates/%s' % oldname, req, 'PUT') - self.assertEquals(303, resp.status) - - # Verify the template - res = json.loads(self.request('/templates/%s' % t['name']).read()) - verify_template(t, res) - - # Try to change template name to empty string - t = {"name": "test1"} - tmpl_name = t['name'] - t['name'] = ' ' - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - # Get the right template name back. - t['name'] = tmpl_name - - # Try to change template memory to a non-number value - t['memory'] = 'invalid-value' - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - - # Try to clean up template memory value - t['memory'] = ' ' - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - - # Try to change template cpus to a non-number value - t['cpus'] = 'invalid-value' - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - - # Try to change template graphics type to invalid value - t['graphics'] = {'type': 'invalid'} - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - - # Try to change template graphics type to invalid listen - t['graphics'] = {'type': 'vnc', 'listen': 'invalid'} - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - - # Try to clean up template cpus value - t['cpus'] = ' ' - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - - # Test nonexistent fields, specify a field 'foo' isn't in the Template - t['foo'] = "bar" - req = json.dumps(t) - resp = self.request('/templates/%s' % tmpl_name, req, 'PUT') - self.assertEquals(400, resp.status) - - # Delete the template - resp = self.request('/templates/%s' % tmpl_name, '{}', 'DELETE') - self.assertEquals(204, resp.status) - - def test_template_integrity(self): - - path = '/tmp/kimchi-iso/' - if not os.path.exists(path): - os.makedirs(path) - iso = path + 'ubuntu12.04.iso' - iso_gen.construct_fake_iso(iso, True, '12.04', 'ubuntu') - - req = json.dumps({'name': 'test-network', - 'connection': 'nat', - 'net': '127.0.1.0/24'}) - resp = request(host, ssl_port, '/networks', req, 'POST') - self.assertEquals(201, resp.status) - - req = json.dumps({'name': 'test-storagepool', - 'path': '/tmp/kimchi-images', - 'type': 'dir'}) - resp = request(host, ssl_port, '/storagepools', req, 'POST') - self.assertEquals(201, resp.status) - - t = {'name': 'test', 'memory': 1024, 'cpus': 1, - 'networks': ['test-network'], 'cdrom': iso, - 'storagepool': '/storagepools/test-storagepool'} - - req = json.dumps(t) - resp = self.request('/templates', req, 'POST') - self.assertEquals(201, resp.status) - - shutil.rmtree(path) - # Try to delete network - # It should fail as it is associated to a template - resp = json.loads(request(host, ssl_port, '/networks/test-network', - '{}', 'DELETE').read()) - self.assertIn("KCHNET0017E", resp["reason"]) - - # Update template to release network and then delete it - params = {'networks': []} - req = json.dumps(params) - resp = request(host, ssl_port, '/templates/test', req, 'PUT') - resp = request(host, ssl_port, '/networks/test-network', '{}', - 'DELETE') - self.assertEquals(204, resp.status) - - # Try to delete the storagepool - # It should fail as it is associated to a template - resp = request(host, ssl_port, '/storagepools/test-storagepool', - '{}', 'DELETE') - self.assertEquals(400, resp.status) - - # Verify the template - res = json.loads(self.request('/templates/test').read()) - self.assertEquals(res['invalid']['cdrom'], [iso]) - - # Delete the template - resp = request(host, ssl_port, '/templates/test', '{}', 'DELETE') - self.assertEquals(204, resp.status) - def test_iso_scan_shallow(self): # fake environment preparation self._create_pool('pool-3') diff --git a/tests/test_template.py b/tests/test_template.py new file mode 100644 index 0000000..4446025 --- /dev/null +++ b/tests/test_template.py @@ -0,0 +1,348 @@ +# -*- coding: utf-8 -*- +# +# Project Kimchi +# +# Copyright IBM, Corp. 2015 +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +import json +import os +import unittest + +from functools import partial + +from kimchi.config import READONLY_POOL_TYPE +from kimchi.mockmodel import MockModel +from utils import get_free_port, patch_auth, request, run_server + + +model = None +test_server = None +host = None +port = None +ssl_port = None +cherrypy_port = None + + +def setUpModule(): + global test_server, model, host, port, ssl_port, cherrypy_port + + patch_auth() + model = MockModel('/tmp/obj-store-test') + host = '127.0.0.1' + port = get_free_port('http') + ssl_port = get_free_port('https') + cherrypy_port = get_free_port('cherrypy_port') + test_server = run_server(host, port, ssl_port, test_mode=True, + cherrypy_port=cherrypy_port, model=model) + + +def tearDownModule(): + test_server.stop() + os.unlink('/tmp/obj-store-test') + + +class TemplateTests(unittest.TestCase): + def setUp(self): + self.request = partial(request, host, ssl_port) + model.reset() + + def test_tmpl_lifecycle(self): + resp = self.request('/templates') + self.assertEquals(200, resp.status) + self.assertEquals(0, len(json.loads(resp.read()))) + + # Create a template without cdrom and disk specified fails with 400 + t = {'name': 'test', 'os_distro': 'ImagineOS', + 'os_version': '1.0', 'memory': 1024, 'cpus': 1, + 'storagepool': '/storagepools/alt'} + req = json.dumps(t) + resp = self.request('/templates', req, 'POST') + self.assertEquals(400, resp.status) + + # Create a template + t = {'name': 'test', 'cdrom': '/tmp/mock.iso'} + req = json.dumps(t) + resp = self.request('/templates', req, 'POST') + self.assertEquals(201, resp.status) + + # Verify the template + keys = ['name', 'icon', 'invalid', 'os_distro', 'os_version', 'cpus', + 'memory', 'cdrom', 'disks', 'storagepool', 'networks', + 'folder', 'graphics', 'cpu_info'] + tmpl = json.loads(self.request('/templates/test').read()) + self.assertEquals(sorted(tmpl.keys()), sorted(keys)) + + # Clone a template + resp = self.request('/templates/test/clone', '{}', 'POST') + self.assertEquals(303, resp.status) + + # Verify the cloned template + tmpl_cloned = json.loads(self.request('/templates/test-clone1').read()) + del tmpl['name'] + del tmpl_cloned['name'] + self.assertEquals(tmpl, tmpl_cloned) + + # Delete the cloned template + resp = self.request('/templates/test-clone1', '{}', 'DELETE') + self.assertEquals(204, resp.status) + + # Create a template with same name fails with 400 + req = json.dumps({'name': 'test', 'cdrom': '/tmp/mock.iso'}) + resp = self.request('/templates', req, 'POST') + self.assertEquals(400, resp.status) + + # Create an image based template + open('/tmp/mock.img', 'w').close() + t = {'name': 'test_img_template', 'disks': [{'base': '/tmp/mock.img'}]} + req = json.dumps(t) + resp = self.request('/templates', req, 'POST') + self.assertEquals(201, resp.status) + os.remove('/tmp/mock.img') + + def test_customized_tmpl(self): + # Create a template + t = {'name': 'test', 'cdrom': '/tmp/mock.iso'} + req = json.dumps(t) + resp = self.request('/templates', req, 'POST') + self.assertEquals(201, resp.status) + tmpl = json.loads(self.request('/templates/test').read()) + + # Update name + new_name = u'kīмсhīTmpl' + new_tmpl_uri = '/templates/%s' % new_name.encode('utf-8') + req = json.dumps({'name': new_name}) + resp = self.request('/templates/test', req, 'PUT') + self.assertEquals(303, resp.status) + resp = self.request(new_tmpl_uri) + update_tmpl = json.loads(resp.read()) + self.assertEquals(new_name, update_tmpl['name']) + del tmpl['name'] + del update_tmpl['name'] + self.assertEquals(tmpl, update_tmpl) + + # Update icon + req = json.dumps({'icon': 'images/icon-fedora.png'}) + resp = self.request(new_tmpl_uri, req, 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals('images/icon-fedora.png', update_tmpl['icon']) + + # Update os_distro and os_version + req = json.dumps({'os_distro': 'fedora', 'os_version': '21'}) + resp = self.request(new_tmpl_uri, req, 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals('fedora', update_tmpl['os_distro']) + self.assertEquals('21', update_tmpl['os_version']) + + # Update cpus + req = json.dumps({'cpus': 2}) + resp = self.request(new_tmpl_uri, req, 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals(2, update_tmpl['cpus']) + + # Update memory + req = json.dumps({'memory': 2048}) + resp = self.request(new_tmpl_uri, req, 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals(2048, update_tmpl['memory']) + + # Update cpu_info + resp = self.request(new_tmpl_uri) + cpu_info = json.loads(resp.read())['cpu_info'] + self.assertEquals(cpu_info, {}) + self.assertEquals(cpu_info.get('topology'), None) + + cpu_info_data = {'cpu_info': {'topology': {'sockets': 1, + 'cores': 2, + 'threads': 1}}} + resp = self.request(new_tmpl_uri, json.dumps(cpu_info_data), 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals(update_tmpl['cpu_info'], cpu_info_data['cpu_info']) + + # Update cdrom + cdrom_data = {'cdrom': '/tmp/mock2.iso'} + resp = self.request(new_tmpl_uri, json.dumps(cdrom_data), 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals(update_tmpl['cdrom'], cdrom_data['cdrom']) + + # Update disks + disk_data = {'disks': [{'index': 0, 'size': 10}, + {'index': 1, 'size': 20}]} + resp = self.request(new_tmpl_uri, json.dumps(disk_data), 'PUT') + self.assertEquals(200, resp.status) + resp = self.request(new_tmpl_uri) + self.assertEquals(200, resp.status) + updated_tmpl = json.loads(resp.read()) + self.assertEquals(updated_tmpl['disks'], disk_data['disks']) + + # For all supported types, edit the template and check if + # the change was made. + disk_types = ['bochs', 'cloop', 'cow', 'dmg', 'qcow', 'qcow2', + 'qed', 'raw', 'vmdk', 'vpc'] + for disk_type in disk_types: + disk_data = {'disks': [{'index': 0, 'format': disk_type, + 'size': 10}]} + resp = self.request(new_tmpl_uri, json.dumps(disk_data), 'PUT') + self.assertEquals(200, resp.status) + + resp = self.request(new_tmpl_uri) + self.assertEquals(200, resp.status) + updated_tmpl = json.loads(resp.read()) + self.assertEquals(updated_tmpl['disks'], disk_data['disks']) + + # Update folder + folder_data = {'folder': ['mock', 'isos']} + resp = self.request(new_tmpl_uri, json.dumps(folder_data), 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals(update_tmpl['folder'], folder_data['folder']) + + # Update graphics + req = json.dumps({'graphics': {'type': 'spice'}}) + resp = self.request(new_tmpl_uri, req, 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals('spice', update_tmpl['graphics']['type']) + + req = json.dumps({'graphics': {'type': 'vnc', 'listen': 'fe00::0'}}) + resp = self.request(new_tmpl_uri, req, 'PUT') + self.assertEquals(200, resp.status) + update_tmpl = json.loads(resp.read()) + self.assertEquals('vnc', update_tmpl['graphics']['type']) + self.assertEquals('fe00::0', update_tmpl['graphics']['listen']) + + def test_customized_network(self): + # Create a template + t = {'name': 'test', 'cdrom': '/tmp/mock.iso'} + req = json.dumps(t) + resp = self.request('/templates', req, 'POST') + self.assertEquals(201, resp.status) + + # Create networks to be used for testing + networks = [{'name': u'kīмсhī-пet', 'connection': 'isolated'}, + {'name': u'nat-network', 'connection': 'nat'}, + {'name': u'subnet-network', 'connection': 'nat', + 'subnet': '127.0.100.0/24'}] + + # Verify the current system has at least one interface to create a + # bridged network + interfaces = json.loads(self.request('/interfaces?type=nic').read()) + if len(interfaces) > 0: + iface = interfaces[0]['name'] + networks.append({'name': u'bridge-network', 'connection': 'bridge', + 'interface': iface}) + networks.append({'name': u'bridge-network', 'connection': 'bridge', + 'interface': iface, 'vlan_id': 987}) + + tmpl_nets = [] + for net in networks: + self.request('/networks', json.dumps(net), 'POST') + tmpl_nets.append(net['name']) + req = json.dumps({'networks': tmpl_nets}) + resp = self.request('/templates/test', req, 'PUT') + self.assertEquals(200, resp.status) + + def test_customized_storagepool(self): + # Create a template + t = {'name': 'test', 'cdrom': '/tmp/mock.iso'} + req = json.dumps(t) + resp = self.request('/templates', req, 'POST') + self.assertEquals(201, resp.status) + + # MockModel always returns 2 partitions (vdx, vdz) + partitions = json.loads(self.request('/host/partitions').read()) + devs = [dev['path'] for dev in partitions] + + # MockModel always returns 3 FC devices + fc_devs = json.loads(self.request('/host/devices?_cap=fc_host').read()) + fc_devs = [dev['name'] for dev in fc_devs] + + poolDefs = [ + {'type': 'dir', 'name': u'kīмсhīUnitTestDirPool', + 'path': '/tmp/kimchi-images'}, + {'type': 'netfs', 'name': u'kīмсhīUnitTestNSFPool', + 'source': {'host': 'localhost', + 'path': '/var/lib/kimchi/nfs-pool'}}, + {'type': 'scsi', 'name': u'kīмсhīUnitTestSCSIFCPool', + 'source': {'adapter_name': fc_devs[0]}}, + {'type': 'iscsi', 'name': u'kīмсhīUnitTestISCSIPool', + 'source': {'host': '127.0.0.1', + 'target': 'iqn.2015-01.localhost.kimchiUnitTest'}}, + {'type': 'logical', 'name': u'kīмсhīUnitTestLogicalPool', + 'source': {'devices': [devs[0]]}}] + + for pool in poolDefs: + self.request('/storagepools', json.dumps(pool), 'POST') + pool_uri = '/storagepools/%s' % pool['name'].encode('utf-8') + self.request(pool_uri + '/activate', '{}', 'POST') + + req = None + if pool['type'] in READONLY_POOL_TYPE: + resp = self.request(pool_uri + '/storagevolumes') + vols = json.loads(resp.read()) + if len(vols) > 0: + vol = vols[0]['name'] + req = json.dumps({'storagepool': pool_uri, + 'disks': [{'volume': vol}]}) + else: + req = json.dumps({'storagepool': pool_uri}) + + if req is not None: + resp = self.request('/templates/test', req, 'PUT') + self.assertEquals(200, resp.status) + + def test_tmpl_integrity(self): + # Create a network and a pool for testing template integrity + net = {'name': u'nat-network', 'connection': 'nat'} + self.request('/networks', json.dumps(net), 'POST') + + pool = {'type': 'dir', 'name': 'dir-pool', 'path': '/tmp/dir-pool'} + self.request('/storagepools', json.dumps(pool), 'POST') + + # Create a template using the custom network and pool + t = {'name': 'test', 'cdrom': '/tmp/mock.iso', + 'networks': ['nat-network'], + 'storagepool': '/storagepools/dir-pool'} + req = json.dumps(t) + resp = self.request('/templates', req, 'POST') + self.assertEquals(201, resp.status) + + # Try to delete network + # It should fail as it is associated to a template + resp = self.request('/networks/nat-network', '{}', 'DELETE') + self.assertIn("KCHNET0017E", json.loads(resp.read())["reason"]) + + # Update template to release network and then delete it + params = {'networks': []} + req = json.dumps(params) + self.request('/templates/test', req, 'PUT') + resp = self.request('/networks/nat-network', '{}', 'DELETE') + self.assertEquals(204, resp.status) + + # Try to delete the storagepool + # It should fail as it is associated to a template + resp = self.request('/storagepools/dir-pool', '{}', 'DELETE') + self.assertEquals(400, resp.status) + + # Verify the template + res = json.loads(self.request('/templates/test').read()) + self.assertEquals(res['invalid']['cdrom'], ['/tmp/mock.iso'])