[Kimchi-devel] [PATCH] Move remaining Kimchi tests to src/wok structure.
Paulo Ricardo Paz Vital
pvital at linux.vnet.ibm.com
Wed Sep 30 16:29:32 UTC 2015
Please, do not consider this.
I'm going to submit a V2 of 'Wok tests' patch rebased with last changes
that will fix this problem.
Regards, Paulo.
On Wed, 2015-09-30 at 13:14 -0300, pvital at linux.vnet.ibm.com wrote:
> From: Paulo Vital <pvital at linux.vnet.ibm.com>
>
> Moving remaining Kimchi tests from old plugins structure to new
> src/wok structure.
>
> Signed-off-by: Paulo Vital <pvital at linux.vnet.ibm.com>
> ---
> 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')
> -
> -
> - at 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')
> +
> +
> + at 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'])
More information about the Kimchi-devel
mailing list