'adapter_name' is a required parameter for a SCSI FC pool. Also add 'persistent' parameter to storage pool information. Signed-off-by: Aline Manera <alinefm@linux.vnet.ibm.com> --- docs/API.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/docs/API.md b/docs/API.md index 67997ea..8c01c2f 100644 --- a/docs/API.md +++ b/docs/API.md @@ -417,7 +417,7 @@ A interface represents available network interface on VM. Pool types: 'iscsi'. * username: Login username of the iSCSI target. * password: Login password of the iSCSI target. - * adapter_name *(optional)*: Scsi host name. + * adapter_name: Scsi host name. ### Resource: Storage Pool @@ -444,9 +444,12 @@ A interface represents available network interface on VM. * nr_volumes: The number of storage volumes for active pools, 0 for inactive pools * autostart: Whether the storage pool will be enabled automatically when the system boots + * persistent: True, when pool persist after a system reboot or be stopped. + All storage pools created by Kimchi are persistent. * source: Source of the storage pool, * addr: mount address of this storage pool(for 'netfs' pool) * path: export path of this storage pool(for 'netfs' pool) + * **PUT**: Set whether the Storage Pool should be enabled automatically when the system boots * autostart: Toggle the autostart flag of the VM. This flag sets whether -- 2.1.0

Update a storage pool with non-ASCII characters was not possible because of the encode/decode error. Fix it. Signed-off-by: Aline Manera <alinefm@linux.vnet.ibm.com> --- src/kimchi/model/storagepools.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/kimchi/model/storagepools.py b/src/kimchi/model/storagepools.py index e03c6bb..b85f3b4 100644 --- a/src/kimchi/model/storagepools.py +++ b/src/kimchi/model/storagepools.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 @@ -79,7 +79,7 @@ class StoragePoolsModel(object): # used before but a volume group will already exist with this name # So check the volume group does not exist to create the pool if params['type'] == 'logical': - vgdisplay_cmd = ['vgdisplay', name] + vgdisplay_cmd = ['vgdisplay', name.encode('utf-8')] output, error, returncode = run_command(vgdisplay_cmd) # From vgdisplay error codes: # 1 error reading VGDA @@ -333,7 +333,7 @@ class StoragePoolModel(object): raise InvalidOperation('KCHPOOL0029E') self._update_lvm_disks(name, params['disks']) ident = pool.name() - return ident + return ident.decode('utf-8') def activate(self, name): pool = self.get_storagepool(name, self.conn) -- 2.1.0

Model does some input validation to make sure a NFS and iSCSI pool will work as expected. To skip that validation on MockModel and be able to create any NFS and iSCSI pool (for testing proposals) the .prepare() must be overridden. Signed-off-by: Aline Manera <alinefm@linux.vnet.ibm.com> --- src/kimchi/mockmodel.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/kimchi/mockmodel.py b/src/kimchi/mockmodel.py index f8e317a..14891ee 100644 --- a/src/kimchi/mockmodel.py +++ b/src/kimchi/mockmodel.py @@ -32,6 +32,8 @@ from kimchi import imageinfo from kimchi import osinfo from kimchi.model.debugreports import DebugReportsModel from kimchi.model.host import DeviceModel +from kimchi.model.libvirtstoragepool import IscsiPoolDef, NetfsPoolDef +from kimchi.model.libvirtstoragepool import StoragePoolDef from kimchi.model.model import Model from kimchi.model.storagevolumes import StorageVolumesModel from kimchi.model.templates import LibvirtVMTemplate @@ -78,6 +80,9 @@ class MockModel(Model): libvirt.virDomain.updateDeviceFlags = MockModel.updateDeviceFlags libvirt.virStorageVol.resize = MockModel.volResize libvirt.virStorageVol.wipePattern = MockModel.volWipePattern + + IscsiPoolDef.prepare = NetfsPoolDef.prepare = StoragePoolDef.prepare + PAMUsersModel.auth_type = 'fake' PAMGroupsModel.auth_type = 'fake' -- 2.1.0

