[Kimchi-devel] [PATCH 2/2] Template tests
Aline Manera
alinefm at linux.vnet.ibm.com
Thu Mar 5 14:50:22 UTC 2015
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'])
--
2.1.0
More information about the Kimchi-devel
mailing list