[Kimchi-devel] [PATCH 09/15] V3 Ginger Base : base plugin tests files

chandra at linux.vnet.ibm.com chandra at linux.vnet.ibm.com
Wed Oct 7 10:44:17 UTC 2015


From: chandrureddy <chandra at linux.vnet.ibm.com>

---
 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 | 154 ++++++++++++
 src/wok/plugins/gingerbase/tests/test_host.py      | 152 +++++++++++
 src/wok/plugins/gingerbase/tests/test_model.py     | 280 +++++++++++++++++++++
 src/wok/plugins/gingerbase/tests/test_rest.py      | 182 ++++++++++++++
 src/wok/plugins/gingerbase/tests/utils.py          | 261 +++++++++++++++++++
 7 files changed, 1133 insertions(+)
 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/utils.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..434d60e
--- /dev/null
+++ b/src/wok/plugins/gingerbase/tests/test_config.py.in
@@ -0,0 +1,154 @@
+#
+# Project Ginger Base
+#
+# Copyright IBM, Corp. 2014-2015
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+
+import unittest
+from cherrypy.lib.reprconf import Parser
+from wok.config import Paths
+
+from wok.plugins.gingerbase.config import 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_paths(self):
+        Paths.get_prefix = lambda self: '@datadir@/wok'
+        paths = Paths()
+        self.assertInstalledPath(paths.state_dir, '/var/lib/wok')
+        self.assertInstalledPath(paths.log_dir, '/var/log/wok')
+        self.assertInstalledPath(paths.conf_dir, '/etc/wok')
+        self.assertInstalledPath(paths.src_dir, '@wokdir@')
+        self.assertInstalledPath(paths.plugins_dir, '@wokdir@/plugins')
+        self.assertInstalledPath(paths.ui_dir, '@datadir@/wok/ui')
+        self.assertInstalledPath(paths.mo_dir, '@prefix@/share/locale')
+
+    def test_uninstalled_paths(self):
+        Paths.get_prefix = lambda self: '/home/user/wok'
+        paths = Paths()
+        self.assertEquals(paths.state_dir, '/home/user/wok/data')
+        self.assertEquals(paths.log_dir, '/home/user/wok/log')
+        self.assertEquals(paths.conf_dir, '/home/user/wok/src')
+        self.assertEquals(paths.src_dir, '/home/user/wok/src/wok')
+        self.assertEquals(paths.plugins_dir, '/home/user/wok/plugins')
+        self.assertEquals(paths.ui_dir, '/home/user/wok/ui')
+        self.assertEquals(paths.mo_dir, '/home/user/wok/mo')
+
+    def test_installed_plugin_paths(self):
+        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/ginger-base')
+        self.assertInstalledPath(paths.ui_dir,
+                                 '@datadir@/wok/plugins/gingerbase/ui')
+        self.assertInstalledPath(paths.mo_dir, '@prefix@/share/locale')
+
+    def test_uninstalled_plugin_paths(self):
+        GingerBasePaths.get_prefix = lambda self: '/home/user/wok'
+        paths = GingerBasePaths()
+        self.assertEquals(paths.conf_dir, '/home/user/wok/plugins/gingerbase')
+        self.assertEquals(
+            paths.conf_file, '/home/user/wok/plugins/gingerbase/gingerbase.conf')
+        self.assertEquals(paths.src_dir, '/home/user/wok/plugins/gingerbase')
+        self.assertEquals(paths.ui_dir, '/home/user/wok/plugins/gingerbase/ui')
+        self.assertEquals(paths.mo_dir, '/home/user/wok/plugins/gingerbase/mo')
+
+    def test_kimchi_config(self):
+        Paths.get_prefix = GingerBasePaths.get_prefix = get_prefix
+        CACHEEXPIRES = 31536000
+        SESSIONSTIMEOUT = 10
+        configObj = {
+            '/': {
+                'tools.trailing_slash.on': False,
+                'request.methods_with_bodies': ('POST', 'PUT'),
+                'tools.nocache.on': True,
+                'tools.proxy.on': True,
+                'tools.sessions.on': True,
+                'tools.sessions.name': 'wok',
+                'tools.sessions.secure': True,
+                'tools.sessions.httponly': True,
+                'tools.sessions.locking': 'explicit',
+                'tools.sessions.storage_type': 'ram',
+                'tools.sessions.timeout': SESSIONSTIMEOUT,
+                'tools.wokauth.on': False
+            },
+            '/css': {
+                'tools.staticdir.on': True,
+                'tools.staticdir.dir': '%s/ui/css' % GingerBasePaths().prefix,
+                'tools.expires.on': True,
+                'tools.expires.secs': CACHEEXPIRES,
+                'tools.nocache.on': False
+            },
+            '/js': {
+                'tools.staticdir.on': True,
+                'tools.staticdir.dir': '%s/ui/js' % GingerBasePaths().prefix,
+                'tools.expires.on': True,
+                'tools.expires.secs': CACHEEXPIRES,
+                'tools.nocache.on': False
+            },
+            '/libs': {
+                'tools.staticdir.on': True,
+                'tools.staticdir.dir': '%s/ui/libs' % GingerBasePaths().prefix,
+                'tools.expires.on': True,
+                'tools.expires.secs': CACHEEXPIRES,
+                'tools.nocache.on': False,
+            },
+            '/images': {
+                'tools.staticdir.on': True,
+                'tools.staticdir.dir': '%s/ui/images' % GingerBasePaths().prefix,
+                'tools.nocache.on': False
+            },
+            '/favicon.ico': {
+                'tools.staticfile.on': True,
+                'tools.staticfile.filename':
+                '%s/images/logo.ico' % GingerBasePaths().ui_dir
+            },
+            '/robots.txt': {
+                'tools.staticfile.on': True,
+                'tools.staticfile.filename': '%s/robots.txt' % GingerBasePaths().ui_dir
+            },
+            '/help': {
+                'tools.staticdir.on': True,
+                'tools.staticdir.dir': '%s/ui/pages/help' % GingerBasePaths().prefix,
+                'tools.staticdir.index': 'en_US/index.html',
+                'tools.nocache.on': True
+            }
+        }
+
+        kimchi_config = Parser().dict_from_file(GingerBasePaths().conf_file)
+        self.assertEquals(kimchi_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..6816e90
--- /dev/null
+++ b/src/wok/plugins/gingerbase/tests/test_host.py
@@ -0,0 +1,152 @@
+# -*- coding: utf-8 -*-
+#
+# Project 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
+
+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))
+
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..18199e2
--- /dev/null
+++ b/src/wok/plugins/gingerbase/tests/test_model.py
@@ -0,0 +1,280 @@
+# -*- coding: utf-8 -*-
+#
+# 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
+
+import os
+import shutil
+import unittest
+
+import wok.objectstore
+from wok.basemodel import Singleton
+from wok.exception import InvalidParameter, NotFoundError, OperationFailed
+from wok.rollbackcontext import RollbackContext
+from wok.plugins.gingerbase.model import model
+
+# import iso_gen
+# import utils
+
+
+invalid_repository_urls = ['www.fedora.org',       # missing protocol
+                           '://www.fedora.org',    # missing protocol
+                           'http://www.fedora',    # invalid domain name
+                           'file:///home/foobar']  # invalid path
+
+TMP_DIR = '/var/lib/kimchi/tests/'
+# UBUNTU_ISO = TMP_DIR + 'ubuntu14.04.iso'
+
+
+def setUpModule():
+    if not os.path.exists(TMP_DIR):
+        os.makedirs(TMP_DIR)
+
+    # iso_gen.construct_fake_iso(UBUNTU_ISO, True, '14.04', 'ubuntu')
+
+    # Some FeatureTests functions depend on server to validate their result.
+    # As CapabilitiesModel is a Singleton class it will get the first result
+    # from FeatureTests which may be wrong when using the Model instance
+    # directly - the case of this test_model.py
+    # So clean Singleton instances to make sure to get the right result when
+    # running the following tests.
+    Singleton._instances = {}
+
+
+def tearDownModule():
+    shutil.rmtree(TMP_DIR)
+
+
+class ModelTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp_store = '/tmp/kimchi-store-test'
+
+    def tearDown(self):
+        # FIXME: Tests using 'test:///default' URI should be moved to
+        # test_rest or test_mockmodel to avoid overriding problems
+        # LibvirtConnection._connections['test:///default'] = {}
+
+        os.unlink(self.tmp_store)
+
+    def test_repository_create(self):
+        inst = model.Model(objstore_loc=self.tmp_store)
+
+        yum_repos = [{'repo_id': 'fedora-fake',
+                      'baseurl': 'http://www.fedora.org'},
+                     {'repo_id': 'fedora-updates-fake',
+                      'config':
+                      {'mirrorlist': 'http://www.fedoraproject.org'}}]
+
+        deb_repos = [{'baseurl': 'http://archive.ubuntu.com/ubuntu/',
+                      'config': {'dist': 'quantal'}},
+                     {'baseurl': 'http://archive.ubuntu.com/ubuntu/',
+                      'config': {'dist': 'quantal', 'comps': ['main']}}]
+
+        yum_invalid_repos = []
+        deb_invalid_repos = []
+
+        for url in invalid_repository_urls:
+            wrong_baseurl = {'repo_id': 'wrong-id', 'baseurl': url}
+            wrong_mirrorlist = {'repo_id': 'wrong-id',
+                                'baseurl': 'www.example.com',
+                                'config': {'mirrorlist': url}}
+            wrong_config_item = {
+                'repo_id': 'wrong-id',
+                'baseurl': 'www.example.com',
+                'config': {
+                    'gpgkey': 'file:///tmp/KEY-fedora-updates-fake-19'}}
+
+            yum_invalid_repos.append(wrong_baseurl)
+            yum_invalid_repos.append(wrong_mirrorlist)
+            yum_invalid_repos.append(wrong_config_item)
+
+            wrong_baseurl['config'] = {'dist': 'tasty'}
+            wrong_config = {'baseurl': deb_repos[0]['baseurl'],
+                            'config': {
+                                'unsupported_item': "a_unsupported_item"}}
+            deb_invalid_repos.append(wrong_baseurl)
+            deb_invalid_repos.append(wrong_config)
+
+        repo_type = inst.capabilities_lookup()['repo_mngt_tool']
+        if repo_type == 'yum':
+            test_repos = yum_repos
+            invalid_repos = yum_invalid_repos
+        elif repo_type == 'deb':
+            test_repos = deb_repos
+            invalid_repos = deb_invalid_repos
+        else:
+            # repository management tool was not recognized by 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'])
+
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..d7cee1c
--- /dev/null
+++ b/src/wok/plugins/gingerbase/tests/test_rest.py
@@ -0,0 +1,182 @@
+# -*- coding: utf-8 -*-
+#
+# Project Kimchi
+#
+# Copyright IBM, Corp. 2013-2015
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+
+import json
+import os
+import time
+import unittest
+from functools import partial
+
+from wok.rollbackcontext import RollbackContext
+
+from wok.plugins.kimchi 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 assertHTTPStatus(self, code, *args):
+        resp = self.request(*args)
+        self.assertEquals(code, resp.status)
+
+    def test_debugreports(self):
+        resp = request(host, ssl_port, '/plugins/kimchi/debugreports')
+        self.assertEquals(200, resp.status)
+
+    def _report_delete(self, name):
+        request(host, ssl_port, '/plugins/kimchi/debugreports/%s' % name, '{}',
+                'DELETE')
+
+    def test_create_debugreport(self):
+        req = json.dumps({'name': 'report1'})
+        with RollbackContext() as rollback:
+            resp = request(host, ssl_port, '/plugins/kimchi/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/kimchi/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/kimchi/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/kimchi/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/kimchi/debugreports/report1')
+            debugreport = json.loads(resp.read())
+            self.assertEquals("report1", debugreport['name'])
+            self.assertEquals(200, resp.status)
+            resp = request(host, ssl_port,
+                           '/plugins/kimchi/debugreports/report1/content')
+            self.assertEquals(200, resp.status)
+            resp = request(host, ssl_port,
+                           '/plugins/kimchi/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/kimchi/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/utils.py b/src/wok/plugins/gingerbase/tests/utils.py
new file mode 100644
index 0000000..8517d7d
--- /dev/null
+++ b/src/wok/plugins/gingerbase/tests/utils.py
@@ -0,0 +1,261 @@
+#
+# 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
+#
+
+import base64
+import cherrypy
+import grp
+import httplib
+import inspect
+import json
+import os
+import socket
+import ssl
+import sys
+import threading
+import time
+import unittest
+from contextlib import closing
+from lxml import etree
+
+import wok.server
+from wok.config import config, PluginPaths
+from wok.auth import User, USER_NAME, USER_GROUPS, USER_ROLES, tabs
+from wok.exception import NotFoundError, OperationFailed
+from wok.utils import wok_log
+
+from wok.plugins.gingerbase import mockmodel
+
+
+_ports = {}
+
+# provide missing unittest decorators and API for python 2.6; these decorators
+# do not actually work, just avoid the syntax failure
+if sys.version_info[:2] == (2, 6):
+    def skipUnless(condition, reason):
+        if not condition:
+            sys.stderr.write('[expected failure] ')
+            raise Exception(reason)
+        return lambda obj: obj
+
+    unittest.skipUnless = skipUnless
+    unittest.expectedFailure = lambda obj: obj
+
+    def assertGreater(self, a, b, msg=None):
+        if not a > b:
+            self.fail('%s not greater than %s' % (repr(a), repr(b)))
+
+    def assertGreaterEqual(self, a, b, msg=None):
+        if not a >= b:
+            self.fail('%s not greater than or equal to %s'
+                      % (repr(a), repr(b)))
+
+    def assertIsInstance(self, obj, cls, msg=None):
+        if not isinstance(obj, cls):
+            self.fail('%s is not an instance of %r' % (repr(obj), cls))
+
+    def assertIn(self, a, b, msg=None):
+        if a not in b:
+            self.fail("%s is not in %b" % (repr(a), repr(b)))
+
+    def assertNotIn(self, a, b, msg=None):
+        if a in b:
+            self.fail("%s is in %b" % (repr(a), repr(b)))
+
+    unittest.TestCase.assertGreaterEqual = assertGreaterEqual
+    unittest.TestCase.assertGreater = assertGreater
+    unittest.TestCase.assertIsInstance = assertIsInstance
+    unittest.TestCase.assertIn = assertIn
+    unittest.TestCase.assertNotIn = assertNotIn
+
+
+def get_free_port(name='http'):
+    global _ports
+    if _ports.get(name) is not None:
+        return _ports[name]
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    with closing(sock):
+        try:
+            sock.bind(("0.0.0.0", 0))
+        except:
+            raise Exception("Could not find a free port")
+        _ports[name] = sock.getsockname()[1]
+        return _ports[name]
+
+
+def run_server(host, port, ssl_port, test_mode, cherrypy_port=None,
+               model=None, environment='development'):
+
+    if cherrypy_port is None:
+        cherrypy_port = get_free_port('cherrypy_port')
+
+    if ssl_port is None:
+        ssl_port = get_free_port('https')
+
+    args = type('_', (object,),
+                {'host': host, 'port': port, 'ssl_port': ssl_port,
+                 'cherrypy_port': cherrypy_port, 'max_body_size': '4*1024',
+                 'ssl_cert': '', 'ssl_key': '',
+                 'test': test_mode, 'access_log': '/dev/null',
+                 'error_log': '/dev/null', 'environment': environment,
+                 'log_level': 'debug'})()
+    if model is not None:
+        setattr(args, 'model', model)
+
+    s = wok.server.Server(args)
+    t = threading.Thread(target=s.start)
+    t.setDaemon(True)
+    t.start()
+    cherrypy.engine.wait(cherrypy.engine.states.STARTED)
+    return s
+
+
+def silence_server():
+    """
+    Silence server status messages on stdout
+    """
+    cherrypy.config.update({"environment": "embedded"})
+
+
+def running_as_root():
+    return os.geteuid() == 0
+
+
+def _request(conn, path, data, method, headers):
+    if headers is None:
+        headers = {'Content-Type': 'application/json',
+                   'Accept': 'application/json'}
+    if 'AUTHORIZATION' not in headers.keys():
+        user, pw = mockmodel.fake_user.items()[0]
+        hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw))
+        headers['AUTHORIZATION'] = hdr
+    conn.request(method, path, data, headers)
+    return conn.getresponse()
+
+
+def request(host, port, path, data=None, method='GET', headers=None):
+    # verify if HTTPSConnection has context parameter
+    if "context" in inspect.getargspec(httplib.HTTPSConnection.__init__).args:
+        context = ssl._create_unverified_context()
+        conn = httplib.HTTPSConnection(host, port, context=context)
+    else:
+        conn = httplib.HTTPSConnection(host, port)
+
+    return _request(conn, path, data, method, headers)
+
+
+def get_remote_iso_path():
+    """
+    Get a remote iso with the right arch from the distro files shipped
+    with kimchi.
+    """
+    host_arch = os.uname()[4]
+    remote_path = ''
+    with open(os.path.join(PluginPaths('kimchi').conf_dir, 'distros.d', 'fedora.json')) \
+            as fedora_isos:
+        # Get a list of dicts
+        json_isos_list = json.load(fedora_isos)
+        for iso in json_isos_list:
+            if (iso.get('os_arch')) == host_arch:
+                remote_path = iso.get('path')
+                break
+
+    return remote_path
+
+
+class FakeUser(User):
+    auth_type = "fake"
+    sudo = True
+
+    def __init__(self, username):
+        self.user = {}
+        self.user[USER_NAME] = username
+        self.user[USER_GROUPS] = None
+        self.user[USER_ROLES] = dict.fromkeys(tabs, 'user')
+
+    def get_groups(self):
+        return sorted([group.gr_name for group in grp.getgrall()])[0:3]
+
+    def get_roles(self):
+        if self.sudo:
+            self.user[USER_ROLES] = dict.fromkeys(tabs, 'admin')
+        return self.user[USER_ROLES]
+
+    def get_user(self):
+        return self.user
+
+    @staticmethod
+    def authenticate(username, password, service="passwd"):
+        try:
+            return mockmodel.fake_user[username] == password
+        except KeyError, e:
+            raise OperationFailed("GGBAUTH0001E", {'username': 'username',
+                                                   'code': e.message})
+
+
+def patch_auth(sudo=True):
+    """
+    Override the authenticate function with a simple test against an
+    internal dict of users and passwords.
+    """
+    config.set("authentication", "method", "fake")
+    FakeUser.sudo = sudo
+
+
+def normalize_xml(xml_str):
+    return etree.tostring(etree.fromstring(xml_str,
+                          etree.XMLParser(remove_blank_text=True)))
+
+
+def wait_task(task_lookup, taskid, timeout=10):
+    for i in range(0, timeout):
+        task_info = task_lookup(taskid)
+        if task_info['status'] == "running":
+            wok_log.info("Waiting task %s, message: %s",
+                            taskid, task_info['message'])
+            time.sleep(1)
+        else:
+            return
+    wok_log.error("Timeout while process long-run task, "
+                     "try to increase timeout value.")
+
+
+# The action functions in model backend raise NotFoundError exception if the
+# element is not found. But in some tests, these functions are called after
+# the element has been deleted if test finishes correctly, then NofFoundError
+# exception is raised and rollback breaks. To avoid it, this wrapper ignores
+# the NotFoundError.
+def rollback_wrapper(func, resource, *args):
+    try:
+        func(resource, *args)
+    except NotFoundError:
+        # VM has been deleted already
+        return
+
+
+# This function is used to test storage volume upload.
+# If we use self.request, we may encode multipart formdata by ourselves
+# requests lib take care of encode part, so use this lib instead
+def fake_auth_header():
+    headers = {'Accept': 'application/json'}
+    user, pw = mockmodel.fake_user.items()[0]
+    hdr = "Basic " + base64.b64encode("%s:%s" % (user, pw))
+    headers['AUTHORIZATION'] = hdr
+    return headers
+
-- 
2.1.0




More information about the Kimchi-devel mailing list