[Kimchi-devel] [PATCH 8/8] Storage pool tests

Create 2 new files: tests/test_mock_storagepool.py and tests/test_model_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(a)linux.vnet.ibm.com&gt; --- tests/test_mock_storagepool.py | 141 ++++++++++++++++++++++++++++++++++++++++ tests/test_model.py | 70 -------------------- tests/test_model_storagepool.py | 104 +++++++++++++++++++++++++++++ tests/test_rest.py | 120 +--------------------------------- tests/utils.py | 11 ++++ 5 files changed, 258 insertions(+), 188 deletions(-) create mode 100644 tests/test_mock_storagepool.py create mode 100644 tests/test_model_storagepool.py diff --git a/tests/test_mock_storagepool.py b/tests/test_mock_storagepool.py new file mode 100644 index 0000000..1dc9277 --- /dev/null +++ b/tests/test_mock_storagepool.py @@ -0,0 +1,141 @@ +# -*- 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.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 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']) + + # Set autostart flag of an active 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 an active 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) + + # 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 an inactive 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 an inactive 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 c956007..24a11d0 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_model_storagepool.py b/tests/test_model_storagepool.py new file mode 100644 index 0000000..eabf875 --- /dev/null +++ b/tests/test_model_storagepool.py @@ -0,0 +1,104 @@ +# -*- 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.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/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/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