[Kimchi-devel] [PATCH 2/2] Template tests
Royce Lv
lvroyce at linux.vnet.ibm.com
Mon Mar 9 08:58:11 UTC 2015
Reviewed-by: Royce Lv<lvroyce at 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 at 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'])
More information about the Kimchi-devel
mailing list