[Kimchi-devel] [PATCH 09/15] V3 Ginger Base : base plugin tests files
Daniel Henrique Barboza
dhbarboza82 at gmail.com
Wed Oct 7 13:03:01 UTC 2015
Whitespace errors when applying the patch:
$ git am \[Kimchi-devel\]\ \[PATCH\ 09_15\]\ V3\ Ginger\ Base\ \:\ base\
plugin\ tests\ files\ -\ chandra at linux.vnet.ibm.com\ -\ 2015-10-07\ 0744.eml
Applying: V3 Ginger Base : base plugin tests files
/home/danielhb/kimchi/wok/kimchi/.git/rebase-apply/patch:451: new blank
line at EOF.
+
/home/danielhb/kimchi/wok/kimchi/.git/rebase-apply/patch:737: new blank
line at EOF.
+
/home/danielhb/kimchi/wok/kimchi/.git/rebase-apply/patch:1192: new blank
line at EOF.
+
warning: 3 lines add whitespace errors.
On 10/07/2015 07:44 AM, chandra at linux.vnet.ibm.com wrote:
> 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
> +
More information about the Kimchi-devel
mailing list