Model does not use libvirt to extend the logical pool. Instead of that, the LVM is extended by using vgextend command. As in MockModel environment we don't have a LVM itself, we need to fake the extend operation. In that case only the pool XML is updated without affecting the current system. Signed-off-by: Aline Manera <alinefm@linux.vnet.ibm.com> --- src/kimchi/mockmodel.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/src/kimchi/mockmodel.py b/src/kimchi/mockmodel.py index affbf42..0c8186c 100644 --- a/src/kimchi/mockmodel.py +++ b/src/kimchi/mockmodel.py @@ -26,6 +26,7 @@ import time import kimchi.model.cpuinfo from lxml import objectify +from lxml.builder import E from kimchi import config from kimchi import imageinfo @@ -35,6 +36,7 @@ from kimchi.model.host import DeviceModel from kimchi.model.libvirtstoragepool import IscsiPoolDef, NetfsPoolDef from kimchi.model.libvirtstoragepool import StoragePoolDef from kimchi.model.model import Model +from kimchi.model.storagepools import StoragePoolModel from kimchi.model.storagevolumes import StorageVolumesModel from kimchi.model.templates import LibvirtVMTemplate from kimchi.model.users import PAMUsersModel @@ -107,6 +109,7 @@ class MockModel(Model): setattr(self, m, mock_method) DeviceModel.lookup = self._mock_device_lookup + StoragePoolModel._update_lvm_disks = self._update_lvm_disks StorageVolumesModel.get_list = self._mock_storagevolumes_get_list DebugReportsModel._gen_debugreport_file = self._gen_debugreport_file LibvirtVMTemplate._get_volume_path = self._get_volume_path @@ -256,6 +259,20 @@ class MockModel(Model): os.rename(tmpf, realf) cb("OK", True) + def _update_lvm_disks(self, pool_name, disks): + conn = self.conn.get() + pool = conn.storagePoolLookupByName(pool_name.encode('utf-8')) + xml = pool.XMLDesc(0) + + root = ET.fromstring(xml) + source = root.xpath('./source')[0] + + for d in disks: + dev = E.device(path=d) + source.append(dev) + + conn.storagePoolDefineXML(ET.tostring(root), 0) + def _mock_storagevolumes_create(self, pool, params): vol_source = ['file', 'url', 'capacity'] index_list = list(i for i in range(len(vol_source)) -- 2.1.0

