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