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

From: chandrureddy <chandra(a)linux.vnet.ibm.com&gt; --- src/wok/plugins/gingerbase/tests/Makefile.am | 49 ++++ src/wok/plugins/gingerbase/tests/run_tests.sh.in | 55 ++++ src/wok/plugins/gingerbase/tests/test_config.py.in | 147 +++++++++++ src/wok/plugins/gingerbase/tests/test_host.py | 154 ++++++++++++ src/wok/plugins/gingerbase/tests/test_model.py | 276 +++++++++++++++++++++ src/wok/plugins/gingerbase/tests/test_rest.py | 190 ++++++++++++++ src/wok/plugins/gingerbase/tests/test_yumparser.py | 165 ++++++++++++ src/wok/plugins/gingerbase/tests/utils.py | 262 +++++++++++++++++++ src/wok/plugins/kimchi/tests/test_yumparser.py | 162 ------------ 9 files changed, 1298 insertions(+), 162 deletions(-) create mode 100644 src/wok/plugins/gingerbase/tests/Makefile.am create mode 100644 src/wok/plugins/gingerbase/tests/run_tests.sh.in create mode 100644 src/wok/plugins/gingerbase/tests/test_config.py.in create mode 100644 src/wok/plugins/gingerbase/tests/test_host.py create mode 100644 src/wok/plugins/gingerbase/tests/test_model.py create mode 100644 src/wok/plugins/gingerbase/tests/test_rest.py create mode 100644 src/wok/plugins/gingerbase/tests/test_yumparser.py create mode 100644 src/wok/plugins/gingerbase/tests/utils.py delete mode 100644 src/wok/plugins/kimchi/tests/test_yumparser.py diff --git a/src/wok/plugins/gingerbase/tests/Makefile.am b/src/wok/plugins/gingerbase/tests/Makefile.am new file mode 100644 index 0000000..477df12 --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/Makefile.am @@ -0,0 +1,49 @@ +# +# Ginger Base +# +# 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 + +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/src/wok/plugins/gingerbase/tests/run_tests.sh.in b/src/wok/plugins/gingerbase/tests/run_tests.sh.in new file mode 100644 index 0000000..1a34418 --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/run_tests.sh.in @@ -0,0 +1,55 @@ +#!/bin/bash +# +# Project Ginger Base +# +# 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/src/wok/plugins/gingerbase/tests/test_config.py.in b/src/wok/plugins/gingerbase/tests/test_config.py.in new file mode 100644 index 0000000..da8ab50 --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/test_config.py.in @@ -0,0 +1,147 @@ +# +# Project Ginger Base +# +# Copyright IBM, Corp. 2014-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 unittest +from cherrypy.lib.reprconf import Parser + +from wok.config import CACHEEXPIRES, SESSIONSTIMEOUT +from wok.config import Paths, PluginPaths + +from wok.plugins.gingerbase.config import get_debugreports_path +from wok.plugins.gingerbase.config import GingerBaseConfig, GingerBasePaths + +get_prefix = None + + +def setUpModule(): + global get_prefix + get_prefix = Paths.get_prefix + + +def tearDownModule(): + Paths.get_prefix = GingerBasePaths.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_plugin_paths(self): + GingerBasePaths.get_prefix = lambda self: '@datadir@/wok' + paths = GingerBasePaths() + 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/gingerbase') + self.assertInstalledPath(paths.ui_dir, + '@datadir@/wok/plugins/gingerbase/ui') + self.assertInstalledPath(paths.mo_dir, '@prefix@/share/locale') + + def test_uninstalled_plugin_paths(self): + paths = GingerBasePaths() + prefix = paths.prefix + self.assertEquals(paths.conf_dir, '%s/src/wok/plugins/gingerbase' + % prefix) + self.assertEquals(paths.conf_file, + '%s/src/wok/plugins/gingerbase/gingerbase.conf' + % prefix) + self.assertEquals(paths.src_dir, '%s/src/wok/plugins/gingerbase' + % prefix) + self.assertEquals(paths.ui_dir, + '%s/src/wok/plugins/gingerbase/ui' % prefix) + self.assertEquals(paths.mo_dir, + '%s/src/wok/plugins/gingerbase/mo' % prefix) + + def test_gingerbase_config(self): + GingerBasePaths.get_prefix = PluginPaths.get_prefix = get_prefix + paths = GingerBasePaths() + pluginPrefix = paths.add_prefix(paths.plugin_dir) + configObj = { + 'wok': { + 'plugin_class': 'GingerBase', + 'enable': True, + 'uri': '/plugins/gingerbase', + 'extra_auth_api_class': 'control.sub_nodes' + }, + '/': { + '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': True + }, + '/help': { + 'tools.nocache.on': True, + 'tools.staticdir.dir': '%s/ui/pages/help' % pluginPrefix, + 'tools.staticdir.on': True + }, + '/js': { + 'tools.wokauth.on': False, + 'tools.nocache.on': False, + 'tools.staticdir.dir': '%s/ui/js' % pluginPrefix, + 'tools.expires.on': True, + 'tools.expires.secs': CACHEEXPIRES, + 'tools.staticdir.on': True + }, + '/css': { + 'tools.wokauth.on': False, + 'tools.nocache.on': False, + 'tools.staticdir.dir': '%s/ui/css' % pluginPrefix, + 'tools.expires.on': True, + 'tools.expires.secs': CACHEEXPIRES, + 'tools.staticdir.on': True + }, + '/images': { + 'tools.wokauth.on': False, + 'tools.nocache.on': False, + 'tools.staticdir.dir': '%s/ui/images' % pluginPrefix, + 'tools.staticdir.on': True + }, + '/ui/config/tab-ext.xml': { + 'tools.nocache.on': True, + 'tools.staticfile.on': True, + 'tools.staticfile.filename': '%s/ui/config/tab-ext.xml' % + pluginPrefix, + }, + '/data/debugreports': { + 'tools.wokauth.on': True, + 'tools.nocache.on': False, + 'tools.staticdir.dir': get_debugreports_path(), + 'tools.staticdir.content_types': {'xz': 'application/x-xz'}, + 'tools.staticdir.on': True + } + } + + gingerbase_config = \ + Parser().dict_from_file(GingerBasePaths().conf_file) + gingerbase_config.update(GingerBaseConfig()) + self.assertEquals(gingerbase_config, configObj) diff --git a/src/wok/plugins/gingerbase/tests/test_host.py b/src/wok/plugins/gingerbase/tests/test_host.py new file mode 100644 index 0000000..8c2e39d --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/test_host.py @@ -0,0 +1,154 @@ +# -*- coding: utf-8 -*- +# +# Project Ginger Base +# +# Copyright IBM, Corp. 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 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('/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('/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)) \ No newline at end of file diff --git a/src/wok/plugins/gingerbase/tests/test_model.py b/src/wok/plugins/gingerbase/tests/test_model.py new file mode 100644 index 0000000..2f2207b --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/test_model.py @@ -0,0 +1,276 @@ +# -*- coding: utf-8 -*- +# +# 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 os +import shutil +import unittest + +import wok.objectstore +from wok.basemodel import Singleton +from wok.exception import InvalidParameter, NotFoundError +from wok.rollbackcontext import RollbackContext +from wok.plugins.gingerbase.model import model + +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/gingerbase/tests/' + + +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/gingerbase-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 Ginger Base + # 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 Ginger Base + # 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 Ginger Base + # 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']) \ No newline at end of file diff --git a/src/wok/plugins/gingerbase/tests/test_rest.py b/src/wok/plugins/gingerbase/tests/test_rest.py new file mode 100644 index 0000000..75729cb --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/test_rest.py @@ -0,0 +1,190 @@ +# -*- coding: utf-8 -*- +# +# 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 json +import os +import time +import unittest +from functools import partial + +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 +host = None +port = None +ssl_port = None +cherrypy_port = None + + +def setUpModule(): + global test_server, model, host, port, ssl_port, cherrypy_port + + patch_auth() + model = mockmodel.MockModel('/tmp/obj-store-test') + host = '127.0.0.1' + port = get_free_port('http') + ssl_port = get_free_port('https') + cherrypy_port = get_free_port('cherrypy_port') + test_server = run_server(host, port, ssl_port, test_mode=True, + cherrypy_port=cherrypy_port, model=model) + + +def tearDownModule(): + test_server.stop() + os.unlink('/tmp/obj-store-test') + + +class RestTests(unittest.TestCase): + def _async_op(self, cb, opaque): + time.sleep(1) + cb('success', True) + + def _except_op(self, cb, opaque): + time.sleep(1) + raise Exception("Oops, this is an exception handle test." + " You can ignore it safely") + cb('success', True) + + def _intermid_op(self, cb, opaque): + time.sleep(1) + cb('in progress') + + def setUp(self): + self.request = partial(request, host, ssl_port) + model.reset() + + def _task_lookup(self, taskid): + return json.loads( + self.request('/tasks/%s' % taskid).read() + ) + + def assertHTTPStatus(self, code, *args): + resp = self.request(*args) + self.assertEquals(code, resp.status) + + def test_debugreports(self): + resp = request(host, ssl_port, '/plugins/gingerbase/debugreports') + self.assertEquals(200, resp.status) + + def _report_delete(self, name): + request(host, ssl_port, '/plugins/gingerbase/debugreports/%s' % name, + '{}', 'DELETE') + + def test_create_debugreport(self): + req = json.dumps({'name': 'report1'}) + with RollbackContext() as rollback: + resp = request(host, ssl_port, '/plugins/gingerbase/debugreports', + req, 'POST') + self.assertEquals(202, resp.status) + task = json.loads(resp.read()) + # make sure the debugreport doesn't exist until the + # the task is finished + wait_task(self._task_lookup, task['id']) + rollback.prependDefer(self._report_delete, 'report2') + resp = request(host, ssl_port, + '/plugins/gingerbase/debugreports/report1') + debugreport = json.loads(resp.read()) + self.assertEquals("report1", debugreport['name']) + self.assertEquals(200, resp.status) + req = json.dumps({'name': 'report2'}) + resp = request(host, ssl_port, + '/plugins/gingerbase/debugreports/report1', + req, 'PUT') + self.assertEquals(303, resp.status) + + def test_debugreport_download(self): + req = json.dumps({'name': 'report1'}) + with RollbackContext() as rollback: + resp = request(host, ssl_port, '/plugins/gingerbase/debugreports', + req, 'POST') + self.assertEquals(202, resp.status) + task = json.loads(resp.read()) + # make sure the debugreport doesn't exist until the + # the task is finished + wait_task(self._task_lookup, task['id'], 20) + rollback.prependDefer(self._report_delete, 'report1') + resp = request(host, ssl_port, + '/plugins/gingerbase/debugreports/report1') + debugreport = json.loads(resp.read()) + self.assertEquals("report1", debugreport['name']) + self.assertEquals(200, resp.status) + resp = request(host, ssl_port, + '/plugins/gingerbase/debugreports/report1/content') + self.assertEquals(200, resp.status) + resp = request(host, ssl_port, + '/plugins/gingerbase/debugreports/report1') + debugre = json.loads(resp.read()) + resp = request(host, ssl_port, debugre['uri']) + self.assertEquals(200, resp.status) + + def test_repositories(self): + def verify_repo(t, res): + for field in ('repo_id', 'enabled', 'baseurl', 'config'): + if field in t.keys(): + self.assertEquals(t[field], res[field]) + + base_uri = '/plugins/gingerbase/host/repositories' + resp = self.request(base_uri) + self.assertEquals(200, resp.status) + # Already have one repo in Kimchi's system + self.assertEquals(1, len(json.loads(resp.read()))) + + # Create a repository + repo = {'repo_id': 'fedora-fake', + 'baseurl': 'http://www.fedora.org'} + req = json.dumps(repo) + resp = self.request(base_uri, req, 'POST') + self.assertEquals(201, resp.status) + + # Verify the repository + res = json.loads(self.request('%s/fedora-fake' % base_uri).read()) + verify_repo(repo, res) + + # Update the repository + params = {} + params['baseurl'] = repo['baseurl'] = 'http://www.fedoraproject.org' + resp = self.request('%s/fedora-fake' % base_uri, json.dumps(params), + 'PUT') + + # Verify the repository + res = json.loads(self.request('%s/fedora-fake' % base_uri).read()) + verify_repo(repo, res) + + # Delete the repository + resp = self.request('%s/fedora-fake' % base_uri, '{}', 'DELETE') + self.assertEquals(204, resp.status) + + +class HttpsRestTests(RestTests): + """ + Run all of the same tests as above, but use https instead + """ + def setUp(self): + self.request = partial(request, host, ssl_port) + model.reset() diff --git a/src/wok/plugins/gingerbase/tests/test_yumparser.py b/src/wok/plugins/gingerbase/tests/test_yumparser.py new file mode 100644 index 0000000..caf7c27 --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/test_yumparser.py @@ -0,0 +1,165 @@ +# +# Project Ginger Base +# +# Copyright IBM, Corp. 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 os +import tempfile +import unittest + +from wok.rollbackcontext import RollbackContext + +from wok.plugins.gingerbase.model import model +from wok.plugins.gingerbase.yumparser \ + import delete_repo_from_file, get_repo_files +from wok.plugins.gingerbase.yumparser import get_yum_packages_list_update +from wok.plugins.gingerbase.yumparser import get_yum_repositories +from wok.plugins.gingerbase.yumparser import write_repo_to_file, YumRepoObject + + +TEMP_REPO_FILE = '' + + +def _is_yum_distro(): + inst = model.Model() + repo_type = inst.capabilities_lookup()['repo_mngt_tool'] + return repo_type == 'yum' + + +def _create_fake_repos(repo_file_name): + repo1 = YumRepoObject('fake-repo-1', repo_file_name) + repo2 = YumRepoObject('fake-repo-2', repo_file_name) + repo3 = YumRepoObject('fake-repo-3', repo_file_name) + repo4 = YumRepoObject('fake-repo-4', repo_file_name) + repos = [repo1, repo2, repo3, repo4] + return repos + + +def _create_empty_repo_file(): + data = """ +# +# This is a repository file with no repositories at all +# No repositories must be added after reading this file. +# + """ + _, tmp_file_name = tempfile.mkstemp(suffix='.repo', + dir='/etc/yum.repos.d') + with open(tmp_file_name, 'w') as f: + f.writelines(data) + + return tmp_file_name + + +def _create_fake_repos_file(): + _, tmp_file_name = tempfile.mkstemp(suffix='.repo', + dir='/etc/yum.repos.d') + + fake_repos = _create_fake_repos(tmp_file_name) + file_data = '' + for repo in fake_repos: + file_data += str(repo) + '\n' + + with open(tmp_file_name, 'w') as f: + f.writelines(file_data) + + return tmp_file_name + + +def _generate_yumcheckupdate_output(): + output = """ +Repository 'REPOSITORY1' is missing name in configuration, using id +Repository 'REPOSITORY1-OPTIONAL' is missing name in configuration, using id + +PACKAGE1.noarch 20150611.-gg-FAKE1 REPOSITORY1 +PACKAGE2.x86_64 20150611.-no-FAKE2 REPOSITORY2 +PACKAGE3.dot.dot.i386 20150611.-re-FAKE3 REPOSITORY3 + +Obsoleting Packages +OBSOLETE4.dot.dot.i386 20150611.FAKE4 REPOSITORY4 +OBSOLETE5.dot.dot.fakearch 20150611.FAKE5 REPOSITORY5 + """ + return output + + +(a)unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test') +def setUpModule(): + global TEMP_REPO_FILE + TEMP_REPO_FILE = _create_fake_repos_file() + + +(a)unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test') +def tearDownModule(): + os.remove(TEMP_REPO_FILE) + + +(a)unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test') +class YumParserTests(unittest.TestCase): + + def test_get_yum_repositories(self): + repo_files = get_repo_files() + repo_objects = get_yum_repositories() + self.assertGreaterEqual(len(repo_objects), len(repo_files)) + + def test_empty_repo_file(self): + with RollbackContext() as rollback: + repos = get_yum_repositories() + tmp_file_name = _create_empty_repo_file() + rollback.prependDefer(os.remove, tmp_file_name) + repos_after = get_yum_repositories() + self.assertEqual(len(repos_after), len(repos)) + + def test_update_repo_attributes(self): + repos = get_yum_repositories() + fake_repo_2 = repos['fake-repo-2'] + fake_repo_2.disable() + fake_repo_2.name = 'This is a fake repo' + fake_repo_2.baseurl = 'http://a.fake.repo.url' + fake_repo_2.gpgkey = 'file://a/fake/gpg/key.fake' + fake_repo_2.gpgcheck = False + fake_repo_2.metalink = 'this is not a true metalink' + fake_repo_2.mirrorlist = 'fake mirrorlist' + write_repo_to_file(fake_repo_2) + + repos = get_yum_repositories() + fake_repo_2 = repos['fake-repo-2'] + self.assertEqual(False, fake_repo_2.enabled) + self.assertEqual(False, fake_repo_2.gpgcheck) + self.assertEqual('This is a fake repo', fake_repo_2.name) + self.assertEqual('http://a.fake.repo.url', fake_repo_2.baseurl) + self.assertEqual('file://a/fake/gpg/key.fake', fake_repo_2.gpgkey) + self.assertEqual('this is not a true metalink', fake_repo_2.metalink) + self.assertEqual('fake mirrorlist', fake_repo_2.mirrorlist) + + def test_delete_repo_from_file(self): + repos = get_yum_repositories() + fake_repo_3 = repos['fake-repo-3'] + delete_repo_from_file(fake_repo_3) + + repos = get_yum_repositories() + repos_id = repos.keys() + self.assertNotIn('fake-repo-3', repos_id) + + def test_yum_checkupdate_parsing(self): + output = _generate_yumcheckupdate_output() + packages = get_yum_packages_list_update(output) + self.assertEqual(len(packages), 3) + self.assertEqual(packages[0].ui_from_repo, 'REPOSITORY1') + self.assertEqual(packages[1].version, '20150611.-no-FAKE2') + self.assertEqual(packages[2].name, 'PACKAGE3.dot.dot') + self.assertEqual(packages[2].arch, 'i386') diff --git a/src/wok/plugins/gingerbase/tests/utils.py b/src/wok/plugins/gingerbase/tests/utils.py new file mode 100644 index 0000000..13054dd --- /dev/null +++ b/src/wok/plugins/gingerbase/tests/utils.py @@ -0,0 +1,262 @@ +# +# 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 kimchi. + """ + 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 diff --git a/src/wok/plugins/kimchi/tests/test_yumparser.py b/src/wok/plugins/kimchi/tests/test_yumparser.py deleted file mode 100644 index be5e95c..0000000 --- a/src/wok/plugins/kimchi/tests/test_yumparser.py +++ /dev/null @@ -1,162 +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 os -import tempfile -import unittest - -from wok.rollbackcontext import RollbackContext - -from wok.plugins.kimchi.model import model -from wok.plugins.kimchi.yumparser import delete_repo_from_file, get_repo_files -from wok.plugins.kimchi.yumparser import get_yum_packages_list_update -from wok.plugins.kimchi.yumparser import get_yum_repositories -from wok.plugins.kimchi.yumparser import write_repo_to_file, YumRepoObject - - -TEMP_REPO_FILE = '' - - -def _is_yum_distro(): - inst = model.Model('test:///default') - repo_type = inst.capabilities_lookup()['repo_mngt_tool'] - return repo_type == 'yum' - - -def _create_fake_repos(repo_file_name): - repo1 = YumRepoObject('fake-repo-1', repo_file_name) - repo2 = YumRepoObject('fake-repo-2', repo_file_name) - repo3 = YumRepoObject('fake-repo-3', repo_file_name) - repo4 = YumRepoObject('fake-repo-4', repo_file_name) - repos = [repo1, repo2, repo3, repo4] - return repos - - -def _create_empty_repo_file(): - data = """ -# -# This is a repository file with no repositories at all -# No repositories must be added after reading this file. -# - """ - _, tmp_file_name = tempfile.mkstemp(suffix='.repo', - dir='/etc/yum.repos.d') - with open(tmp_file_name, 'w') as f: - f.writelines(data) - - return tmp_file_name - - -def _create_fake_repos_file(): - _, tmp_file_name = tempfile.mkstemp(suffix='.repo', - dir='/etc/yum.repos.d') - - fake_repos = _create_fake_repos(tmp_file_name) - file_data = '' - for repo in fake_repos: - file_data += str(repo) + '\n' - - with open(tmp_file_name, 'w') as f: - f.writelines(file_data) - - return tmp_file_name - - -def _generate_yumcheckupdate_output(): - output = """ -Repository 'REPOSITORY1' is missing name in configuration, using id -Repository 'REPOSITORY1-OPTIONAL' is missing name in configuration, using id - -PACKAGE1.noarch 20150611.-gg-FAKE1 REPOSITORY1 -PACKAGE2.x86_64 20150611.-no-FAKE2 REPOSITORY2 -PACKAGE3.dot.dot.i386 20150611.-re-FAKE3 REPOSITORY3 - -Obsoleting Packages -OBSOLETE4.dot.dot.i386 20150611.FAKE4 REPOSITORY4 -OBSOLETE5.dot.dot.fakearch 20150611.FAKE5 REPOSITORY5 - """ - return output - - -(a)unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test') -def setUpModule(): - global TEMP_REPO_FILE - TEMP_REPO_FILE = _create_fake_repos_file() - - -(a)unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test') -def tearDownModule(): - os.remove(TEMP_REPO_FILE) - - -(a)unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test') -class YumParserTests(unittest.TestCase): - - def test_get_yum_repositories(self): - repo_files = get_repo_files() - repo_objects = get_yum_repositories() - self.assertGreaterEqual(len(repo_objects), len(repo_files)) - - def test_empty_repo_file(self): - with RollbackContext() as rollback: - repos = get_yum_repositories() - tmp_file_name = _create_empty_repo_file() - rollback.prependDefer(os.remove, tmp_file_name) - repos_after = get_yum_repositories() - self.assertEqual(len(repos_after), len(repos)) - - def test_update_repo_attributes(self): - repos = get_yum_repositories() - fake_repo_2 = repos['fake-repo-2'] - fake_repo_2.disable() - fake_repo_2.name = 'This is a fake repo' - fake_repo_2.baseurl = 'http://a.fake.repo.url' - fake_repo_2.gpgkey = 'file://a/fake/gpg/key.fake' - fake_repo_2.gpgcheck = False - fake_repo_2.metalink = 'this is not a true metalink' - fake_repo_2.mirrorlist = 'fake mirrorlist' - write_repo_to_file(fake_repo_2) - - repos = get_yum_repositories() - fake_repo_2 = repos['fake-repo-2'] - self.assertEqual(False, fake_repo_2.enabled) - self.assertEqual(False, fake_repo_2.gpgcheck) - self.assertEqual('This is a fake repo', fake_repo_2.name) - self.assertEqual('http://a.fake.repo.url', fake_repo_2.baseurl) - self.assertEqual('file://a/fake/gpg/key.fake', fake_repo_2.gpgkey) - self.assertEqual('this is not a true metalink', fake_repo_2.metalink) - self.assertEqual('fake mirrorlist', fake_repo_2.mirrorlist) - - def test_delete_repo_from_file(self): - repos = get_yum_repositories() - fake_repo_3 = repos['fake-repo-3'] - delete_repo_from_file(fake_repo_3) - - repos = get_yum_repositories() - repos_id = repos.keys() - self.assertNotIn('fake-repo-3', repos_id) - - def test_yum_checkupdate_parsing(self): - output = _generate_yumcheckupdate_output() - packages = get_yum_packages_list_update(output) - self.assertEqual(len(packages), 3) - self.assertEqual(packages[0].ui_from_repo, 'REPOSITORY1') - self.assertEqual(packages[1].version, '20150611.-no-FAKE2') - self.assertEqual(packages[2].name, 'PACKAGE3.dot.dot') - self.assertEqual(packages[2].arch, 'i386') -- 2.1.0