[Kimchi-devel] [PATCH] Gingerbase: adapting unit tests to use WoK /tests/utils.py

From: Daniel Henrique Barboza <dhbarboza82(a)gmail.com&gt; As a follow up of the recent changes in the WoK /tests/utils.py module in Kimchi, the same procedure is now being done in Gingerbase: - unit tests now points to tests.utils - Gingerbase tests/utils.py module was deleted Signed-off-by: Daniel Henrique Barboza <dhbarboza82(a)gmail.com&gt; --- src/wok/plugins/ginger | 2 +- src/wok/plugins/gingerbase/tests/run_tests.sh.in | 3 +- .../plugins/gingerbase/tests/test_authorization.py | 3 +- src/wok/plugins/gingerbase/tests/test_host.py | 4 +- src/wok/plugins/gingerbase/tests/test_rest.py | 6 +- src/wok/plugins/gingerbase/tests/utils.py | 262 --------------------- 6 files changed, 11 insertions(+), 269 deletions(-) delete mode 100644 src/wok/plugins/gingerbase/tests/utils.py diff --git a/src/wok/plugins/ginger b/src/wok/plugins/ginger index 84f36db..a0cc2a3 160000 --- a/src/wok/plugins/ginger +++ b/src/wok/plugins/ginger @@ -1 +1 @@ -Subproject commit 84f36dbbfe63869fa97cd68e9abdeaffdf1846d9 +Subproject commit a0cc2a369abf5ec93068924c3b681957d1c82cb3 diff --git a/src/wok/plugins/gingerbase/tests/run_tests.sh.in b/src/wok/plugins/gingerbase/tests/run_tests.sh.in index 3b08fa6..ab3de12 100644 --- a/src/wok/plugins/gingerbase/tests/run_tests.sh.in +++ b/src/wok/plugins/gingerbase/tests/run_tests.sh.in @@ -52,6 +52,7 @@ for ((i=0;i<${#LIST[@]};i++)); do fi done +# ../../../../../ refers to wok root # ../../../../ refers to wok directory # ../../../ refers to plugins directory -PYTHONPATH=../../../../:../../../ $CMD $OPTS ${MODEL_LIST[@]} ${MOCK_LIST[@]} +PYTHONPATH=../../../../../:../../../../:../../../ $CMD $OPTS ${MODEL_LIST[@]} ${MOCK_LIST[@]} diff --git a/src/wok/plugins/gingerbase/tests/test_authorization.py b/src/wok/plugins/gingerbase/tests/test_authorization.py index d0fc57c..0650f0a 100644 --- a/src/wok/plugins/gingerbase/tests/test_authorization.py +++ b/src/wok/plugins/gingerbase/tests/test_authorization.py @@ -23,9 +23,10 @@ import os import unittest from functools import partial +from tests.utils import get_free_port, patch_auth, request, run_server + from wok.plugins.gingerbase import mockmodel -from utils import get_free_port, patch_auth, request, run_server test_server = None model = None diff --git a/src/wok/plugins/gingerbase/tests/test_host.py b/src/wok/plugins/gingerbase/tests/test_host.py index 4f3e848..5d158bb 100644 --- a/src/wok/plugins/gingerbase/tests/test_host.py +++ b/src/wok/plugins/gingerbase/tests/test_host.py @@ -29,8 +29,10 @@ import time import unittest from functools import partial +from tests.utils import get_free_port, patch_auth, request +from tests.utils import run_server, wait_task + from wok.plugins.gingerbase.mockmodel import MockModel -from utils import get_free_port, patch_auth, request, run_server, wait_task test_server = None diff --git a/src/wok/plugins/gingerbase/tests/test_rest.py b/src/wok/plugins/gingerbase/tests/test_rest.py index a8138c6..c723a0e 100644 --- a/src/wok/plugins/gingerbase/tests/test_rest.py +++ b/src/wok/plugins/gingerbase/tests/test_rest.py @@ -26,13 +26,13 @@ import time import unittest from functools import partial +from tests.utils import get_free_port, patch_auth, request +from tests.utils import run_server, wait_task + from wok.rollbackcontext import RollbackContext from wok.plugins.gingerbase import mockmodel -from utils import get_free_port, patch_auth, request -from utils import run_server, wait_task - test_server = None model = None diff --git a/src/wok/plugins/gingerbase/tests/utils.py b/src/wok/plugins/gingerbase/tests/utils.py deleted file mode 100644 index 111cb02..0000000 --- a/src/wok/plugins/gingerbase/tests/utils.py +++ /dev/null @@ -1,262 +0,0 @@ -# -# Project Ginger Base -# -# Copyright IBM, Corp. 2013-2015 -# -# Code derived from Project Kimchi -# -# 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 grp -import httplib -import inspect -import json -import os -import socket -import ssl -import sys -import threading -import time -import unittest -from contextlib import closing -from lxml import etree - -import wok.server -from wok.config import config, PluginPaths -from wok.auth import User, USER_NAME, USER_GROUPS, USER_ROLES, tabs -from wok.exception import NotFoundError, OperationFailed -from wok.utils import wok_log - -from wok.plugins.gingerbase import mockmodel - - -_ports = {} - -# provide missing unittest decorators and API for python 2.6; these decorators -# do not actually work, just avoid the syntax failure -if sys.version_info[:2] == (2, 6): - def skipUnless(condition, reason): - if not condition: - sys.stderr.write('[expected failure] ') - raise Exception(reason) - return lambda obj: obj - - unittest.skipUnless = skipUnless - unittest.expectedFailure = lambda obj: obj - - def assertGreater(self, a, b, msg=None): - if not a > b: - self.fail('%s not greater than %s' % (repr(a), repr(b))) - - def assertGreaterEqual(self, a, b, msg=None): - if not a >= b: - self.fail('%s not greater than or equal to %s' - % (repr(a), repr(b))) - - def assertIsInstance(self, obj, cls, msg=None): - if not isinstance(obj, cls): - self.fail('%s is not an instance of %r' % (repr(obj), cls)) - - def assertIn(self, a, b, msg=None): - if a not in b: - self.fail("%s is not in %b" % (repr(a), repr(b))) - - def assertNotIn(self, a, b, msg=None): - if a in b: - self.fail("%s is in %b" % (repr(a), repr(b))) - - unittest.TestCase.assertGreaterEqual = assertGreaterEqual - unittest.TestCase.assertGreater = assertGreater - unittest.TestCase.assertIsInstance = assertIsInstance - unittest.TestCase.assertIn = assertIn - unittest.TestCase.assertNotIn = assertNotIn - - -def get_free_port(name='http'): - global _ports - if _ports.get(name) is not None: - return _ports[name] - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - with closing(sock): - try: - sock.bind(("0.0.0.0", 0)) - except: - raise Exception("Could not find a free port") - _ports[name] = sock.getsockname()[1] - return _ports[name] - - -def run_server(host, port, ssl_port, test_mode, cherrypy_port=None, - model=None, environment='development'): - - if cherrypy_port is None: - cherrypy_port = get_free_port('cherrypy_port') - - if ssl_port is None: - ssl_port = get_free_port('https') - - args = type('_', (object,), - {'host': host, 'port': port, 'ssl_port': ssl_port, - 'cherrypy_port': cherrypy_port, 'max_body_size': '4*1024', - 'ssl_cert': '', 'ssl_key': '', - 'test': test_mode, 'access_log': '/dev/null', - 'error_log': '/dev/null', 'environment': environment, - 'log_level': 'debug'})() - if model is not None: - setattr(args, 'model', model) - - s = wok.server.Server(args) - t = threading.Thread(target=s.start) - t.setDaemon(True) - t.start() - cherrypy.engine.wait(cherrypy.engine.states.STARTED) - return s - - -def silence_server(): - """ - Silence server status messages on stdout - """ - cherrypy.config.update({"environment": "embedded"}) - - -def running_as_root(): - return os.geteuid() == 0 - - -def _request(conn, path, data, method, headers): - if headers is None: - headers = {'Content-Type': 'application/json', - 'Accept': 'application/json'} - if 'AUTHORIZATION' not in headers.keys(): - user, pw = mockmodel.fake_user.items()[0] - hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw)) - headers['AUTHORIZATION'] = hdr - conn.request(method, path, data, headers) - return conn.getresponse() - - -def request(host, port, path, data=None, method='GET', headers=None): - # verify if HTTPSConnection has context parameter - if "context" in inspect.getargspec(httplib.HTTPSConnection.__init__).args: - context = ssl._create_unverified_context() - conn = httplib.HTTPSConnection(host, port, context=context) - else: - conn = httplib.HTTPSConnection(host, port) - - return _request(conn, path, data, method, headers) - - -def get_remote_iso_path(): - """ - Get a remote iso with the right arch from the distro files shipped - with gingerbase. - """ - host_arch = os.uname()[4] - remote_path = '' - with open(os.path.join(PluginPaths('gingerbase').conf_dir, 'distros.d', - 'fedora.json')) as fedora_isos: - # Get a list of dicts - json_isos_list = json.load(fedora_isos) - for iso in json_isos_list: - if (iso.get('os_arch')) == host_arch: - remote_path = iso.get('path') - break - - return remote_path - - -class FakeUser(User): - auth_type = "fake" - sudo = True - - def __init__(self, username): - self.user = {} - self.user[USER_NAME] = username - self.user[USER_GROUPS] = None - self.user[USER_ROLES] = dict.fromkeys(tabs, 'user') - - def get_groups(self): - return sorted([group.gr_name for group in grp.getgrall()])[0:3] - - def get_roles(self): - if self.sudo: - self.user[USER_ROLES] = dict.fromkeys(tabs, 'admin') - return self.user[USER_ROLES] - - def get_user(self): - return self.user - - @staticmethod - def authenticate(username, password, service="passwd"): - try: - return mockmodel.fake_user[username] == password - except KeyError, e: - raise OperationFailed("GGBAUTH0001E", {'username': 'username', - 'code': e.message}) - - -def patch_auth(sudo=True): - """ - Override the authenticate function with a simple test against an - internal dict of users and passwords. - """ - config.set("authentication", "method", "fake") - FakeUser.sudo = sudo - - -def normalize_xml(xml_str): - return etree.tostring(etree.fromstring(xml_str, - etree.XMLParser(remove_blank_text=True))) - - -def wait_task(task_lookup, taskid, timeout=10): - for i in range(0, timeout): - task_info = task_lookup(taskid) - if task_info['status'] == "running": - wok_log.info("Waiting task %s, message: %s", - taskid, task_info['message']) - time.sleep(1) - else: - return - wok_log.error("Timeout while process long-run task, " - "try to increase timeout value.") - - -# The action functions in model backend raise NotFoundError exception if the -# element is not found. But in some tests, these functions are called after -# the element has been deleted if test finishes correctly, then NofFoundError -# exception is raised and rollback breaks. To avoid it, this wrapper ignores -# the NotFoundError. -def rollback_wrapper(func, resource, *args): - try: - func(resource, *args) - except NotFoundError: - # VM has been deleted already - return - - -# This function is used to test storage volume upload. -# If we use self.request, we may encode multipart formdata by ourselves -# requests lib take care of encode part, so use this lib instead -def fake_auth_header(): - headers = {'Accept': 'application/json'} - user, pw = mockmodel.fake_user.items()[0] - hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw)) - headers['AUTHORIZATION'] = hdr - return headers -- 2.4.3