[Kimchi-devel] [PATCH 8/8] Storage pool tests
Aline Manera
alinefm at linux.vnet.ibm.com
Thu Jan 22 13:32:58 UTC 2015
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 at linux.vnet.ibm.com>
---
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
More information about the Kimchi-devel
mailing list