Create 2 new files: tests/test_mock_storagepool.py and tests/test_storagepool.py. The first one has all the MockModel tests and the latter the Model tests. As most of storage pool can not be tested automatically by Model, all storage pool types are covered on MockModel tests which uses the libvirt Test driver. Signed-off-by: Aline Manera <alinefm@linux.vnet.ibm.com> --- tests/test_mock_storagepool.py | 156 +++++++++++++++++++++++++++++++++++++++++ tests/test_model.py | 70 ------------------ tests/test_rest.py | 120 +------------------------------ tests/test_storagepool.py | 104 +++++++++++++++++++++++++++ tests/utils.py | 11 +++ 5 files changed, 273 insertions(+), 188 deletions(-) create mode 100644 tests/test_mock_storagepool.py create mode 100644 tests/test_storagepool.py diff --git a/tests/test_mock_storagepool.py b/tests/test_mock_storagepool.py new file mode 100644 index 0000000..934fdef --- /dev/null +++ b/tests/test_mock_storagepool.py @@ -0,0 +1,156 @@ +# -*- 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 requests +import unittest + +from functools import partial + +from kimchi.config import paths +from kimchi.mockmodel import MockModel +from utils import fake_auth_header, get_free_port, patch_auth, request +from utils import run_server, wait_task + + +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 MockStoragepoolTests(unittest.TestCase): + def setUp(self): + self.request = partial(request, host, ssl_port) + model.reset() + + def _task_lookup(self, taskid): + return json.loads(self.request('/tasks/%s' % taskid).read()) + + def test_storagepool(self): + # 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]]}}] + + def _do_test(params): + name = params['name'] + uri = '/storagepools/%s' % name.encode('utf-8') + + req = json.dumps(params) + resp = self.request('/storagepools', req, 'POST') + self.assertEquals(201, resp.status) + + # activate the storage pool + resp = self.request(uri + '/activate', '{}', 'POST') + storagepool = json.loads(self.request(uri).read()) + self.assertEquals('active', storagepool['state']) + + # Quick test to verify download and upload features for each pool + # Skip READ-ONLY storage pools (scsi and iscsi) + url = 'https://github.com/kimchi-project/kimchi/raw/master/COPYING' + filepath = os.path.join(paths.get_prefix(), 'COPYING.LGPL') + if params['type'] not in ['scsi', 'iscsi']: + # Test download + req = json.dumps({'url': url}) + resp = self.request(uri + '/storagevolumes', req, 'POST') + self.assertEquals(202, resp.status) + task = json.loads(resp.read()) + wait_task(self._task_lookup, task['id']) + resp = self.request(uri + '/storagevolumes/COPYING') + self.assertEquals(200, resp.status) + + # Test upload + url = 'https://%s:%s' % (host, ssl_port) + url += uri + '/storagevolumes' + with open(filepath, 'rb') as fd: + r = requests.post(url, files={'file': fd}, + verify=False, + headers=fake_auth_header()) + self.assertEquals(r.status_code, 202) + task = r.json() + wait_task(self._task_lookup, task['id']) + resp = self.request(uri + '/storagevolumes/COPYING.LGPL') + self.assertEquals(200, resp.status) + + # Deactivate the storage pool + resp = self.request(uri + '/deactivate', '{}', 'POST') + storagepool = json.loads(self.request(uri).read()) + self.assertEquals('inactive', storagepool['state']) + + # Set autostart flag of the storage pool + for autostart in [True, False]: + t = {'autostart': autostart} + req = json.dumps(t) + resp = self.request(uri, req, 'PUT') + storagepool = json.loads(self.request(uri).read()) + self.assertEquals(autostart, storagepool['autostart']) + + # Extend logical pool + if params['type'] == 'logical': + t = {'disks': [devs[1]]} + req = json.dumps(t) + resp = self.request(uri, req, 'PUT') + self.assertEquals(200, resp.status) + + # Delete the storage pool + resp = self.request(uri, '{}', 'DELETE') + self.assertEquals(204, resp.status) + + for pool in poolDefs: + _do_test(pool) diff --git a/tests/test_model.py b/tests/test_model.py index 0820386..35d0297 100644 --- a/tests/test_model.py +++ b/tests/test_model.py @@ -41,7 +41,6 @@ from kimchi import netinfo from kimchi.config import config, paths from kimchi.exception import InvalidOperation from kimchi.exception import InvalidParameter, NotFoundError, OperationFailed -from kimchi.iscsi import TargetClient from kimchi.model import model from kimchi.model.libvirtconnection import LibvirtConnection from kimchi.rollbackcontext import RollbackContext @@ -550,75 +549,6 @@ class ModelTests(unittest.TestCase): self.assertFalse(os.access(disk_path, os.F_OK)) @unittest.skipUnless(utils.running_as_root(), 'Must be run as root') - def test_storagepool(self): - inst = model.Model(None, self.tmp_store) - - poolDefs = [ - {'type': 'dir', - 'name': u'kīмсhīUnitTestDirPool', - 'path': '/tmp/kimchi-images'}, - {'type': 'iscsi', - 'name': u'kīмсhīUnitTestISCSIPool', - 'source': {'host': '127.0.0.1', - 'target': 'iqn.2013-12.localhost.kimchiUnitTest'}}] - - for poolDef in poolDefs: - with RollbackContext() as rollback: - path = poolDef.get('path') - name = poolDef['name'] - - if poolDef['type'] == 'iscsi': - if not TargetClient(**poolDef['source']).validate(): - continue - - pools = inst.storagepools_get_list() - num = len(pools) + 1 - - inst.storagepools_create(poolDef) - if poolDef['type'] == 'dir': - rollback.prependDefer(shutil.rmtree, poolDef['path']) - rollback.prependDefer(inst.storagepool_delete, name) - - pools = inst.storagepools_get_list() - self.assertEquals(num, len(pools)) - - poolinfo = inst.storagepool_lookup(name) - if path is not None: - self.assertEquals(path, poolinfo['path']) - self.assertEquals('inactive', poolinfo['state']) - if poolinfo['type'] == 'dir': - self.assertEquals(True, poolinfo['autostart']) - else: - self.assertEquals(False, poolinfo['autostart']) - - inst.storagepool_activate(name) - rollback.prependDefer(inst.storagepool_deactivate, name) - - poolinfo = inst.storagepool_lookup(name) - self.assertEquals('active', poolinfo['state']) - - autostart = poolinfo['autostart'] - ori_params = {'autostart': - True} if autostart else {'autostart': False} - for i in [True, False]: - params = {'autostart': i} - inst.storagepool_update(name, params) - rollback.prependDefer(inst.storagepool_update, name, - ori_params) - poolinfo = inst.storagepool_lookup(name) - self.assertEquals(i, poolinfo['autostart']) - inst.storagepool_update(name, ori_params) - - pools = inst.storagepools_get_list() - self.assertIn('default', pools) - poolinfo = inst.storagepool_lookup('default') - self.assertEquals('active', poolinfo['state']) - self.assertIn('ISO', pools) - poolinfo = inst.storagepool_lookup('ISO') - self.assertEquals('active', poolinfo['state']) - self.assertEquals((num - 1), len(pools)) - - @unittest.skipUnless(utils.running_as_root(), 'Must be run as root') def test_storagevolume(self): inst = model.Model(None, self.tmp_store) diff --git a/tests/test_rest.py b/tests/test_rest.py index 7416463..fe6020f 100644 --- a/tests/test_rest.py +++ b/tests/test_rest.py @@ -34,10 +34,9 @@ from functools import partial import iso_gen import kimchi.mockmodel import kimchi.server -from kimchi.config import paths from kimchi.rollbackcontext import RollbackContext from kimchi.utils import add_task -from utils import get_free_port, patch_auth, request +from utils import fake_auth_header, get_free_port, patch_auth, request from utils import run_server, wait_task @@ -1010,99 +1009,6 @@ class RestTests(unittest.TestCase): self.request('/storagepools/default-pool/storagevolumes').read()) self.assertEquals(1, len(resp)) - def test_get_storagepools(self): - storagepools = json.loads(self.request('/storagepools').read()) - self.assertEquals(2, len(storagepools)) - self.assertEquals('default-pool', storagepools[0]['name']) - self.assertEquals('active', storagepools[0]['state']) - self.assertEquals('kimchi_isos', storagepools[1]['name']) - self.assertEquals('kimchi-iso', storagepools[1]['type']) - - # Now add a couple of StoragePools to the mock model - for i in xrange(5): - name = 'kīмсhī-storagepool-%i' % i - req = json.dumps({'name': name, - 'capacity': 1024, - 'allocated': 512, - 'path': '/var/lib/libvirt/images/%i' % i, - 'type': 'dir'}) - resp = self.request('/storagepools', req, 'POST') - self.assertEquals(201, resp.status) - - req = json.dumps({'name': 'kīмсhī-storagepool-1', - 'capacity': 1024, - 'allocated': 512, - 'path': '/var/lib/libvirt/images/%i' % i, - 'type': 'dir'}) - resp = self.request('/storagepools', req, 'POST') - self.assertEquals(400, resp.status) - - # Reserved pool return 400 - req = json.dumps({'name': 'kimchi_isos', - 'capacity': 1024, - 'allocated': 512, - 'path': '/var/lib/libvirt/images/%i' % i, - 'type': 'dir'}) - resp = request(host, ssl_port, '/storagepools', req, 'POST') - self.assertEquals(400, resp.status) - - storagepools = json.loads(self.request('/storagepools').read()) - self.assertEquals(7, len(storagepools)) - - resp = self.request('/storagepools/kīмсhī-storagepool-1') - storagepool = json.loads(resp.read()) - self.assertEquals('kīмсhī-storagepool-1', - storagepool['name'].encode("utf-8")) - self.assertEquals('inactive', storagepool['state']) - self.assertIn('source', storagepool) - - def test_storagepool_action(self): - # Create a storage pool - req = json.dumps({'name': 'test-pool', - 'capacity': 1024, - 'allocated': 512, - 'path': '/var/lib/libvirt/images/', - 'type': 'dir'}) - resp = self.request('/storagepools', req, 'POST') - self.assertEquals(201, resp.status) - - # Verify the storage pool - storagepool = json.loads( - self.request('/storagepools/test-pool').read()) - self.assertEquals('inactive', storagepool['state']) - if storagepool['type'] == 'dir': - self.assertEquals(True, storagepool['autostart']) - else: - self.assertEquals(False, storagepool['autostart']) - - # Test if storage pool is persistent - self.assertEquals(True, storagepool['persistent']) - - # activate the storage pool - resp = self.request('/storagepools/test-pool/activate', '{}', 'POST') - storagepool = json.loads( - self.request('/storagepools/test-pool').read()) - self.assertEquals('active', storagepool['state']) - - # Deactivate the storage pool - resp = self.request('/storagepools/test-pool/deactivate', '{}', 'POST') - storagepool = json.loads( - self.request('/storagepools/test-pool').read()) - self.assertEquals('inactive', storagepool['state']) - - # Set autostart flag of the storage pool - for autostart in [True, False]: - t = {'autostart': autostart} - req = json.dumps(t) - resp = self.request('/storagepools/test-pool', req, 'PUT') - storagepool = json.loads( - self.request('/storagepools/test-pool').read()) - self.assertEquals(autostart, storagepool['autostart']) - - # Delete the storage pool - resp = self.request('/storagepools/test-pool', '{}', 'DELETE') - self.assertEquals(204, resp.status) - def test_get_storagevolumes(self): # Now add a StoragePool to the mock model self._create_pool('pool-1') @@ -1935,33 +1841,10 @@ class RestTests(unittest.TestCase): self.assertEquals(204, resp.status) def test_upload(self): - # If we use self.request, we may encode multipart formdata by ourselves - # requests lib take care of encode part, so use this lib instead - def fake_auth_header(): - headers = {'Accept': 'application/json'} - user, pw = kimchi.mockmodel.fake_user.items()[0] - hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw)) - headers['AUTHORIZATION'] = hdr - return headers - with RollbackContext() as rollback: - vol_path = os.path.join(paths.get_prefix(), 'COPYING') url = "https://%s:%s/storagepools/default-pool/storagevolumes" % \ (host, ssl_port) - with open(vol_path, 'rb') as fd: - r = requests.post(url, - files={'file': fd}, - verify=False, - headers=fake_auth_header()) - - self.assertEquals(r.status_code, 202) - task = r.json() - wait_task(self._task_lookup, task['id']) - uri = '/storagepools/default-pool/storagevolumes/%s' - resp = self.request(uri % task['target_uri'].split('/')[-1]) - self.assertEquals(200, resp.status) - # Create a file with 3M to upload vol_path = '/tmp/3m-file' with open(vol_path, 'wb') as fd: @@ -1978,6 +1861,7 @@ class RestTests(unittest.TestCase): self.assertEquals(r.status_code, 202) task = r.json() wait_task(self._task_lookup, task['id'], 15) + uri = '/storagepools/default-pool/storagevolumes/%s' resp = self.request(uri % task['target_uri'].split('/')[-1]) self.assertEquals(200, resp.status) diff --git a/tests/test_storagepool.py b/tests/test_storagepool.py new file mode 100644 index 0000000..1b87828 --- /dev/null +++ b/tests/test_storagepool.py @@ -0,0 +1,104 @@ +# -*- coding: utf-8 -*- +# +# Project Kimchi +# +# 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 +# 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.model.model import Model +from kimchi.rollbackcontext import RollbackContext +from utils import get_free_port, patch_auth, request +from utils import 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 = Model(None, '/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 StoragepoolTests(unittest.TestCase): + def setUp(self): + self.request = partial(request, host, ssl_port) + + def test_get_storagepools(self): + storagepools = json.loads(self.request('/storagepools').read()) + self.assertIn('default', [pool['name'] for pool in storagepools]) + + with RollbackContext() as rollback: + # Now add a couple of StoragePools to the mock model + for i in xrange(3): + name = u'kīмсhī-storagepool-%i' % i + req = json.dumps({'name': name, 'type': 'dir', + 'path': '/var/lib/libvirt/images/%i' % i}) + resp = self.request('/storagepools', req, 'POST') + rollback.prependDefer(model.storagepool_delete, name) + + self.assertEquals(201, resp.status) + + # Pool name must be unique + req = json.dumps({'name': name, 'type': 'dir', + 'path': '/var/lib/libvirt/images/%i' % i}) + resp = self.request('/storagepools', req, 'POST') + self.assertEquals(400, resp.status) + + # Verify pool information + resp = self.request('/storagepools/%s' % name.encode("utf-8")) + p = json.loads(resp.read()) + keys = [u'name', u'state', u'capacity', u'allocated', + u'available', u'path', u'source', u'type', + u'nr_volumes', u'autostart', u'persistent'] + self.assertEquals(sorted(keys), sorted(p.keys())) + self.assertEquals(name, p['name']) + self.assertEquals('inactive', p['state']) + self.assertEquals(True, p['persistent']) + self.assertEquals(True, p['autostart']) + self.assertEquals(0, p['nr_volumes']) + + pools = json.loads(self.request('/storagepools').read()) + self.assertEquals(len(storagepools) + 3, len(pools)) + + # Reserved pool return 400 + req = json.dumps({'name': 'kimchi_isos', 'type': 'dir', + 'path': '/var/lib/libvirt/images/%i' % i}) + resp = request(host, ssl_port, '/storagepools', req, 'POST') + self.assertEquals(400, resp.status) diff --git a/tests/utils.py b/tests/utils.py index 72078cc..7e70f2a 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -239,3 +239,14 @@ def rollback_wrapper(func, resource): except NotFoundError: # VM has been deleted already return + + +# This function is used to test storage volume upload. +# If we use self.request, we may encode multipart formdata by ourselves +# requests lib take care of encode part, so use this lib instead +def fake_auth_header(): + headers = {'Accept': 'application/json'} + user, pw = kimchi.mockmodel.fake_user.items()[0] + hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw)) + headers['AUTHORIZATION'] = hdr + return headers -- 2.1.0