[Kimchi-devel] [PATCH] Move remaining Kimchi tests to src/wok structure.

From: Paulo Vital <pvital(a)linux.vnet.ibm.com&gt; Moving remaining Kimchi tests from old plugins structure to new src/wok structure. Signed-off-by: Paulo Vital <pvital(a)linux.vnet.ibm.com&gt; --- plugins/kimchi/tests/test_exception.py | 123 --------- plugins/kimchi/tests/test_objectstore.py | 97 ------- plugins/kimchi/tests/test_plugin.py | 126 --------- plugins/kimchi/tests/test_rollbackcontext.py | 99 ------- plugins/kimchi/tests/test_server.py | 289 --------------------- plugins/kimchi/tests/test_utils.py | 69 ----- src/wok/plugins/kimchi/tests/test_exception.py | 123 +++++++++ src/wok/plugins/kimchi/tests/test_objectstore.py | 97 +++++++ src/wok/plugins/kimchi/tests/test_plugin.py | 126 +++++++++ .../plugins/kimchi/tests/test_rollbackcontext.py | 99 +++++++ src/wok/plugins/kimchi/tests/test_server.py | 289 +++++++++++++++++++++ src/wok/plugins/kimchi/tests/test_utils.py | 69 +++++ 12 files changed, 803 insertions(+), 803 deletions(-) delete mode 100644 plugins/kimchi/tests/test_exception.py delete mode 100644 plugins/kimchi/tests/test_objectstore.py delete mode 100644 plugins/kimchi/tests/test_plugin.py delete mode 100644 plugins/kimchi/tests/test_rollbackcontext.py delete mode 100644 plugins/kimchi/tests/test_server.py delete mode 100644 plugins/kimchi/tests/test_utils.py create mode 100644 src/wok/plugins/kimchi/tests/test_exception.py create mode 100644 src/wok/plugins/kimchi/tests/test_objectstore.py create mode 100644 src/wok/plugins/kimchi/tests/test_plugin.py create mode 100644 src/wok/plugins/kimchi/tests/test_rollbackcontext.py create mode 100644 src/wok/plugins/kimchi/tests/test_server.py create mode 100644 src/wok/plugins/kimchi/tests/test_utils.py diff --git a/plugins/kimchi/tests/test_exception.py b/plugins/kimchi/tests/test_exception.py deleted file mode 100644 index 4459aa6..0000000 --- a/plugins/kimchi/tests/test_exception.py +++ /dev/null @@ -1,123 +0,0 @@ -# -# Kimchi -# -# Copyright IBM, Corp. 2013-2014 -# -# 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 wok.plugins.kimchi import mockmodel - -from utils import get_free_port, patch_auth, request, run_server - - -test_server = None -model = None -host = None -port = None -ssl_port = None - - -def setup_server(environment='development'): - global test_server, model, host, port, ssl_port - - patch_auth() - model = mockmodel.MockModel('/tmp/obj-store-test') - host = '127.0.0.1' - port = get_free_port('http') - ssl_port = get_free_port('https') - test_server = run_server(host, port, ssl_port, test_mode=True, model=model, - environment=environment) - - -class ExceptionTests(unittest.TestCase): - def tearDown(self): - test_server.stop() - os.unlink('/tmp/obj-store-test') - - def test_production_env(self): - """ - Test reasons sanitized in production env - """ - setup_server('production') - # test 404 - resp = json.loads( - request(host, ssl_port, '/plugins/kimchi/vms/blah').read() - ) - self.assertEquals('404 Not Found', resp.get('code')) - - # test 405 wrong method - resp = json.loads(request(host, ssl_port, '/', None, 'DELETE').read()) - msg = u'WOKAPI0002E: Delete is not allowed for wokroot' - self.assertEquals('405 Method Not Allowed', resp.get('code')) - self.assertEquals(msg, resp.get('reason')) - - # test 400 parse error - resp = json.loads( - request(host, ssl_port, '/plugins/kimchi/vms', '{', 'POST').read() - ) - msg = u'WOKAPI0006E: Unable to parse JSON request' - self.assertEquals('400 Bad Request', resp.get('code')) - self.assertEquals(msg, resp.get('reason')) - self.assertNotIn('call_stack', resp) - - # test 400 missing required parameter - req = json.dumps({}) - resp = json.loads( - request(host, ssl_port, '/plugins/kimchi/vms', req, 'POST').read() - ) - self.assertEquals('400 Bad Request', resp.get('code')) - m = u"KCHVM0016E: Specify a template to create a virtual machine from" - self.assertEquals(m, resp.get('reason')) - self.assertNotIn('call_stack', resp) - - def test_development_env(self): - """ - Test traceback thrown in development env - """ - setup_server() - # test 404 - resp = json.loads( - request(host, ssl_port, '/plugins/kimchi/vms/blah').read() - ) - self.assertEquals('404 Not Found', resp.get('code')) - - # test 405 wrong method - resp = json.loads(request(host, ssl_port, '/', None, 'DELETE').read()) - msg = u'WOKAPI0002E: Delete is not allowed for wokroot' - self.assertEquals('405 Method Not Allowed', resp.get('code')) - self.assertEquals(msg, resp.get('reason')) - - # test 400 parse error - resp = json.loads( - request(host, ssl_port, '/plugins/kimchi/vms', '{', 'POST').read() - ) - msg = u'WOKAPI0006E: Unable to parse JSON request' - self.assertEquals('400 Bad Request', resp.get('code')) - self.assertEquals(msg, resp.get('reason')) - self.assertIn('call_stack', resp) - - # test 400 missing required parameter - req = json.dumps({}) - resp = json.loads( - request(host, ssl_port, '/plugins/kimchi/vms', req, 'POST').read() - ) - m = u"KCHVM0016E: Specify a template to create a virtual machine from" - self.assertEquals('400 Bad Request', resp.get('code')) - self.assertEquals(m, resp.get('reason')) - self.assertIn('call_stack', resp) diff --git a/plugins/kimchi/tests/test_objectstore.py b/plugins/kimchi/tests/test_objectstore.py deleted file mode 100644 index 632786f..0000000 --- a/plugins/kimchi/tests/test_objectstore.py +++ /dev/null @@ -1,97 +0,0 @@ -# -*- 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 os -import tempfile -import threading -import unittest - -from wok import objectstore -from wok.exception import NotFoundError - - -tmpfile = None - - -def setUpModule(): - global tmpfile - tmpfile = tempfile.mktemp() - - -def tearDownModule(): - os.unlink(tmpfile) - - -class ObjectStoreTests(unittest.TestCase): - def test_objectstore(self): - store = objectstore.ObjectStore(tmpfile) - - with store as session: - # Test create - session.store('fǒǒ', 'těst1', {'α': 1}) - session.store('fǒǒ', 'těst2', {'β': 2}) - - # Test list - items = session.get_list('fǒǒ') - self.assertTrue(u'těst1' in items) - self.assertTrue(u'těst2' in items) - - # Test get - item = session.get('fǒǒ', 'těst1') - self.assertEquals(1, item[u'α']) - - # Test delete - session.delete('fǒǒ', 'těst2') - self.assertEquals(1, len(session.get_list('fǒǒ'))) - - # Test get non-existent item - - self.assertRaises(NotFoundError, session.get, - 'α', 'β') - - # Test delete non-existent item - self.assertRaises(NotFoundError, session.delete, - 'fǒǒ', 'těst2') - - # Test refresh existing item - session.store('fǒǒ', 'těst1', {'α': 2}) - item = session.get('fǒǒ', 'těst1') - self.assertEquals(2, item[u'α']) - - def test_object_store_threaded(self): - def worker(ident): - with store as session: - session.store('foo', ident, {}) - - store = objectstore.ObjectStore(tmpfile) - - threads = [] - for i in xrange(50): - t = threading.Thread(target=worker, args=(i,)) - t.setDaemon(True) - t.start() - threads.append(t) - - for t in threads: - t.join() - - with store as session: - self.assertEquals(50, len(session.get_list('foo'))) - self.assertEquals(10, len(store._connections.keys())) diff --git a/plugins/kimchi/tests/test_plugin.py b/plugins/kimchi/tests/test_plugin.py deleted file mode 100644 index fc8e277..0000000 --- a/plugins/kimchi/tests/test_plugin.py +++ /dev/null @@ -1,126 +0,0 @@ -# -# Project Kimchi -# -# Copyright IBM, Corp. 2013-2014 -# -# 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 wok.utils import get_enabled_plugins - -from wok.plugins.kimchi import mockmodel - -import utils - - -test_server = None -model = None -host = None -port = None -ssl_port = None - - -def setUpModule(): - global test_server, model, host, port, ssl_port - - utils.patch_auth() - model = mockmodel.MockModel('/tmp/obj-store-test') - host = '127.0.0.1' - port = utils.get_free_port('http') - ssl_port = utils.get_free_port('https') - test_server = utils.run_server(host, port, ssl_port, test_mode=True, - model=model) - - -def tearDownModule(): - test_server.stop() - os.unlink('/tmp/obj-store-test') - - -(a)unittest.skipUnless( - 'sample' in [plugin for plugin, _config in get_enabled_plugins()], - 'sample plugin is not enabled, skip this test!') -class PluginTests(unittest.TestCase): - - def setUp(self): - self.request = partial(utils.request, host, ssl_port) - - def _create_rectangle(self, name, length, width): - req = json.dumps({'name': name, 'length': length, 'width': width}) - resp = self.request('/plugins/sample/rectangles', req, 'POST') - return resp - - def _get_rectangle(self, name): - resp = self.request('/plugins/sample/rectangles/%s' % name) - return json.loads(resp.read()) - - def _create_rectangle_and_assert(self, name, length, width): - resp = self._create_rectangle(name, length, width) - self.assertEquals(201, resp.status) - - rectangle = self._get_rectangle(name) - self.assertEquals(rectangle['name'], name) - self.assertEquals(rectangle['length'], length) - self.assertEquals(rectangle['width'], width) - - def _get_rectangles_list(self): - resp = self.request('/plugins/sample/rectangles') - rectangles = json.loads(resp.read()) - name_list = [rectangle['name'] for rectangle in rectangles] - return name_list - - def test_rectangles(self): - # Create two new rectangles - self._create_rectangle_and_assert('small', 10, 8) - self._create_rectangle_and_assert('big', 20, 16) - - # Verify they're in the list - name_list = self._get_rectangles_list() - self.assertIn('small', name_list) - self.assertIn('big', name_list) - - # Update the big rectangle. - req = json.dumps({'length': 40, 'width': 30}) - resp = self.request('/plugins/sample/rectangles/big', req, 'PUT') - self.assertEquals(200, resp.status) - big = self._get_rectangle('big') - self.assertEquals(big['length'], 40) - self.assertEquals(big['width'], 30) - - # Delete two rectangles - resp = self.request('/plugins/sample/rectangles/big', '{}', 'DELETE') - self.assertEquals(204, resp.status) - resp = self.request('/plugins/sample/rectangles/small', '{}', 'DELETE') - self.assertEquals(204, resp.status) - name_list = self._get_rectangles_list() - self.assertEquals([], name_list) - - def test_bad_params(self): - # Bad name - resp = self._create_rectangle(1.0, 30, 40) - self.assertEquals(400, resp.status) - - # Bad length value - resp = self._create_rectangle('test', -10.0, 40) - self.assertEquals(400, resp.status) - - # Missing param for width - req = json.dumps({'name': 'nowidth', 'length': 40}) - resp = self.request('/plugins/sample/rectangles', req, 'POST') - self.assertEquals(400, resp.status) diff --git a/plugins/kimchi/tests/test_rollbackcontext.py b/plugins/kimchi/tests/test_rollbackcontext.py deleted file mode 100644 index 6eac6d0..0000000 --- a/plugins/kimchi/tests/test_rollbackcontext.py +++ /dev/null @@ -1,99 +0,0 @@ -# -# Project Kimchi -# -# Copyright IBM, Corp. 2014 -# -# 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 unittest - -from wok.rollbackcontext import RollbackContext - - -class FirstError(Exception): - '''A hypothetical exception to be raise in the test firstly.''' - pass - - -class SecondError(Exception): - '''A hypothetical exception to be raise in the test secondly.''' - pass - - -class RollbackContextTests(unittest.TestCase): - - def setUp(self): - self._counter = 0 - - def _inc_counter(self): - self._counter += 1 - - def _raise(self, exception=FirstError): - raise exception() - - def test_rollback(self): - with RollbackContext() as rollback: - rollback.prependDefer(self._inc_counter) - rollback.prependDefer(self._inc_counter) - self.assertEquals(self._counter, 2) - - def test_raise(self): - try: - with RollbackContext() as rollback: - rollback.prependDefer(self._inc_counter) - rollback.prependDefer(self._inc_counter) - raise FirstError() - rollback.prependDefer(self._inc_counter) - except FirstError: - # All undo before the FirstError should be run - self.assertEquals(self._counter, 2) - else: - self.fail('Should have raised FirstError') - - def test_raise_undo(self): - try: - with RollbackContext() as rollback: - rollback.prependDefer(self._inc_counter) - rollback.prependDefer(self._raise) - rollback.prependDefer(self._inc_counter) - except FirstError: - # All undo should be run - self.assertEquals(self._counter, 2) - else: - self.fail('Should have raised FirstError') - - def test_raise_prefer_original(self): - try: - with RollbackContext() as rollback: - rollback.prependDefer(self._raise, SecondError) - raise FirstError() - except FirstError: - pass - except SecondError: - self.fail('Should have preferred FirstError to SecondError') - else: - self.fail('Should have raised FirstError') - - def test_raise_prefer_first_undo(self): - try: - with RollbackContext() as rollback: - rollback.prependDefer(self._raise, SecondError) - rollback.prependDefer(self._raise, FirstError) - except FirstError: - pass - except SecondError: - self.fail('Should have preferred FirstError to SecondError') - else: - self.fail('Should have raised FirstError') diff --git a/plugins/kimchi/tests/test_server.py b/plugins/kimchi/tests/test_server.py deleted file mode 100644 index d5ef565..0000000 --- a/plugins/kimchi/tests/test_server.py +++ /dev/null @@ -1,289 +0,0 @@ -# -# Project Kimchi -# -# 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 -# 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 base64 -import cherrypy -import json -import os -import tempfile -import threading -import unittest -from functools import partial - -from wok.control.base import Collection, Resource - -from wok.plugins.kimchi import mockmodel - -import utils - - -test_server = None -model = None -host = None -port = None -ssl_port = None -cherrypy_port = None -tmpfile = None - - -def setUpModule(): - global test_server, model, host, port, ssl_port, cherrypy_port, tmpfile - - utils.patch_auth() - tmpfile = tempfile.mktemp() - model = mockmodel.MockModel(tmpfile) - host = '127.0.0.1' - port = utils.get_free_port('http') - ssl_port = utils.get_free_port('https') - cherrypy_port = utils.get_free_port('cherrypy_port') - test_server = utils.run_server(host, port, ssl_port, test_mode=True, - cherrypy_port=cherrypy_port, model=model) - - -def tearDownModule(): - test_server.stop() - os.unlink(tmpfile) - - -class ServerTests(unittest.TestCase): - def setUp(self): - self.request = partial(utils.request, host, ssl_port) - model.reset() - - def assertValidJSON(self, txt): - try: - json.loads(txt) - except ValueError: - self.fail("Invalid JSON: %s" % txt) - - def test_server_start(self): - """ - Test that we can start a server and receive HTTP:200. - """ - resp = self.request('/') - self.assertEquals(200, resp.status) - - def test_multithreaded_connection(self): - def worker(): - for i in xrange(100): - ret = model.vms_get_list() - self.assertEquals('test', ret[0]) - - threads = [] - for i in xrange(100): - t = threading.Thread(target=worker) - t.setDaemon(True) - t.start() - threads.append(t) - for t in threads: - t.join() - - def test_collection(self): - c = Collection(model) - - # The base Collection is always empty - cherrypy.request.method = 'GET' - cherrypy.request.headers['Accept'] = 'application/json' - self.assertEquals('[]', c.index()) - - # POST and DELETE raise HTTP:405 by default - for method in ('POST', 'DELETE'): - cherrypy.request.method = method - try: - c.index() - except cherrypy.HTTPError, e: - self.assertEquals(405, e.code) - else: - self.fail("Expected exception not raised") - - def test_resource(self): - r = Resource(model) - - # Test the base Resource representation - cherrypy.request.method = 'GET' - cherrypy.request.headers['Accept'] = 'application/json' - self.assertEquals('{}', r.index()) - - # POST and DELETE raise HTTP:405 by default - for method in ('POST', 'DELETE'): - cherrypy.request.method = method - try: - r.index() - except cherrypy.HTTPError, e: - self.assertEquals(405, e.code) - else: - self.fail("Expected exception not raised") - - def test_404(self): - """ - A non-existent path should return HTTP:404 - """ - url_list = ['/plugins/kimchi/doesnotexist', '/plugins/kimchi/vms/blah'] - for url in url_list: - resp = self.request(url) - self.assertEquals(404, resp.status) - - # Verify it works for DELETE too - resp = self.request('/plugins/kimchi/templates/blah', '', 'DELETE') - self.assertEquals(404, resp.status) - - def test_accepts(self): - """ - Verify the following expectations regarding the client Accept header: - If omitted, default to html - If 'application/json', serve the rest api - If 'text/html', serve the UI - If both of the above (in any order), serve the rest api - If neither of the above, HTTP:406 - """ - resp = self.request("/", headers={}) - location = resp.getheader('location') - self.assertTrue(location.endswith("login.html")) - resp = self.request("/login.html", headers={}) - self.assertTrue('<!doctype html>' in resp.read().lower()) - - resp = self.request("/", headers={'Accept': 'application/json'}) - self.assertValidJSON(resp.read()) - - resp = self.request("/", headers={'Accept': 'text/html'}) - location = resp.getheader('location') - self.assertTrue(location.endswith("login.html")) - - resp = self.request("/", headers={'Accept': - 'application/json, text/html'}) - self.assertValidJSON(resp.read()) - - resp = self.request("/", headers={'Accept': - 'text/html, application/json'}) - self.assertValidJSON(resp.read()) - - h = {'Accept': 'text/plain'} - resp = self.request('/', None, 'GET', h) - self.assertEquals(406, resp.status) - - def test_auth_unprotected(self): - hdrs = {'AUTHORIZATION': ''} - uris = ['/plugins/kimchi/js/kimchi.min.js', - '/plugins/kimchi/css/theme-default.min.css', - '/plugins/kimchi/images/icon-vm.png', - '/libs/jquery-1.10.0.min.js', - '/login.html', - '/logout'] - - for uri in uris: - resp = self.request(uri, None, 'HEAD', hdrs) - self.assertEquals(200, resp.status) - - def test_auth_protected(self): - hdrs = {'AUTHORIZATION': ''} - uris = ['/plugins/kimchi/vms', - '/plugins/kimchi/vms/doesnotexist', - '/tasks'] - - for uri in uris: - resp = self.request(uri, None, 'GET', hdrs) - self.assertEquals(401, resp.status) - - def test_auth_bad_creds(self): - # Test HTTPBA - hdrs = {'AUTHORIZATION': "Basic " + base64.b64encode("nouser:badpass")} - resp = self.request('/plugins/kimchi/vms', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - - # Test REST API - hdrs = {'AUTHORIZATION': ''} - req = json.dumps({'username': 'nouser', 'password': 'badpass'}) - resp = self.request('/login', req, 'POST', hdrs) - self.assertEquals(401, resp.status) - - def test_auth_browser_no_httpba(self): - # Kimchi detects REST requests from the browser by looking for a - # specific header - hdrs = {"X-Requested-With": "XMLHttpRequest"} - - # Try our request (Note that request() will add a valid HTTPBA header) - resp = self.request('/plugins/kimchi/vms', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - self.assertEquals(None, resp.getheader('WWW-Authenticate')) - - def test_auth_session(self): - hdrs = {'AUTHORIZATION': '', - 'Content-Type': 'application/json', - 'Accept': 'application/json'} - - # Test we are logged out - resp = self.request('/tasks', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - - # Execute a login call - user, pw = mockmodel.fake_user.items()[0] - req = json.dumps({'username': user, 'password': pw}) - resp = self.request('/login', req, 'POST', hdrs) - self.assertEquals(200, resp.status) - - user_info = json.loads(resp.read()) - self.assertEquals(sorted(user_info.keys()), - ['groups', 'roles', 'username']) - roles = user_info['roles'] - for tab, role in roles.iteritems(): - self.assertEquals(role, u'admin') - - cookie = resp.getheader('set-cookie') - hdrs['Cookie'] = cookie - - # Test we are logged in with the cookie - resp = self.request('/tasks', None, 'GET', hdrs) - self.assertEquals(200, resp.status) - - # Execute a logout call - resp = self.request('/logout', '{}', 'POST', hdrs) - self.assertEquals(200, resp.status) - del hdrs['Cookie'] - - # Test we are logged out - resp = self.request('/tasks', None, 'GET', hdrs) - self.assertEquals(401, resp.status) - - def test_get_param(self): - # Create a mock ISO file - mockiso = '/tmp/mock.iso' - open('/tmp/mock.iso', 'w').close() - - # Create 2 different templates - req = json.dumps({'name': 'test-tmpl1', 'cdrom': mockiso}) - self.request('/plugins/kimchi/templates', req, 'POST') - - req = json.dumps({'name': 'test-tmpl2', 'cdrom': mockiso}) - self.request('/plugins/kimchi/templates', req, 'POST') - - # Remove mock iso - os.unlink(mockiso) - - # Get the templates - resp = self.request('/plugins/kimchi/templates') - self.assertEquals(200, resp.status) - res = json.loads(resp.read()) - self.assertEquals(2, len(res)) - - # Get a specific template - resp = self.request('/plugins/kimchi/templates?name=test-tmpl1') - self.assertEquals(200, resp.status) - res = json.loads(resp.read()) - self.assertEquals(1, len(res)) - self.assertEquals('test-tmpl1', res[0]['name']) diff --git a/plugins/kimchi/tests/test_utils.py b/plugins/kimchi/tests/test_utils.py deleted file mode 100644 index bcb14e2..0000000 --- a/plugins/kimchi/tests/test_utils.py +++ /dev/null @@ -1,69 +0,0 @@ -# -# 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 unittest - -from wok.exception import InvalidParameter -from wok.utils import convert_data_size - - -class UtilsTests(unittest.TestCase): - def test_convert_data_size(self): - failure_data = [{'val': None, 'from': 'MiB'}, - {'val': self, 'from': 'MiB'}, - {'val': 1, 'from': None}, - {'val': 1, 'from': ''}, - {'val': 1, 'from': 'foo'}, - {'val': 1, 'from': 'kib'}, - {'val': 1, 'from': 'MiB', 'to': None}, - {'val': 1, 'from': 'MiB', 'to': ''}, - {'val': 1, 'from': 'MiB', 'to': 'foo'}, - {'val': 1, 'from': 'MiB', 'to': 'kib'}] - - for d in failure_data: - if 'to' in d: - self.assertRaises(InvalidParameter, convert_data_size, - d['val'], d['from'], d['to']) - else: - self.assertRaises(InvalidParameter, convert_data_size, - d['val'], d['from']) - - success_data = [{'got': convert_data_size(5, 'MiB', 'MiB'), - 'want': 5}, - {'got': convert_data_size(5, 'MiB', 'KiB'), - 'want': 5120}, - {'got': convert_data_size(5, 'MiB', 'M'), - 'want': 5.24288}, - {'got': convert_data_size(5, 'MiB', 'GiB'), - 'want': 0.0048828125}, - {'got': convert_data_size(5, 'MiB', 'Tb'), - 'want': 4.194304e-05}, - {'got': convert_data_size(5, 'KiB', 'MiB'), - 'want': 0.0048828125}, - {'got': convert_data_size(5, 'M', 'MiB'), - 'want': 4.76837158203125}, - {'got': convert_data_size(5, 'GiB', 'MiB'), - 'want': 5120}, - {'got': convert_data_size(5, 'Tb', 'MiB'), - 'want': 596046.4477539062}, - {'got': convert_data_size(5, 'MiB'), - 'want': convert_data_size(5, 'MiB', 'B')}] - - for d in success_data: - self.assertEquals(d['got'], d['want']) diff --git a/src/wok/plugins/kimchi/tests/test_exception.py b/src/wok/plugins/kimchi/tests/test_exception.py new file mode 100644 index 0000000..4459aa6 --- /dev/null +++ b/src/wok/plugins/kimchi/tests/test_exception.py @@ -0,0 +1,123 @@ +# +# Kimchi +# +# Copyright IBM, Corp. 2013-2014 +# +# 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 wok.plugins.kimchi import mockmodel + +from utils import get_free_port, patch_auth, request, run_server + + +test_server = None +model = None +host = None +port = None +ssl_port = None + + +def setup_server(environment='development'): + global test_server, model, host, port, ssl_port + + patch_auth() + model = mockmodel.MockModel('/tmp/obj-store-test') + host = '127.0.0.1' + port = get_free_port('http') + ssl_port = get_free_port('https') + test_server = run_server(host, port, ssl_port, test_mode=True, model=model, + environment=environment) + + +class ExceptionTests(unittest.TestCase): + def tearDown(self): + test_server.stop() + os.unlink('/tmp/obj-store-test') + + def test_production_env(self): + """ + Test reasons sanitized in production env + """ + setup_server('production') + # test 404 + resp = json.loads( + request(host, ssl_port, '/plugins/kimchi/vms/blah').read() + ) + self.assertEquals('404 Not Found', resp.get('code')) + + # test 405 wrong method + resp = json.loads(request(host, ssl_port, '/', None, 'DELETE').read()) + msg = u'WOKAPI0002E: Delete is not allowed for wokroot' + self.assertEquals('405 Method Not Allowed', resp.get('code')) + self.assertEquals(msg, resp.get('reason')) + + # test 400 parse error + resp = json.loads( + request(host, ssl_port, '/plugins/kimchi/vms', '{', 'POST').read() + ) + msg = u'WOKAPI0006E: Unable to parse JSON request' + self.assertEquals('400 Bad Request', resp.get('code')) + self.assertEquals(msg, resp.get('reason')) + self.assertNotIn('call_stack', resp) + + # test 400 missing required parameter + req = json.dumps({}) + resp = json.loads( + request(host, ssl_port, '/plugins/kimchi/vms', req, 'POST').read() + ) + self.assertEquals('400 Bad Request', resp.get('code')) + m = u"KCHVM0016E: Specify a template to create a virtual machine from" + self.assertEquals(m, resp.get('reason')) + self.assertNotIn('call_stack', resp) + + def test_development_env(self): + """ + Test traceback thrown in development env + """ + setup_server() + # test 404 + resp = json.loads( + request(host, ssl_port, '/plugins/kimchi/vms/blah').read() + ) + self.assertEquals('404 Not Found', resp.get('code')) + + # test 405 wrong method + resp = json.loads(request(host, ssl_port, '/', None, 'DELETE').read()) + msg = u'WOKAPI0002E: Delete is not allowed for wokroot' + self.assertEquals('405 Method Not Allowed', resp.get('code')) + self.assertEquals(msg, resp.get('reason')) + + # test 400 parse error + resp = json.loads( + request(host, ssl_port, '/plugins/kimchi/vms', '{', 'POST').read() + ) + msg = u'WOKAPI0006E: Unable to parse JSON request' + self.assertEquals('400 Bad Request', resp.get('code')) + self.assertEquals(msg, resp.get('reason')) + self.assertIn('call_stack', resp) + + # test 400 missing required parameter + req = json.dumps({}) + resp = json.loads( + request(host, ssl_port, '/plugins/kimchi/vms', req, 'POST').read() + ) + m = u"KCHVM0016E: Specify a template to create a virtual machine from" + self.assertEquals('400 Bad Request', resp.get('code')) + self.assertEquals(m, resp.get('reason')) + self.assertIn('call_stack', resp) diff --git a/src/wok/plugins/kimchi/tests/test_objectstore.py b/src/wok/plugins/kimchi/tests/test_objectstore.py new file mode 100644 index 0000000..632786f --- /dev/null +++ b/src/wok/plugins/kimchi/tests/test_objectstore.py @@ -0,0 +1,97 @@ +# -*- 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 os +import tempfile +import threading +import unittest + +from wok import objectstore +from wok.exception import NotFoundError + + +tmpfile = None + + +def setUpModule(): + global tmpfile + tmpfile = tempfile.mktemp() + + +def tearDownModule(): + os.unlink(tmpfile) + + +class ObjectStoreTests(unittest.TestCase): + def test_objectstore(self): + store = objectstore.ObjectStore(tmpfile) + + with store as session: + # Test create + session.store('fǒǒ', 'těst1', {'α': 1}) + session.store('fǒǒ', 'těst2', {'β': 2}) + + # Test list + items = session.get_list('fǒǒ') + self.assertTrue(u'těst1' in items) + self.assertTrue(u'těst2' in items) + + # Test get + item = session.get('fǒǒ', 'těst1') + self.assertEquals(1, item[u'α']) + + # Test delete + session.delete('fǒǒ', 'těst2') + self.assertEquals(1, len(session.get_list('fǒǒ'))) + + # Test get non-existent item + + self.assertRaises(NotFoundError, session.get, + 'α', 'β') + + # Test delete non-existent item + self.assertRaises(NotFoundError, session.delete, + 'fǒǒ', 'těst2') + + # Test refresh existing item + session.store('fǒǒ', 'těst1', {'α': 2}) + item = session.get('fǒǒ', 'těst1') + self.assertEquals(2, item[u'α']) + + def test_object_store_threaded(self): + def worker(ident): + with store as session: + session.store('foo', ident, {}) + + store = objectstore.ObjectStore(tmpfile) + + threads = [] + for i in xrange(50): + t = threading.Thread(target=worker, args=(i,)) + t.setDaemon(True) + t.start() + threads.append(t) + + for t in threads: + t.join() + + with store as session: + self.assertEquals(50, len(session.get_list('foo'))) + self.assertEquals(10, len(store._connections.keys())) diff --git a/src/wok/plugins/kimchi/tests/test_plugin.py b/src/wok/plugins/kimchi/tests/test_plugin.py new file mode 100644 index 0000000..fc8e277 --- /dev/null +++ b/src/wok/plugins/kimchi/tests/test_plugin.py @@ -0,0 +1,126 @@ +# +# Project Kimchi +# +# Copyright IBM, Corp. 2013-2014 +# +# 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 wok.utils import get_enabled_plugins + +from wok.plugins.kimchi import mockmodel + +import utils + + +test_server = None +model = None +host = None +port = None +ssl_port = None + + +def setUpModule(): + global test_server, model, host, port, ssl_port + + utils.patch_auth() + model = mockmodel.MockModel('/tmp/obj-store-test') + host = '127.0.0.1' + port = utils.get_free_port('http') + ssl_port = utils.get_free_port('https') + test_server = utils.run_server(host, port, ssl_port, test_mode=True, + model=model) + + +def tearDownModule(): + test_server.stop() + os.unlink('/tmp/obj-store-test') + + +(a)unittest.skipUnless( + 'sample' in [plugin for plugin, _config in get_enabled_plugins()], + 'sample plugin is not enabled, skip this test!') +class PluginTests(unittest.TestCase): + + def setUp(self): + self.request = partial(utils.request, host, ssl_port) + + def _create_rectangle(self, name, length, width): + req = json.dumps({'name': name, 'length': length, 'width': width}) + resp = self.request('/plugins/sample/rectangles', req, 'POST') + return resp + + def _get_rectangle(self, name): + resp = self.request('/plugins/sample/rectangles/%s' % name) + return json.loads(resp.read()) + + def _create_rectangle_and_assert(self, name, length, width): + resp = self._create_rectangle(name, length, width) + self.assertEquals(201, resp.status) + + rectangle = self._get_rectangle(name) + self.assertEquals(rectangle['name'], name) + self.assertEquals(rectangle['length'], length) + self.assertEquals(rectangle['width'], width) + + def _get_rectangles_list(self): + resp = self.request('/plugins/sample/rectangles') + rectangles = json.loads(resp.read()) + name_list = [rectangle['name'] for rectangle in rectangles] + return name_list + + def test_rectangles(self): + # Create two new rectangles + self._create_rectangle_and_assert('small', 10, 8) + self._create_rectangle_and_assert('big', 20, 16) + + # Verify they're in the list + name_list = self._get_rectangles_list() + self.assertIn('small', name_list) + self.assertIn('big', name_list) + + # Update the big rectangle. + req = json.dumps({'length': 40, 'width': 30}) + resp = self.request('/plugins/sample/rectangles/big', req, 'PUT') + self.assertEquals(200, resp.status) + big = self._get_rectangle('big') + self.assertEquals(big['length'], 40) + self.assertEquals(big['width'], 30) + + # Delete two rectangles + resp = self.request('/plugins/sample/rectangles/big', '{}', 'DELETE') + self.assertEquals(204, resp.status) + resp = self.request('/plugins/sample/rectangles/small', '{}', 'DELETE') + self.assertEquals(204, resp.status) + name_list = self._get_rectangles_list() + self.assertEquals([], name_list) + + def test_bad_params(self): + # Bad name + resp = self._create_rectangle(1.0, 30, 40) + self.assertEquals(400, resp.status) + + # Bad length value + resp = self._create_rectangle('test', -10.0, 40) + self.assertEquals(400, resp.status) + + # Missing param for width + req = json.dumps({'name': 'nowidth', 'length': 40}) + resp = self.request('/plugins/sample/rectangles', req, 'POST') + self.assertEquals(400, resp.status) diff --git a/src/wok/plugins/kimchi/tests/test_rollbackcontext.py b/src/wok/plugins/kimchi/tests/test_rollbackcontext.py new file mode 100644 index 0000000..6eac6d0 --- /dev/null +++ b/src/wok/plugins/kimchi/tests/test_rollbackcontext.py @@ -0,0 +1,99 @@ +# +# Project Kimchi +# +# Copyright IBM, Corp. 2014 +# +# 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 unittest + +from wok.rollbackcontext import RollbackContext + + +class FirstError(Exception): + '''A hypothetical exception to be raise in the test firstly.''' + pass + + +class SecondError(Exception): + '''A hypothetical exception to be raise in the test secondly.''' + pass + + +class RollbackContextTests(unittest.TestCase): + + def setUp(self): + self._counter = 0 + + def _inc_counter(self): + self._counter += 1 + + def _raise(self, exception=FirstError): + raise exception() + + def test_rollback(self): + with RollbackContext() as rollback: + rollback.prependDefer(self._inc_counter) + rollback.prependDefer(self._inc_counter) + self.assertEquals(self._counter, 2) + + def test_raise(self): + try: + with RollbackContext() as rollback: + rollback.prependDefer(self._inc_counter) + rollback.prependDefer(self._inc_counter) + raise FirstError() + rollback.prependDefer(self._inc_counter) + except FirstError: + # All undo before the FirstError should be run + self.assertEquals(self._counter, 2) + else: + self.fail('Should have raised FirstError') + + def test_raise_undo(self): + try: + with RollbackContext() as rollback: + rollback.prependDefer(self._inc_counter) + rollback.prependDefer(self._raise) + rollback.prependDefer(self._inc_counter) + except FirstError: + # All undo should be run + self.assertEquals(self._counter, 2) + else: + self.fail('Should have raised FirstError') + + def test_raise_prefer_original(self): + try: + with RollbackContext() as rollback: + rollback.prependDefer(self._raise, SecondError) + raise FirstError() + except FirstError: + pass + except SecondError: + self.fail('Should have preferred FirstError to SecondError') + else: + self.fail('Should have raised FirstError') + + def test_raise_prefer_first_undo(self): + try: + with RollbackContext() as rollback: + rollback.prependDefer(self._raise, SecondError) + rollback.prependDefer(self._raise, FirstError) + except FirstError: + pass + except SecondError: + self.fail('Should have preferred FirstError to SecondError') + else: + self.fail('Should have raised FirstError') diff --git a/src/wok/plugins/kimchi/tests/test_server.py b/src/wok/plugins/kimchi/tests/test_server.py new file mode 100644 index 0000000..d5ef565 --- /dev/null +++ b/src/wok/plugins/kimchi/tests/test_server.py @@ -0,0 +1,289 @@ +# +# Project Kimchi +# +# 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 +# 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 base64 +import cherrypy +import json +import os +import tempfile +import threading +import unittest +from functools import partial + +from wok.control.base import Collection, Resource + +from wok.plugins.kimchi import mockmodel + +import utils + + +test_server = None +model = None +host = None +port = None +ssl_port = None +cherrypy_port = None +tmpfile = None + + +def setUpModule(): + global test_server, model, host, port, ssl_port, cherrypy_port, tmpfile + + utils.patch_auth() + tmpfile = tempfile.mktemp() + model = mockmodel.MockModel(tmpfile) + host = '127.0.0.1' + port = utils.get_free_port('http') + ssl_port = utils.get_free_port('https') + cherrypy_port = utils.get_free_port('cherrypy_port') + test_server = utils.run_server(host, port, ssl_port, test_mode=True, + cherrypy_port=cherrypy_port, model=model) + + +def tearDownModule(): + test_server.stop() + os.unlink(tmpfile) + + +class ServerTests(unittest.TestCase): + def setUp(self): + self.request = partial(utils.request, host, ssl_port) + model.reset() + + def assertValidJSON(self, txt): + try: + json.loads(txt) + except ValueError: + self.fail("Invalid JSON: %s" % txt) + + def test_server_start(self): + """ + Test that we can start a server and receive HTTP:200. + """ + resp = self.request('/') + self.assertEquals(200, resp.status) + + def test_multithreaded_connection(self): + def worker(): + for i in xrange(100): + ret = model.vms_get_list() + self.assertEquals('test', ret[0]) + + threads = [] + for i in xrange(100): + t = threading.Thread(target=worker) + t.setDaemon(True) + t.start() + threads.append(t) + for t in threads: + t.join() + + def test_collection(self): + c = Collection(model) + + # The base Collection is always empty + cherrypy.request.method = 'GET' + cherrypy.request.headers['Accept'] = 'application/json' + self.assertEquals('[]', c.index()) + + # POST and DELETE raise HTTP:405 by default + for method in ('POST', 'DELETE'): + cherrypy.request.method = method + try: + c.index() + except cherrypy.HTTPError, e: + self.assertEquals(405, e.code) + else: + self.fail("Expected exception not raised") + + def test_resource(self): + r = Resource(model) + + # Test the base Resource representation + cherrypy.request.method = 'GET' + cherrypy.request.headers['Accept'] = 'application/json' + self.assertEquals('{}', r.index()) + + # POST and DELETE raise HTTP:405 by default + for method in ('POST', 'DELETE'): + cherrypy.request.method = method + try: + r.index() + except cherrypy.HTTPError, e: + self.assertEquals(405, e.code) + else: + self.fail("Expected exception not raised") + + def test_404(self): + """ + A non-existent path should return HTTP:404 + """ + url_list = ['/plugins/kimchi/doesnotexist', '/plugins/kimchi/vms/blah'] + for url in url_list: + resp = self.request(url) + self.assertEquals(404, resp.status) + + # Verify it works for DELETE too + resp = self.request('/plugins/kimchi/templates/blah', '', 'DELETE') + self.assertEquals(404, resp.status) + + def test_accepts(self): + """ + Verify the following expectations regarding the client Accept header: + If omitted, default to html + If 'application/json', serve the rest api + If 'text/html', serve the UI + If both of the above (in any order), serve the rest api + If neither of the above, HTTP:406 + """ + resp = self.request("/", headers={}) + location = resp.getheader('location') + self.assertTrue(location.endswith("login.html")) + resp = self.request("/login.html", headers={}) + self.assertTrue('<!doctype html>' in resp.read().lower()) + + resp = self.request("/", headers={'Accept': 'application/json'}) + self.assertValidJSON(resp.read()) + + resp = self.request("/", headers={'Accept': 'text/html'}) + location = resp.getheader('location') + self.assertTrue(location.endswith("login.html")) + + resp = self.request("/", headers={'Accept': + 'application/json, text/html'}) + self.assertValidJSON(resp.read()) + + resp = self.request("/", headers={'Accept': + 'text/html, application/json'}) + self.assertValidJSON(resp.read()) + + h = {'Accept': 'text/plain'} + resp = self.request('/', None, 'GET', h) + self.assertEquals(406, resp.status) + + def test_auth_unprotected(self): + hdrs = {'AUTHORIZATION': ''} + uris = ['/plugins/kimchi/js/kimchi.min.js', + '/plugins/kimchi/css/theme-default.min.css', + '/plugins/kimchi/images/icon-vm.png', + '/libs/jquery-1.10.0.min.js', + '/login.html', + '/logout'] + + for uri in uris: + resp = self.request(uri, None, 'HEAD', hdrs) + self.assertEquals(200, resp.status) + + def test_auth_protected(self): + hdrs = {'AUTHORIZATION': ''} + uris = ['/plugins/kimchi/vms', + '/plugins/kimchi/vms/doesnotexist', + '/tasks'] + + for uri in uris: + resp = self.request(uri, None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + def test_auth_bad_creds(self): + # Test HTTPBA + hdrs = {'AUTHORIZATION': "Basic " + base64.b64encode("nouser:badpass")} + resp = self.request('/plugins/kimchi/vms', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + # Test REST API + hdrs = {'AUTHORIZATION': ''} + req = json.dumps({'username': 'nouser', 'password': 'badpass'}) + resp = self.request('/login', req, 'POST', hdrs) + self.assertEquals(401, resp.status) + + def test_auth_browser_no_httpba(self): + # Kimchi detects REST requests from the browser by looking for a + # specific header + hdrs = {"X-Requested-With": "XMLHttpRequest"} + + # Try our request (Note that request() will add a valid HTTPBA header) + resp = self.request('/plugins/kimchi/vms', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + self.assertEquals(None, resp.getheader('WWW-Authenticate')) + + def test_auth_session(self): + hdrs = {'AUTHORIZATION': '', + 'Content-Type': 'application/json', + 'Accept': 'application/json'} + + # Test we are logged out + resp = self.request('/tasks', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + # Execute a login call + user, pw = mockmodel.fake_user.items()[0] + req = json.dumps({'username': user, 'password': pw}) + resp = self.request('/login', req, 'POST', hdrs) + self.assertEquals(200, resp.status) + + user_info = json.loads(resp.read()) + self.assertEquals(sorted(user_info.keys()), + ['groups', 'roles', 'username']) + roles = user_info['roles'] + for tab, role in roles.iteritems(): + self.assertEquals(role, u'admin') + + cookie = resp.getheader('set-cookie') + hdrs['Cookie'] = cookie + + # Test we are logged in with the cookie + resp = self.request('/tasks', None, 'GET', hdrs) + self.assertEquals(200, resp.status) + + # Execute a logout call + resp = self.request('/logout', '{}', 'POST', hdrs) + self.assertEquals(200, resp.status) + del hdrs['Cookie'] + + # Test we are logged out + resp = self.request('/tasks', None, 'GET', hdrs) + self.assertEquals(401, resp.status) + + def test_get_param(self): + # Create a mock ISO file + mockiso = '/tmp/mock.iso' + open('/tmp/mock.iso', 'w').close() + + # Create 2 different templates + req = json.dumps({'name': 'test-tmpl1', 'cdrom': mockiso}) + self.request('/plugins/kimchi/templates', req, 'POST') + + req = json.dumps({'name': 'test-tmpl2', 'cdrom': mockiso}) + self.request('/plugins/kimchi/templates', req, 'POST') + + # Remove mock iso + os.unlink(mockiso) + + # Get the templates + resp = self.request('/plugins/kimchi/templates') + self.assertEquals(200, resp.status) + res = json.loads(resp.read()) + self.assertEquals(2, len(res)) + + # Get a specific template + resp = self.request('/plugins/kimchi/templates?name=test-tmpl1') + self.assertEquals(200, resp.status) + res = json.loads(resp.read()) + self.assertEquals(1, len(res)) + self.assertEquals('test-tmpl1', res[0]['name']) diff --git a/src/wok/plugins/kimchi/tests/test_utils.py b/src/wok/plugins/kimchi/tests/test_utils.py new file mode 100644 index 0000000..bcb14e2 --- /dev/null +++ b/src/wok/plugins/kimchi/tests/test_utils.py @@ -0,0 +1,69 @@ +# +# 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 unittest + +from wok.exception import InvalidParameter +from wok.utils import convert_data_size + + +class UtilsTests(unittest.TestCase): + def test_convert_data_size(self): + failure_data = [{'val': None, 'from': 'MiB'}, + {'val': self, 'from': 'MiB'}, + {'val': 1, 'from': None}, + {'val': 1, 'from': ''}, + {'val': 1, 'from': 'foo'}, + {'val': 1, 'from': 'kib'}, + {'val': 1, 'from': 'MiB', 'to': None}, + {'val': 1, 'from': 'MiB', 'to': ''}, + {'val': 1, 'from': 'MiB', 'to': 'foo'}, + {'val': 1, 'from': 'MiB', 'to': 'kib'}] + + for d in failure_data: + if 'to' in d: + self.assertRaises(InvalidParameter, convert_data_size, + d['val'], d['from'], d['to']) + else: + self.assertRaises(InvalidParameter, convert_data_size, + d['val'], d['from']) + + success_data = [{'got': convert_data_size(5, 'MiB', 'MiB'), + 'want': 5}, + {'got': convert_data_size(5, 'MiB', 'KiB'), + 'want': 5120}, + {'got': convert_data_size(5, 'MiB', 'M'), + 'want': 5.24288}, + {'got': convert_data_size(5, 'MiB', 'GiB'), + 'want': 0.0048828125}, + {'got': convert_data_size(5, 'MiB', 'Tb'), + 'want': 4.194304e-05}, + {'got': convert_data_size(5, 'KiB', 'MiB'), + 'want': 0.0048828125}, + {'got': convert_data_size(5, 'M', 'MiB'), + 'want': 4.76837158203125}, + {'got': convert_data_size(5, 'GiB', 'MiB'), + 'want': 5120}, + {'got': convert_data_size(5, 'Tb', 'MiB'), + 'want': 596046.4477539062}, + {'got': convert_data_size(5, 'MiB'), + 'want': convert_data_size(5, 'MiB', 'B')}] + + for d in success_data: + self.assertEquals(d['got'], d['want']) -- 2.4.3