[Kimchi-devel] [PATCH 08/17] Ginger Base : base plugin tests files

From: chandrureddy <chandra(a)linux.vnet.ibm.com&gt; --- plugins/gingerbase/tests/Makefile.am | 49 +++++ plugins/gingerbase/tests/run_tests.sh.in | 55 ++++++ plugins/gingerbase/tests/test_config.py.in | 154 ++++++++++++++++ plugins/gingerbase/tests/test_host.py | 152 ++++++++++++++++ plugins/gingerbase/tests/test_model.py | 280 +++++++++++++++++++++++++++++ plugins/gingerbase/tests/utils.py | 261 +++++++++++++++++++++++++++ plugins/kimchi/tests/test_host.py | 200 --------------------- 7 files changed, 951 insertions(+), 200 deletions(-) create mode 100644 plugins/gingerbase/tests/Makefile.am create mode 100644 plugins/gingerbase/tests/run_tests.sh.in create mode 100644 plugins/gingerbase/tests/test_config.py.in create mode 100644 plugins/gingerbase/tests/test_host.py create mode 100644 plugins/gingerbase/tests/test_model.py create mode 100644 plugins/gingerbase/tests/utils.py delete mode 100644 plugins/kimchi/tests/test_host.py diff --git a/plugins/gingerbase/tests/Makefile.am b/plugins/gingerbase/tests/Makefile.am new file mode 100644 index 0000000..23e18f4 --- /dev/null +++ b/plugins/gingerbase/tests/Makefile.am @@ -0,0 +1,49 @@ +# +# Kimchi +# +# Copyright IBM Corp, 2013 +# +# 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 + +EXTRA_DIST = \ + Makefile.am \ + run_tests.sh.in \ + test_config.py.in \ + $(filter-out test_config.py, $(wildcard *.py)) \ + $(NULL) + +noinst_SCRIPTS = run_tests.sh + +do_substitution = \ + sed -e 's,[@]HAVE_PYMOD_UNITTEST[@],$(HAVE_PYMOD_UNITTEST),g' \ + -e 's,[@]prefix[@],$(prefix),g' \ + -e 's,[@]datadir[@],$(datadir),g' \ + -e 's,[@]PYTHON_VERSION[@],$(PYTHON_VERSION),g' \ + -e 's,[@]wokdir[@],$(pythondir)/wok,g' \ + -e 's,[@]pkgdatadir[@],$(pkgdatadir),g' + + +run_tests.sh: run_tests.sh.in Makefile + $(do_substitution) < $(srcdir)/run_tests.sh.in > run_tests.sh + chmod +x run_tests.sh + +test_config.py: test_config.py.in Makefile + $(do_substitution) < $(srcdir)/test_config.py.in > test_config.py + +check-local: + ./run_tests.sh + +BUILT_SOURCES = test_config.py +CLEANFILES = run_tests.sh test_config.py diff --git a/plugins/gingerbase/tests/run_tests.sh.in b/plugins/gingerbase/tests/run_tests.sh.in new file mode 100644 index 0000000..beef75e --- /dev/null +++ b/plugins/gingerbase/tests/run_tests.sh.in @@ -0,0 +1,55 @@ +#!/bin/bash +# +# 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 + +HAVE_UNITTEST=@HAVE_PYMOD_UNITTEST@ +PYTHON_VER=@PYTHON_VERSION@ + +if [ "$1" = "-v" ]; then + OPTS="-v" + shift +else + OPTS="" +fi + +if [ $# -ne 0 ]; then + ARGS="$@" +else + ARGS=`find -name "test_*.py" | xargs -I @ basename @ .py` +fi + +if [ "$HAVE_UNITTEST" != "yes" -o "$PYTHON_VER" == "2.6" ]; then + CMD="unit2" +else + CMD="python -m unittest" +fi + +LIST=($ARGS) +MODEL_LIST=() +MOCK_LIST=() +for ((i=0;i<${#LIST[@]};i++)); do + + if [[ ${LIST[$i]} == test_model* ]]; then + MODEL_LIST+=(${LIST[$i]}) + else + MOCK_LIST+=(${LIST[$i]}) + fi +done + +PYTHONPATH=../plugins:../src:../ $CMD $OPTS ${MODEL_LIST[@]} ${MOCK_LIST[@]} diff --git a/plugins/gingerbase/tests/test_config.py.in b/plugins/gingerbase/tests/test_config.py.in new file mode 100644 index 0000000..4ffe919 --- /dev/null +++ b/plugins/gingerbase/tests/test_config.py.in @@ -0,0 +1,154 @@ +# +# Project Kimchi +# +# Copyright IBM, Corp. 2014-2015 +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +import unittest +from cherrypy.lib.reprconf import Parser +from wok.config import Paths + +from wok.plugins.gingerbase.config import KimchiPaths + + + +get_prefix = None + + +def setUpModule(): + global get_prefix + get_prefix = Paths.get_prefix + + +def tearDownModule(): + Paths.get_prefix = KimchiPaths.get_prefix = get_prefix + + +class ConfigTests(unittest.TestCase): + def assertInstalledPath(self, actual, expected): + if '@pkgdatadir@' != '/usr/share/gingerbase': + usr_local = '/usr/local' + if not expected.startswith('/usr'): + expected = usr_local + expected + self.assertEquals(actual, expected) + + def test_installed_paths(self): + Paths.get_prefix = lambda self: '@datadir@/wok' + paths = Paths() + self.assertInstalledPath(paths.state_dir, '/var/lib/wok') + self.assertInstalledPath(paths.log_dir, '/var/log/wok') + self.assertInstalledPath(paths.conf_dir, '/etc/wok') + self.assertInstalledPath(paths.src_dir, '@wokdir@') + self.assertInstalledPath(paths.plugins_dir, '@wokdir@/plugins') + self.assertInstalledPath(paths.ui_dir, '@datadir@/wok/ui') + self.assertInstalledPath(paths.mo_dir, '@prefix@/share/locale') + + def test_uninstalled_paths(self): + Paths.get_prefix = lambda self: '/home/user/wok' + paths = Paths() + self.assertEquals(paths.state_dir, '/home/user/wok/data') + self.assertEquals(paths.log_dir, '/home/user/wok/log') + self.assertEquals(paths.conf_dir, '/home/user/wok/src') + self.assertEquals(paths.src_dir, '/home/user/wok/src/wok') + self.assertEquals(paths.plugins_dir, '/home/user/wok/plugins') + self.assertEquals(paths.ui_dir, '/home/user/wok/ui') + self.assertEquals(paths.mo_dir, '/home/user/wok/mo') + + def test_installed_plugin_paths(self): + KimchiPaths.get_prefix = lambda self: '@datadir@/wok' + paths = KimchiPaths() + self.assertInstalledPath(paths.conf_dir, '/etc/wok/plugins.d') + self.assertInstalledPath(paths.conf_file, + '/etc/wok/plugins.d/gingerbase.conf') + self.assertInstalledPath(paths.src_dir, '@wokdir@/plugins/ginger-base') + self.assertInstalledPath(paths.ui_dir, + '@datadir@/wok/plugins/gingerbase/ui') + self.assertInstalledPath(paths.mo_dir, '@prefix@/share/locale') + + def test_uninstalled_plugin_paths(self): + KimchiPaths.get_prefix = lambda self: '/home/user/wok' + paths = KimchiPaths() + self.assertEquals(paths.conf_dir, '/home/user/wok/plugins/gingerbase') + self.assertEquals( + paths.conf_file, '/home/user/wok/plugins/gingerbase/gingerbase.conf') + self.assertEquals(paths.src_dir, '/home/user/wok/plugins/gingerbase') + self.assertEquals(paths.ui_dir, '/home/user/wok/plugins/gingerbase/ui') + self.assertEquals(paths.mo_dir, '/home/user/wok/plugins/gingerbase/mo') + + def test_kimchi_config(self): + Paths.get_prefix = KimchiPaths.get_prefix = get_prefix + CACHEEXPIRES = 31536000 + SESSIONSTIMEOUT = 10 + configObj = { + '/': { + 'tools.trailing_slash.on': False, + 'request.methods_with_bodies': ('POST', 'PUT'), + 'tools.nocache.on': True, + 'tools.proxy.on': True, + 'tools.sessions.on': True, + 'tools.sessions.name': 'wok', + 'tools.sessions.secure': True, + 'tools.sessions.httponly': True, + 'tools.sessions.locking': 'explicit', + 'tools.sessions.storage_type': 'ram', + 'tools.sessions.timeout': SESSIONSTIMEOUT, + 'tools.wokauth.on': False + }, + '/css': { + 'tools.staticdir.on': True, + 'tools.staticdir.dir': '%s/ui/css' % KimchiPaths().prefix, + 'tools.expires.on': True, + 'tools.expires.secs': CACHEEXPIRES, + 'tools.nocache.on': False + }, + '/js': { + 'tools.staticdir.on': True, + 'tools.staticdir.dir': '%s/ui/js' % KimchiPaths().prefix, + 'tools.expires.on': True, + 'tools.expires.secs': CACHEEXPIRES, + 'tools.nocache.on': False + }, + '/libs': { + 'tools.staticdir.on': True, + 'tools.staticdir.dir': '%s/ui/libs' % KimchiPaths().prefix, + 'tools.expires.on': True, + 'tools.expires.secs': CACHEEXPIRES, + 'tools.nocache.on': False, + }, + '/images': { + 'tools.staticdir.on': True, + 'tools.staticdir.dir': '%s/ui/images' % KimchiPaths().prefix, + 'tools.nocache.on': False + }, + '/favicon.ico': { + 'tools.staticfile.on': True, + 'tools.staticfile.filename': + '%s/images/logo.ico' % KimchiPaths().ui_dir + }, + '/robots.txt': { + 'tools.staticfile.on': True, + 'tools.staticfile.filename': '%s/robots.txt' % KimchiPaths().ui_dir + }, + '/help': { + 'tools.staticdir.on': True, + 'tools.staticdir.dir': '%s/ui/pages/help' % KimchiPaths().prefix, + 'tools.staticdir.index': 'en_US/index.html', + 'tools.nocache.on': True + } + } + + kimchi_config = Parser().dict_from_file(KimchiPaths().conf_file) + self.assertEquals(kimchi_config, configObj) diff --git a/plugins/gingerbase/tests/test_host.py b/plugins/gingerbase/tests/test_host.py new file mode 100644 index 0000000..78765c0 --- /dev/null +++ b/plugins/gingerbase/tests/test_host.py @@ -0,0 +1,152 @@ +# -*- coding: utf-8 -*- +# +# Project Kimchi +# +# Copyright IBM, Corp. 2015 +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + +import json +import os +import platform +import psutil +import tempfile +import time +import unittest +from functools import partial + +from wok.plugins.gingerbase.mockmodel import MockModel +from utils import get_free_port, patch_auth, request, run_server, wait_task + + +test_server = None +model = None +host = None +ssl_port = None +tmpfile = None + + +def setUpModule(): + global test_server, model, host, ssl_port, tmpfile + + patch_auth() + tmpfile = tempfile.mktemp() + model = MockModel(tmpfile) + host = '127.0.0.1' + port = get_free_port('http') + ssl_port = get_free_port('https') + cherrypy_port = get_free_port('cherrypy_port') + test_server = run_server(host, port, ssl_port, test_mode=True, + cherrypy_port=cherrypy_port, model=model) + + +def tearDownModule(): + test_server.stop() + os.unlink(tmpfile) + + +class HostTests(unittest.TestCase): + def setUp(self): + self.request = partial(request, host, ssl_port) + + def test_hostinfo(self): + resp = self.request('/plugins/gingerbase/host').read() + info = json.loads(resp) + keys = ['os_distro', 'os_version', 'os_codename', 'cpu_model', + 'memory', 'cpus'] + self.assertEquals(sorted(keys), sorted(info.keys())) + + distro, version, codename = platform.linux_distribution() + self.assertEquals(distro, info['os_distro']) + self.assertEquals(version, info['os_version']) + self.assertEquals(unicode(codename, "utf-8"), info['os_codename']) + self.assertEquals(psutil.TOTAL_PHYMEM, info['memory']) + + def test_hoststats(self): + time.sleep(1) + stats_keys = ['cpu_utilization', 'memory', 'disk_read_rate', + 'disk_write_rate', 'net_recv_rate', 'net_sent_rate'] + resp = self.request('/plugins/gingerbase/host/stats').read() + stats = json.loads(resp) + self.assertEquals(sorted(stats_keys), sorted(stats.keys())) + + cpu_utilization = stats['cpu_utilization'] + self.assertIsInstance(cpu_utilization, float) + self.assertGreaterEqual(cpu_utilization, 0.0) + self.assertTrue(cpu_utilization <= 100.0) + + memory_stats = stats['memory'] + self.assertIn('total', memory_stats) + self.assertIn('free', memory_stats) + self.assertIn('cached', memory_stats) + self.assertIn('buffers', memory_stats) + self.assertIn('avail', memory_stats) + + resp = self.request('/plugins/gingerbase/host/stats/history').read() + history = json.loads(resp) + self.assertEquals(sorted(stats_keys), sorted(history.keys())) + + def test_host_actions(self): + def _task_lookup(taskid): + return json.loads(self.request('/plugins/gingerbase/tasks/%s' % + taskid).read()) + + resp = self.request('/plugins/gingerbase/host/shutdown', '{}', 'POST') + self.assertEquals(200, resp.status) + resp = self.request('/plugins/gingerbase/host/reboot', '{}', 'POST') + self.assertEquals(200, resp.status) + + # Test system update + resp = self.request('/plugins/gingerbase/host/packagesupdate', None, 'GET') + pkgs = json.loads(resp.read()) + self.assertEquals(3, len(pkgs)) + + pkg_keys = ['package_name', 'repository', 'arch', 'version'] + for p in pkgs: + name = p['package_name'] + resp = self.request('/plugins/gingerbase/host/packagesupdate/' + name, + None, 'GET') + info = json.loads(resp.read()) + self.assertEquals(sorted(pkg_keys), sorted(info.keys())) + + resp = self.request('/plugins/gingerbase/host/swupdate', '{}', 'POST') + task = json.loads(resp.read()) + task_params = [u'id', u'message', u'status', u'target_uri'] + self.assertEquals(sorted(task_params), sorted(task.keys())) + + resp = self.request('/plugins/gingerbase/tasks/' + task[u'id'], None, 'GET') + task_info = json.loads(resp.read()) + self.assertEquals(task_info['status'], 'running') + wait_task(_task_lookup, task_info['id']) + resp = self.request('/plugins/gingerbase/tasks/' + task[u'id'], None, 'GET') + task_info = json.loads(resp.read()) + self.assertEquals(task_info['status'], 'finished') + self.assertIn(u'All packages updated', task_info['message']) + pkgs = model.packagesupdate_get_list() + self.assertEquals(0, len(pkgs)) + + def test_host_partitions(self): + resp = self.request('/plugins/gingerbase/host/partitions') + self.assertEquals(200, resp.status) + partitions = json.loads(resp.read()) + + keys = ['name', 'path', 'type', 'fstype', 'size', 'mountpoint', + 'available'] + for item in partitions: + resp = self.request('/plugins/gingerbase/host/partitions/%s' % + item['name']) + info = json.loads(resp.read()) + self.assertEquals(sorted(info.keys()), sorted(keys)) + diff --git a/plugins/gingerbase/tests/test_model.py b/plugins/gingerbase/tests/test_model.py new file mode 100644 index 0000000..d471ce5 --- /dev/null +++ b/plugins/gingerbase/tests/test_model.py @@ -0,0 +1,280 @@ +# -*- coding: utf-8 -*- +# +# 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 os +import shutil +import unittest + +import wok.objectstore +from wok.basemodel import Singleton +from wok.exception import InvalidParameter, NotFoundError, OperationFailed +from wok.rollbackcontext import RollbackContext +from wok.plugins.gingerbase.model import model + +# import iso_gen +# import utils + + +invalid_repository_urls = ['www.fedora.org', # missing protocol + '://www.fedora.org', # missing protocol + 'http://www.fedora', # invalid domain name + 'file:///home/foobar'] # invalid path + +TMP_DIR = '/var/lib/kimchi/tests/' +# UBUNTU_ISO = TMP_DIR + 'ubuntu14.04.iso' + + +def setUpModule(): + if not os.path.exists(TMP_DIR): + os.makedirs(TMP_DIR) + + # iso_gen.construct_fake_iso(UBUNTU_ISO, True, '14.04', 'ubuntu') + + # Some FeatureTests functions depend on server to validate their result. + # As CapabilitiesModel is a Singleton class it will get the first result + # from FeatureTests which may be wrong when using the Model instance + # directly - the case of this test_model.py + # So clean Singleton instances to make sure to get the right result when + # running the following tests. + Singleton._instances = {} + + +def tearDownModule(): + shutil.rmtree(TMP_DIR) + + +class ModelTests(unittest.TestCase): + def setUp(self): + self.tmp_store = '/tmp/kimchi-store-test' + + def tearDown(self): + # FIXME: Tests using 'test:///default' URI should be moved to + # test_rest or test_mockmodel to avoid overriding problems + # LibvirtConnection._connections['test:///default'] = {} + + os.unlink(self.tmp_store) + + def test_repository_create(self): + inst = model.Model(objstore_loc=self.tmp_store) + + yum_repos = [{'repo_id': 'fedora-fake', + 'baseurl': 'http://www.fedora.org'}, + {'repo_id': 'fedora-updates-fake', + 'config': + {'mirrorlist': 'http://www.fedoraproject.org'}}] + + deb_repos = [{'baseurl': 'http://archive.ubuntu.com/ubuntu/', + 'config': {'dist': 'quantal'}}, + {'baseurl': 'http://archive.ubuntu.com/ubuntu/', + 'config': {'dist': 'quantal', 'comps': ['main']}}] + + yum_invalid_repos = [] + deb_invalid_repos = [] + + for url in invalid_repository_urls: + wrong_baseurl = {'repo_id': 'wrong-id', 'baseurl': url} + wrong_mirrorlist = {'repo_id': 'wrong-id', + 'baseurl': 'www.example.com', + 'config': {'mirrorlist': url}} + wrong_config_item = { + 'repo_id': 'wrong-id', + 'baseurl': 'www.example.com', + 'config': { + 'gpgkey': 'file:///tmp/KEY-fedora-updates-fake-19'}} + + yum_invalid_repos.append(wrong_baseurl) + yum_invalid_repos.append(wrong_mirrorlist) + yum_invalid_repos.append(wrong_config_item) + + wrong_baseurl['config'] = {'dist': 'tasty'} + wrong_config = {'baseurl': deb_repos[0]['baseurl'], + 'config': { + 'unsupported_item': "a_unsupported_item"}} + deb_invalid_repos.append(wrong_baseurl) + deb_invalid_repos.append(wrong_config) + + repo_type = inst.capabilities_lookup()['repo_mngt_tool'] + if repo_type == 'yum': + test_repos = yum_repos + invalid_repos = yum_invalid_repos + elif repo_type == 'deb': + test_repos = deb_repos + invalid_repos = deb_invalid_repos + else: + # repository management tool was not recognized by Kimchi + # skip test case + return + + # create repositories with invalid data + for repo in invalid_repos: + self.assertRaises(InvalidParameter, inst.repositories_create, repo) + + for repo in test_repos: + system_host_repos = len(inst.repositories_get_list()) + repo_id = inst.repositories_create(repo) + host_repos = inst.repositories_get_list() + self.assertEquals(system_host_repos + 1, len(host_repos)) + + repo_info = inst.repository_lookup(repo_id) + self.assertEquals(repo_id, repo_info['repo_id']) + self.assertEquals(True, repo_info.get('enabled')) + self.assertEquals(repo.get('baseurl', ''), + repo_info.get('baseurl')) + + original_config = repo.get('config', {}) + config_info = repo_info.get('config', {}) + + if repo_type == 'yum': + self.assertEquals(original_config.get('mirrorlist', ''), + config_info.get('mirrorlist', '')) + self.assertEquals(True, config_info['gpgcheck']) + else: + self.assertEquals(original_config['dist'], config_info['dist']) + self.assertEquals(original_config.get('comps', []), + config_info.get('comps', [])) + + inst.repository_delete(repo_id) + self.assertRaises(NotFoundError, inst.repository_lookup, repo_id) + + self.assertRaises(NotFoundError, inst.repository_lookup, 'google') + + def test_repository_update(self): + inst = model.Model(objstore_loc=self.tmp_store) + + yum_repo = {'repo_id': 'fedora-fake', + 'baseurl': 'http://www.fedora.org'} + yum_new_repo = {'baseurl': 'http://www.fedoraproject.org'} + + deb_repo = {'baseurl': 'http://archive.ubuntu.com/ubuntu/', + 'config': {'dist': 'quantal'}} + deb_new_repo = {'baseurl': 'http://br.archive.canonical.com/ubuntu/', + 'config': {'dist': 'utopic'}} + + yum_invalid_repos = [] + deb_invalid_repos = [] + + for url in invalid_repository_urls: + wrong_baseurl = {'baseurl': url} + wrong_mirrorlist = {'baseurl': 'www.example.com', + 'config': {'mirrorlist': url}} + + yum_invalid_repos.append(wrong_baseurl) + yum_invalid_repos.append(wrong_mirrorlist) + + wrong_baseurl['config'] = {'dist': 'tasty'} + deb_invalid_repos.append(wrong_baseurl) + + repo_type = inst.capabilities_lookup()['repo_mngt_tool'] + if repo_type == 'yum': + repo = yum_repo + new_repo = yum_new_repo + invalid_repos = yum_invalid_repos + elif repo_type == 'deb': + repo = deb_repo + new_repo = deb_new_repo + invalid_repos = deb_invalid_repos + else: + # repository management tool was not recognized by Kimchi + # skip test case + return + + system_host_repos = len(inst.repositories_get_list()) + + with RollbackContext() as rollback: + repo_id = inst.repositories_create(repo) + rollback.prependDefer(inst.repository_delete, repo_id) + + host_repos = inst.repositories_get_list() + self.assertEquals(system_host_repos + 1, len(host_repos)) + + # update repositories with invalid data + for tmp_repo in invalid_repos: + self.assertRaises(InvalidParameter, inst.repository_update, + repo_id, tmp_repo) + + new_repo_id = inst.repository_update(repo_id, new_repo) + repo_info = inst.repository_lookup(new_repo_id) + + self.assertEquals(new_repo_id, repo_info['repo_id']) + self.assertEquals(new_repo['baseurl'], repo_info['baseurl']) + self.assertEquals(True, repo_info['enabled']) + inst.repository_update(new_repo_id, repo) + + def test_repository_disable_enable(self): + inst = model.Model(objstore_loc=self.tmp_store) + + yum_repo = {'repo_id': 'fedora-fake', + 'baseurl': 'http://www.fedora.org'} + deb_repo = {'baseurl': 'http://archive.ubuntu.com/ubuntu/', + 'config': {'dist': 'quantal'}} + + repo_type = inst.capabilities_lookup()['repo_mngt_tool'] + if repo_type == 'yum': + repo = yum_repo + elif repo_type == 'deb': + repo = deb_repo + else: + # repository management tool was not recognized by Kimchi + # skip test case + return + + system_host_repos = len(inst.repositories_get_list()) + + repo_id = inst.repositories_create(repo) + + host_repos = inst.repositories_get_list() + self.assertEquals(system_host_repos + 1, len(host_repos)) + + repo_info = inst.repository_lookup(repo_id) + self.assertEquals(True, repo_info['enabled']) + + inst.repository_disable(repo_id) + repo_info = inst.repository_lookup(repo_id) + self.assertEquals(False, repo_info['enabled']) + + inst.repository_enable(repo_id) + repo_info = inst.repository_lookup(repo_id) + self.assertEquals(True, repo_info['enabled']) + + # remove files creates + inst.repository_delete(repo_id) + + +class BaseModelTests(unittest.TestCase): + class FoosModel(object): + def __init__(self): + self.data = {} + + def create(self, params): + self.data.update(params) + + def get_list(self): + return list(self.data) + + class TestModel(wok.basemodel.BaseModel): + def __init__(self): + foo = BaseModelTests.FoosModel() + super(BaseModelTests.TestModel, self).__init__([foo]) + + def test_root_model(self): + t = BaseModelTests.TestModel() + t.foos_create({'item1': 10}) + self.assertEquals(t.foos_get_list(), ['item1']) + diff --git a/plugins/gingerbase/tests/utils.py b/plugins/gingerbase/tests/utils.py new file mode 100644 index 0000000..986d91b --- /dev/null +++ b/plugins/gingerbase/tests/utils.py @@ -0,0 +1,261 @@ +# +# 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 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 kimchi. + """ + host_arch = os.uname()[4] + remote_path = '' + with open(os.path.join(PluginPaths('kimchi').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 + diff --git a/plugins/kimchi/tests/test_host.py b/plugins/kimchi/tests/test_host.py deleted file mode 100644 index e2aa196..0000000 --- a/plugins/kimchi/tests/test_host.py +++ /dev/null @@ -1,200 +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 json -import os -import platform -import psutil -import tempfile -import time -import unittest -from functools import partial - -from wok.plugins.kimchi.mockmodel import MockModel - -from utils import get_free_port, patch_auth, request, run_server, wait_task - - -test_server = None -model = None -host = None -ssl_port = None -tmpfile = None - - -def setUpModule(): - global test_server, model, host, ssl_port, tmpfile - - patch_auth() - tmpfile = tempfile.mktemp() - model = MockModel(tmpfile) - host = '127.0.0.1' - port = get_free_port('http') - ssl_port = get_free_port('https') - cherrypy_port = get_free_port('cherrypy_port') - test_server = run_server(host, port, ssl_port, test_mode=True, - cherrypy_port=cherrypy_port, model=model) - - -def tearDownModule(): - test_server.stop() - os.unlink(tmpfile) - - -class HostTests(unittest.TestCase): - def setUp(self): - self.request = partial(request, host, ssl_port) - - def test_hostinfo(self): - resp = self.request('/plugins/kimchi/host').read() - info = json.loads(resp) - keys = ['os_distro', 'os_version', 'os_codename', 'cpu_model', - 'memory', 'cpus'] - self.assertEquals(sorted(keys), sorted(info.keys())) - - distro, version, codename = platform.linux_distribution() - self.assertEquals(distro, info['os_distro']) - self.assertEquals(version, info['os_version']) - self.assertEquals(unicode(codename, "utf-8"), info['os_codename']) - self.assertEquals(psutil.TOTAL_PHYMEM, info['memory']) - - def test_hoststats(self): - time.sleep(1) - stats_keys = ['cpu_utilization', 'memory', 'disk_read_rate', - 'disk_write_rate', 'net_recv_rate', 'net_sent_rate'] - resp = self.request('/plugins/kimchi/host/stats').read() - stats = json.loads(resp) - self.assertEquals(sorted(stats_keys), sorted(stats.keys())) - - cpu_utilization = stats['cpu_utilization'] - self.assertIsInstance(cpu_utilization, float) - self.assertGreaterEqual(cpu_utilization, 0.0) - self.assertTrue(cpu_utilization <= 100.0) - - memory_stats = stats['memory'] - self.assertIn('total', memory_stats) - self.assertIn('free', memory_stats) - self.assertIn('cached', memory_stats) - self.assertIn('buffers', memory_stats) - self.assertIn('avail', memory_stats) - - resp = self.request('/plugins/kimchi/host/stats/history').read() - history = json.loads(resp) - self.assertEquals(sorted(stats_keys), sorted(history.keys())) - - def test_host_actions(self): - def _task_lookup(taskid): - return json.loads(self.request('/plugins/kimchi/tasks/%s' % - taskid).read()) - - resp = self.request('/plugins/kimchi/host/shutdown', '{}', 'POST') - self.assertEquals(200, resp.status) - resp = self.request('/plugins/kimchi/host/reboot', '{}', 'POST') - self.assertEquals(200, resp.status) - - # Test system update - resp = self.request('/plugins/kimchi/host/packagesupdate', None, 'GET') - pkgs = json.loads(resp.read()) - self.assertEquals(3, len(pkgs)) - - pkg_keys = ['package_name', 'repository', 'arch', 'version'] - for p in pkgs: - name = p['package_name'] - resp = self.request('/plugins/kimchi/host/packagesupdate/' + name, - None, 'GET') - info = json.loads(resp.read()) - self.assertEquals(sorted(pkg_keys), sorted(info.keys())) - - resp = self.request('/plugins/kimchi/host/swupdate', '{}', 'POST') - task = json.loads(resp.read()) - task_params = [u'id', u'message', u'status', u'target_uri'] - self.assertEquals(sorted(task_params), sorted(task.keys())) - - resp = self.request('/plugins/kimchi/tasks/' + task[u'id'], None, 'GET') - task_info = json.loads(resp.read()) - self.assertEquals(task_info['status'], 'running') - wait_task(_task_lookup, task_info['id']) - resp = self.request('/plugins/kimchi/tasks/' + task[u'id'], None, 'GET') - task_info = json.loads(resp.read()) - self.assertEquals(task_info['status'], 'finished') - self.assertIn(u'All packages updated', task_info['message']) - pkgs = model.packagesupdate_get_list() - self.assertEquals(0, len(pkgs)) - - def test_host_partitions(self): - resp = self.request('/plugins/kimchi/host/partitions') - self.assertEquals(200, resp.status) - partitions = json.loads(resp.read()) - - keys = ['name', 'path', 'type', 'fstype', 'size', 'mountpoint', - 'available'] - for item in partitions: - resp = self.request('/plugins/kimchi/host/partitions/%s' % - item['name']) - info = json.loads(resp.read()) - self.assertEquals(sorted(info.keys()), sorted(keys)) - - def test_host_devices(self): - def asset_devices_type(devices, dev_type): - for dev in devices: - self.assertEquals(dev['device_type'], dev_type) - - resp = self.request('/plugins/kimchi/host/devices?_cap=scsi_host') - nodedevs = json.loads(resp.read()) - # Mockmodel brings 3 preconfigured scsi fc_host - self.assertEquals(3, len(nodedevs)) - - nodedev = json.loads(self.request( - '/plugins/kimchi/host/devices/scsi_host2').read()) - # Mockmodel generates random wwpn and wwnn - self.assertEquals('scsi_host2', nodedev['name']) - self.assertEquals('fc_host', nodedev['adapter']['type']) - self.assertEquals(16, len(nodedev['adapter']['wwpn'])) - self.assertEquals(16, len(nodedev['adapter']['wwnn'])) - - devs = json.loads(self.request('/plugins/kimchi/host/devices').read()) - dev_names = [dev['name'] for dev in devs] - for dev_type in ('pci', 'usb_device', 'scsi'): - resp = self.request('/plugins/kimchi/host/devices?_cap=%s' % - dev_type) - devsByType = json.loads(resp.read()) - names = [dev['name'] for dev in devsByType] - self.assertTrue(set(names) <= set(dev_names)) - asset_devices_type(devsByType, dev_type) - - resp = self.request('/plugins/kimchi/host/devices?_passthrough=true') - passthru_devs = [dev['name'] for dev in json.loads(resp.read())] - self.assertTrue(set(passthru_devs) <= set(dev_names)) - - for dev_type in ('pci', 'usb_device', 'scsi'): - resp = self.request( - '/plugins/kimchi/host/devices?_cap=%s&_passthrough=true' % - dev_type) - filteredDevs = json.loads(resp.read()) - filteredNames = [dev['name'] for dev in filteredDevs] - self.assertTrue(set(filteredNames) <= set(dev_names)) - asset_devices_type(filteredDevs, dev_type) - - for dev in passthru_devs: - resp = self.request( - '/plugins/kimchi/host/devices?_passthrough_affected_by=%s' % - dev) - affected_devs = [dev['name'] for dev in json.loads(resp.read())] - self.assertTrue(set(affected_devs) <= set(dev_names)) -- 2.1.0