[PATCH 0/2] Template tests
Aline Manera (2): bug fix: Allow adding a iSCSI/SCSI volume from a non-ASCII pool to a template Template tests src/kimchi/model/templates.py | 4 +- tests/test_mockmodel.py | 70 +-------- tests/test_model.py | 238 ----------------------------- tests/test_rest.py | 337 ---------------------------------------- tests/test_template.py | 348 ++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 351 insertions(+), 646 deletions(-) create mode 100644 tests/test_template.py -- 2.1.0
When setting a iSCSI/SCSI pool to a template, the user must also specify the volume as those pools are read-only. But it was not possible when the targeted pool name had non-ASCII characteres. That is because the libvirt API is based on string and Kimchi on unicode. So when using Kimchi internal functions we must use unicode. To fix it convert the string returned by libvrit API to unicode. Signed-off-by: Aline Manera <alinefm@linux.vnet.ibm.com> --- src/kimchi/model/templates.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/kimchi/model/templates.py b/src/kimchi/model/templates.py index 85f0839..e91636b 100644 --- a/src/kimchi/model/templates.py +++ b/src/kimchi/model/templates.py @@ -1,7 +1,7 @@ # # Project Kimchi # -# Copyright IBM, Corp. 2014 +# Copyright IBM, Corp. 2014-2015 # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -111,7 +111,7 @@ class TemplatesModel(object): def template_volume_validate(self, tmp_volumes, pool): kwargs = {'conn': self.conn, 'objstore': self.objstore} pool_type = xpath_get_text(pool.XMLDesc(0), "/pool/@type")[0] - pool_name = pool.name() + pool_name = unicode(pool.name(), 'utf-8') # as we discussion, we do not mix disks from 2 different types of # storage pools, for instance: we do not create a template with 2 -- 2.1.0
Reviewed-by: Royce Lv<lvroyce@linux.vnet.ibm.com> On 03/05/2015 09:50 AM, Aline Manera wrote:
When setting a iSCSI/SCSI pool to a template, the user must also specify the volume as those pools are read-only. But it was not possible when the targeted pool name had non-ASCII characteres. That is because the libvirt API is based on string and Kimchi on unicode. So when using Kimchi internal functions we must use unicode. To fix it convert the string returned by libvrit API to unicode.
Signed-off-by: Aline Manera <alinefm@linux.vnet.ibm.com> --- src/kimchi/model/templates.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/src/kimchi/model/templates.py b/src/kimchi/model/templates.py index 85f0839..e91636b 100644 --- a/src/kimchi/model/templates.py +++ b/src/kimchi/model/templates.py @@ -1,7 +1,7 @@ # # Project Kimchi # -# Copyright IBM, Corp. 2014 +# Copyright IBM, Corp. 2014-2015 # # This library is free software; you can redistribute it and/or # modify it under the terms of the GNU Lesser General Public @@ -111,7 +111,7 @@ class TemplatesModel(object): def template_volume_validate(self, tmp_volumes, pool): kwargs = {'conn': self.conn, 'objstore': self.objstore} pool_type = xpath_get_text(pool.XMLDesc(0), "/pool/@type")[0] - pool_name = pool.name() + pool_name = unicode(pool.name(), 'utf-8')
# as we discussion, we do not mix disks from 2 different types of # storage pools, for instance: we do not create a template with 2
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@linux.vnet.ibm.com> --- 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']) -- 2.1.0
Reviewed-by: Royce Lv<lvroyce@linux.vnet.ibm.com> On 03/05/2015 09:50 AM, Aline Manera wrote:
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@linux.vnet.ibm.com> --- 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'])
participants (2)
-
Aline Manera -
Royce Lv