[Kimchi-devel] [PATCH 10/16] V6 Ginger Base : base plugin tests files
chandra at linux.vnet.ibm.com
chandra at linux.vnet.ibm.com
Fri Oct 16 14:39:10 UTC 2015
From: chandrureddy <chandra at linux.vnet.ibm.com>
---
src/wok/plugins/gingerbase/mockmodel.py | 78 ------
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 ------------
10 files changed, 1298 insertions(+), 240 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/mockmodel.py b/src/wok/plugins/gingerbase/mockmodel.py
index 62ea509..1feadfc 100644
--- a/src/wok/plugins/gingerbase/mockmodel.py
+++ b/src/wok/plugins/gingerbase/mockmodel.py
@@ -42,7 +42,6 @@ class MockModel(Model):
# Override osinfo.defaults to ajust the values according to
# test:///default driver
- self._mock_devices = MockDevices()
self._mock_partitions = MockPartitions()
self._mock_swupdate = MockSoftwareUpdate()
self._mock_repositories = MockRepositories()
@@ -176,83 +175,6 @@ class MockPartitions(object):
"size": "2147483648"}}
-class MockDevices(object):
- def __init__(self):
- self.devices = {
- 'computer': {'device_type': 'system',
- 'firmware': {'release_date': '01/01/2012',
- 'vendor': 'LENOVO',
- 'version': 'XXXXX (X.XX )'},
- 'hardware': {'serial': 'PXXXXX',
- 'uuid':
- '9d660370-820f-4241-8731-5a60c97e8aa6',
- 'vendor': 'LENOVO',
- 'version': 'ThinkPad T420'},
- 'name': 'computer',
- 'parent': None,
- 'product': '4180XXX'},
- 'pci_0000_03_00_0': {'bus': 3,
- 'device_type': 'pci',
- 'domain': 0,
- 'driver': {'name': 'iwlwifi'},
- 'function': 0,
- 'iommuGroup': 7,
- 'name': 'pci_0000_03_00_0',
- 'parent': 'computer',
- 'path':
- '/sys/devices/pci0000:00/0000:03:00.0',
- 'product': {
- 'description':
- 'Centrino Advanced-N 6205 [Taylor Peak]',
- 'id': '0x0085'},
- 'slot': 0,
- 'vendor': {'description': 'Intel Corporation',
- 'id': '0x8086'}},
- 'pci_0000_0d_00_0': {'bus': 13,
- 'device_type': 'pci',
- 'domain': 0,
- 'driver': {'name': 'sdhci-pci'},
- 'function': 0,
- 'iommuGroup': 7,
- 'name': 'pci_0000_0d_00_0',
- 'parent': 'computer',
- 'path':
- '/sys/devices/pci0000:00/0000:0d:00.0',
- 'product': {'description':
- 'PCIe SDXC/MMC Host Controller',
- 'id': '0xe823'},
- 'slot': 0,
- 'vendor': {'description': 'Ricoh Co Ltd',
- 'id': '0x1180'}},
- 'scsi_host0': {'adapter': {'fabric_wwn': '37df6c1efa1b4388',
- 'type': 'fc_host',
- 'wwnn': 'efb6563f06434a98',
- 'wwpn': '742f32073aab45d7'},
- 'device_type': 'scsi_host',
- 'host': 0,
- 'name': 'scsi_host0',
- 'parent': 'computer',
- 'path': '/sys/devices/pci0000:00/0000:40:00.0/0'},
- 'scsi_host1': {'adapter': {'fabric_wwn': '542efa5dced34123',
- 'type': 'fc_host',
- 'wwnn': 'b7433a40c9b84092',
- 'wwpn': '25c1f485ae42497f'},
- 'device_type': 'scsi_host',
- 'host': 0,
- 'name': 'scsi_host1',
- 'parent': 'computer',
- 'path': '/sys/devices/pci0000:00/0000:40:00.0/1'},
- 'scsi_host2': {'adapter': {'fabric_wwn': '5c373c334c20478d',
- 'type': 'fc_host',
- 'wwnn': 'f2030bec4a254e6b',
- 'wwpn': '07dbca4164d44096'},
- 'device_type': 'scsi_host',
- 'host': 0,
- 'name': 'scsi_host2',
- 'parent': 'computer',
- 'path': '/sys/devices/pci0000:00/0000:40:00.0/2'}}
-
-
class MockSoftwareUpdate(object):
def __init__(self):
self.pkgs = {
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
+
+
+ at unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test')
+def setUpModule():
+ global TEMP_REPO_FILE
+ TEMP_REPO_FILE = _create_fake_repos_file()
+
+
+ at unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test')
+def tearDownModule():
+ os.remove(TEMP_REPO_FILE)
+
+
+ at 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
-
-
- at unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test')
-def setUpModule():
- global TEMP_REPO_FILE
- TEMP_REPO_FILE = _create_fake_repos_file()
-
-
- at unittest.skipIf(not _is_yum_distro(), 'Skipping: YUM exclusive test')
-def tearDownModule():
- os.remove(TEMP_REPO_FILE)
-
-
- at 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
More information about the Kimchi-devel
mailing list