almalinux/lorax
12
0
Fork 0
Code Issues Pull Requests Releases Activity

Add the tests from lorax-composer branch

Browse Source 
These currently fail for several reasons and will be adjusted as the
code it ported to this branch.
This commit is contained in:
Brian C. Lane 2018-04-30 11:34:48 -07:00
parent b3d6dab83b
commit b4096ccb62
20 changed files with 1943 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
tests/pylorax/blueprints/atlas.toml Normal file
View File

@ -0,0 +1,10 @@
name = "atlas"
description = "Automatically Tuned Linear Algebra Software"
[[modules]]
name = "atlas"
version = "3.10.*"
[[modules]]
name = "numpy"
version = "1.7.*"

82
tests/pylorax/blueprints/development.toml Normal file
View File

@ -0,0 +1,82 @@
name = "development"
description = "A general purpose development image"
[[packages]]
name = "cmake"
version = "*"
[[packages]]
name = "curl"
version = "*"
[[packages]]
name = "file"
version = "*"
[[packages]]
name = "gcc"
version = "*"
[[packages]]
name = "gcc-c++"
version = "*"
[[packages]]
name = "gdb"
version = "*"
[[packages]]
name = "git"
version = "*"
[[packages]]
name = "glibc-devel"
version = "*"
[[packages]]
name = "gnupg2"
version = "*"
[[packages]]
name = "libcurl-devel"
version = "*"
[[packages]]
name = "make"
version = "*"
[[packages]]
name = "openssl-devel"
version = "*"
[[packages]]
name = "openssl-devel"
version = "*"
[[packages]]
name = "sqlite"
version = "*"
[[packages]]
name = "sqlite-devel"
version = "*"
[[packages]]
name = "sudo"
version = "*"
[[packages]]
name = "tar"
version = "*"
[[packages]]
name = "xz"
version = "*"
[[packages]]
name = "xz-devel"
version = "*"
[[packages]]
name = "zlib-devel"
version = "*"

14
tests/pylorax/blueprints/glusterfs.toml Normal file
View File

@ -0,0 +1,14 @@
name = "glusterfs"
description = "An example GlusterFS server with samba"
[[modules]]
name = "glusterfs"
version = "3.7.*"
[[modules]]
name = "glusterfs-cli"
version = "3.7.*"
[[packages]]
name = "samba"
version = "4.2.*"

35
tests/pylorax/blueprints/http-server.toml Normal file
View File

@ -0,0 +1,35 @@
name = "http-server"
description = "An example http server with PHP and MySQL support."
version = "0.0.1"
[[modules]]
name = "httpd"
version = "2.4.*"
[[modules]]
name = "mod_auth_kerb"
version = "5.4"
[[modules]]
name = "mod_ssl"
version = "2.4.*"
[[modules]]
name = "php"
version = "5.4.*"
[[modules]]
name = "php-mysql"
version = "5.4.*"
[[packages]]
name = "tmux"
version = "2.2"
[[packages]]
name = "openssh-server"
version = "6.6.*"
[[packages]]
name = "rsync"
version = "3.0.*"

14
tests/pylorax/blueprints/jboss.toml Normal file
View File

@ -0,0 +1,14 @@
name = "jboss"
description = "An example jboss server"
[[modules]]
name = "jboss-servlet-3.0-api"
version = "1.0.*"
[[modules]]
name = "jboss-interceptors-1.1-api"
version = "1.0.*"
[[modules]]
name = "java-1.8.0-openjdk"
version = "1.8.0.*"

26
tests/pylorax/blueprints/kubernetes.toml Normal file
View File

@ -0,0 +1,26 @@
name = "kubernetes"
description = "An example kubernetes master"
[[modules]]
name = "kubernetes"
version = "1.2.*"
[[modules]]
name = "docker"
version = "1.10.*"
[[modules]]
name = "docker-lvm-plugin"
version = "1.10.*"
[[modules]]
name = "etcd"
version = "2.3.*"
[[modules]]
name = "flannel"
version = "0.5.*"
[[packages]]
name = "oci-systemd-hook"
version = "0.1.*"

1
tests/pylorax/results/full-recipe.dict Normal file
View File

@ -0,0 +1 @@
{'description': u'An example http server with PHP and MySQL support.', 'packages': [{'version': u'6.6.*', 'name': u'openssh-server'}, {'version': u'3.0.*', 'name': u'rsync'}, {'version': u'2.2', 'name': u'tmux'}], 'modules': [{'version': u'2.4.*', 'name': u'httpd'}, {'version': u'5.4', 'name': u'mod_auth_kerb'}, {'version': u'2.4.*', 'name': u'mod_ssl'}, {'version': u'5.4.*', 'name': u'php'}, {'version': u'5.4.*', 'name': u'php-mysql'}], 'version': u'0.0.1', 'name': u'http-server'}

35
tests/pylorax/results/full-recipe.toml Normal file
View File

@ -0,0 +1,35 @@
name = "http-server"
description = "An example http server with PHP and MySQL support."
version = "0.0.1"
[[modules]]
name = "httpd"
version = "2.4.*"
[[modules]]
name = "mod_auth_kerb"
version = "5.4"
[[modules]]
name = "mod_ssl"
version = "2.4.*"
[[modules]]
name = "php"
version = "5.4.*"
[[modules]]
name = "php-mysql"
version = "5.4.*"
[[packages]]
name = "tmux"
version = "2.2"
[[packages]]
name = "openssh-server"
version = "6.6.*"
[[packages]]
name = "rsync"
version = "3.0.*"

1
tests/pylorax/results/minimal.dict Normal file
View File

@ -0,0 +1 @@
{'description': u'An example http server with PHP and MySQL support.', 'packages': [], 'modules': [], 'version': u'0.0.1', 'name': u'http-server'}

3
tests/pylorax/results/minimal.toml Normal file
View File

@ -0,0 +1,3 @@
name = "http-server"
description = "An example http server with PHP and MySQL support."
version = "0.0.1"

1
tests/pylorax/results/modules-only.dict Normal file
View File

@ -0,0 +1 @@
{'description': u'An example http server with PHP and MySQL support.', 'packages': [], 'modules': [{'version': u'2.4.*', 'name': u'httpd'}, {'version': u'5.4', 'name': u'mod_auth_kerb'}, {'version': u'2.4.*', 'name': u'mod_ssl'}, {'version': u'5.4.*', 'name': u'php'}, {'version': u'5.4.*', 'name': u'php-mysql'}], 'version': u'0.0.1', 'name': u'http-server'}

23
tests/pylorax/results/modules-only.toml Normal file
View File

@ -0,0 +1,23 @@
name = "http-server"
description = "An example http server with PHP and MySQL support."
version = "0.0.1"
[[modules]]
name = "httpd"
version = "2.4.*"
[[modules]]
name = "mod_auth_kerb"
version = "5.4"
[[modules]]
name = "mod_ssl"
version = "2.4.*"
[[modules]]
name = "php"
version = "5.4.*"
[[modules]]
name = "php-mysql"
version = "5.4.*"

1
tests/pylorax/results/packages-only.dict Normal file
View File

@ -0,0 +1 @@
{'description': u'An example http server with PHP and MySQL support.', 'packages': [{'version': u'6.6.*', 'name': u'openssh-server'}, {'version': u'3.0.*', 'name': u'rsync'}, {'version': u'2.2', 'name': u'tmux'}], 'modules': [], 'version': u'0.0.1', 'name': u'http-server'}

15
tests/pylorax/results/packages-only.toml Normal file
View File

@ -0,0 +1,15 @@
name = "http-server"
description = "An example http server with PHP and MySQL support."
version = "0.0.1"
[[packages]]
name = "tmux"
version = "2.2"
[[packages]]
name = "openssh-server"
version = "6.6.*"
[[packages]]
name = "rsync"
version = "3.0.*"

118
tests/pylorax/test_crossdomain.py Normal file
View File

@ -0,0 +1,118 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import unittest
from flask import Flask
from datetime import timedelta
from pylorax.api.crossdomain import crossdomain
server = Flask(__name__)
@server.route('/01')
@crossdomain(origin='*', methods=['GET'])
def hello_world_01():
return 'Hello, World!'
@server.route('/02')
@crossdomain(origin='*', headers=['TESTING'])
def hello_world_02():
return 'Hello, World!'
@server.route('/03')
@crossdomain(origin='*', max_age=timedelta(days=7))
def hello_world_03():
return 'Hello, World!'
@server.route('/04')
@crossdomain(origin='*', attach_to_all=False)
def hello_world_04():
return 'Hello, World!'
@server.route('/05')
@crossdomain(origin='*', automatic_options=False)
def hello_world_05():
return 'Hello, World!'
@server.route('/06')
@crossdomain(origin=['https://redhat.com', 'http://weldr.io'])
def hello_world_06():
return 'Hello, World!'
class CrossdomainTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.server = server.test_client()
def test_01_with_methods_specified(self):
# first send a preflight request to check what methods are allowed
response = self.server.options("/01")
self.assertEqual(200, response.status_code)
self.assertIn('GET', response.headers['Access-Control-Allow-Methods'])
# then try to issue a POST request which isn't allowed
response = self.server.post("/01")
self.assertEqual(405, response.status_code)
def test_02_with_headers_specified(self):
response = self.server.get("/02")
self.assertEqual(200, response.status_code)
self.assertEqual('Hello, World!', response.data)
self.assertEqual('TESTING', response.headers['Access-Control-Allow-Headers'])
def test_03_with_max_age_as_timedelta(self):
response = self.server.get("/03")
self.assertEqual(200, response.status_code)
self.assertEqual('Hello, World!', response.data)
expected_max_age = int(timedelta(days=7).total_seconds())
actual_max_age = int(response.headers['Access-Control-Max-Age'])
self.assertEqual(expected_max_age, actual_max_age)
def test_04_attach_to_all_false(self):
response = self.server.get("/04")
self.assertEqual(200, response.status_code)
self.assertEqual('Hello, World!', response.data)
# when attach_to_all is False the decorator will not assign
# the Access-Control-* headers to the response
for header, _ in response.headers:
self.assertFalse(header.startswith('Access-Control-'))
def test_05_options_request(self):
response = self.server.options("/05")
self.assertEqual(200, response.status_code)
self.assertEqual('Hello, World!', response.data)
self.assertEqual(response.headers['Access-Control-Allow-Methods'], 'HEAD, OPTIONS, GET')
def test_06_with_origin_as_list(self):
response = self.server.get("/06")
self.assertEqual(200, response.status_code)
self.assertEqual('Hello, World!', response.data)
for header, value in response.headers:
if header == 'Access-Control-Allow-Origin':
self.assertIn(value, ['https://redhat.com', 'http://weldr.io'])

233
tests/pylorax/test_projects.py Normal file
View File

@ -0,0 +1,233 @@
#
# Copyright (C) 2017 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os
import mock
import time
import shutil
import tempfile
import unittest
from yum.Errors import YumBaseError
from pylorax.api.config import configure, make_yum_dirs
from pylorax.api.projects import api_time, api_changelog, yaps_to_project, yaps_to_project_info
from pylorax.api.projects import tm_to_dep, yaps_to_module, projects_list, projects_info, projects_depsolve
from pylorax.api.projects import modules_list, modules_info, ProjectsError, dep_evra, dep_nevra
from pylorax.api.yumbase import get_base_object
class Yaps(object):
"""Test class for yaps tests"""
name = "name"
summary = "summary"
description = "description"
url = "url"
epoch = 1
release = "release"
arch = "arch"
buildtime = 499222800
license = "license"
version = "version"
def returnChangelog(self):
return [[0,1,"Heavy!"]]
class TM(object):
"""Test class for tm test"""
name = "name"
epoch = 1
version = "version"
release = "release"
arch = "arch"
class ProjectsTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.tmp_dir = tempfile.mkdtemp(prefix="lorax.test.repo.")
self.config = configure(root_dir=self.tmp_dir, test_config=True)
make_yum_dirs(self.config)
self.yb = get_base_object(self.config)
os.environ["TZ"] = "UTC"
time.tzset()
@classmethod
def tearDownClass(self):
shutil.rmtree(self.tmp_dir)
def test_api_time(self):
self.assertEqual(api_time(499222800), "1985-10-27T01:00:00")
def test_api_changelog(self):
self.assertEqual(api_changelog([[0,1,"Heavy!"], [0, 1, "Light!"]]), "Heavy!")
def test_api_changelog_empty_list(self):
self.assertEqual(api_changelog([]), '')
def test_api_changelog_missing_text_entry(self):
self.assertEqual(api_changelog([('now', 'atodorov')]), '')
def test_yaps_to_project(self):
result = {"name":"name",
"summary":"summary",
"description":"description",
"homepage":"url",
"upstream_vcs":"UPSTREAM_VCS"}
y = Yaps()
self.assertEqual(yaps_to_project(y), result)
def test_yaps_to_project_info(self):
build = {"epoch":1,
"release":"release",
"arch":"arch",
"build_time":"1985-10-27T01:00:00",
"changelog":"Heavy!",
"build_config_ref": "BUILD_CONFIG_REF",
"build_env_ref": "BUILD_ENV_REF",
"metadata": {},
"source": {"license":"license",
"version":"version",
"source_ref": "SOURCE_REF",
"metadata": {}}}
result = {"name":"name",
"summary":"summary",
"description":"description",
"homepage":"url",
"upstream_vcs":"UPSTREAM_VCS",
"builds": [build]}
y = Yaps()
self.assertEqual(yaps_to_project_info(y), result)
def test_tm_to_dep(self):
result = {"name":"name",
"epoch":1,
"version":"version",
"release":"release",
"arch":"arch"}
tm = TM()
self.assertEqual(tm_to_dep(tm), result)
def test_yaps_to_module(self):
result = {"name":"name",
"group_type":"rpm"}
y = Yaps()
self.assertEqual(yaps_to_module(y), result)
def test_dep_evra(self):
dep = {"arch": "noarch",
"epoch": 0,
"name": "basesystem",
"release": "7.el7",
"version": "10.0"}
self.assertEqual(dep_evra(dep), "10.0-7.el7.noarch")
def test_dep_evra_with_epoch_not_zero(self):
dep = {"arch": "x86_64",
"epoch": 2,
"name": "tog-pegasus-libs",
"release": "3.el7",
"version": "2.14.1"}
self.assertEqual(dep_evra(dep), "2:2.14.1-3.el7.x86_64")
def test_dep_nevra(self):
dep = {"arch": "noarch",
"epoch": 0,
"name": "basesystem",
"release": "7.el7",
"version": "10.0"}
self.assertEqual(dep_nevra(dep), "basesystem-10.0-7.el7.noarch")
def test_projects_list(self):
projects = projects_list(self.yb)
self.assertEqual(len(projects) > 10, True)
def test_projects_list_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
projects_list(self.yb)
def test_projects_info(self):
projects = projects_info(self.yb, ["bash"])
self.assertEqual(projects[0]["name"], "bash")
self.assertEqual(projects[0]["builds"][0]["source"]["license"], "GPLv3+")
def test_projects_info_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
projects_info(self.yb, ["bash"])
def test_projects_depsolve(self):
deps = projects_depsolve(self.yb, ["bash"])
self.assertEqual(deps[0]["name"], "basesystem")
def test_projects_depsolve_fail(self):
with self.assertRaises(ProjectsError):
projects_depsolve(self.yb, ["nada-package"])
def test_modules_list(self):
modules = modules_list(self.yb, None)
self.assertEqual(len(modules) > 10, True)
self.assertEqual(modules[0]["group_type"], "rpm")
modules = modules_list(self.yb, ["g*"])
self.assertEqual(modules[0]["name"].startswith("g"), True)
def test_modules_list_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
modules_list(self.yb, None)
def test_modules_info(self):
modules = modules_info(self.yb, ["bash"])
print(modules)
self.assertEqual(modules[0]["name"], "bash")
self.assertEqual(modules[0]["dependencies"][0]["name"], "basesystem")
def test_modules_info_yum_raises_exception(self):
with self.assertRaises(ProjectsError):
with mock.patch.object(self.yb, 'doPackageLists', side_effect=YumBaseError('TESTING')):
modules_info(self.yb, ["bash"])
class ConfigureTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.tmp_dir = tempfile.mkdtemp(prefix="lorax.test.configure.")
self.conf_file = os.path.join(self.tmp_dir, 'test.conf')
open(self.conf_file, 'w').write("[composer]\ncache_dir = /tmp/cache-test")
@classmethod
def tearDownClass(self):
shutil.rmtree(self.tmp_dir)
def test_configure_reads_existing_file(self):
config = configure(conf_file=self.conf_file)
self.assertEqual(config.get('composer', 'cache_dir'), '/tmp/cache-test')
def test_configure_reads_non_existing_file(self):
config = configure(conf_file=self.conf_file + '.non-existing')
self.assertEqual(config.get('composer', 'cache_dir'), '/var/tmp/composer/cache')

343
tests/pylorax/test_recipes.py Normal file
View File

@ -0,0 +1,343 @@
#
# Copyright (C) 2017 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os
import mock
from pytoml import TomlError
import shutil
import tempfile
import unittest
import pylorax.api.recipes as recipes
from pylorax.sysutils import joinpaths
class BasicRecipeTest(unittest.TestCase):
@classmethod
def setUpClass(self):
# Input toml is in .toml and python dict string is in .dict
input_recipes = [("full-recipe.toml", "full-recipe.dict"),
("minimal.toml", "minimal.dict"),
("modules-only.toml", "modules-only.dict"),
("packages-only.toml", "packages-only.dict")]
results_path = "./tests/pylorax/results/"
self.input_toml = []
for (recipe_toml, recipe_dict) in input_recipes:
with open(joinpaths(results_path, recipe_toml)) as f_toml:
with open(joinpaths(results_path, recipe_dict)) as f_dict:
# XXX Warning, can run arbitrary code
result_dict = eval(f_dict.read())
self.input_toml.append((f_toml.read(), result_dict))
self.old_modules = [recipes.RecipeModule("toml", "2.1"),
recipes.RecipeModule("bash", "4.*"),
recipes.RecipeModule("httpd", "3.7.*")]
self.old_packages = [recipes.RecipePackage("python", "2.7.*"),
recipes.RecipePackage("parted", "3.2")]
self.new_modules = [recipes.RecipeModule("toml", "2.1"),
recipes.RecipeModule("httpd", "3.8.*"),
recipes.RecipeModule("openssh", "2.8.1")]
self.new_packages = [recipes.RecipePackage("python", "2.7.*"),
recipes.RecipePackage("parted", "3.2"),
recipes.RecipePackage("git", "2.13.*")]
self.modules_result = [{"new": {"Modules": {"version": "2.8.1", "name": "openssh"}},
"old": None},
{"new": None,
"old": {"Modules": {"name": "bash", "version": "4.*"}}},
{"new": {"Modules": {"version": "3.8.*", "name": "httpd"}},
"old": {"Modules": {"version": "3.7.*", "name": "httpd"}}}]
self.packages_result = [{"new": {"Packages": {"name": "git", "version": "2.13.*"}}, "old": None}]
@classmethod
def tearDownClass(self):
pass
def toml_to_recipe_test(self):
"""Test converting the TOML string to a Recipe object"""
for (toml_str, recipe_dict) in self.input_toml:
result = recipes.recipe_from_toml(toml_str)
self.assertEqual(result, recipe_dict)
def toml_to_recipe_fail_test(self):
"""Test trying to convert a non-TOML string to a Recipe"""
with self.assertRaises(TomlError):
recipes.recipe_from_toml("This is not a TOML string\n")
with self.assertRaises(recipes.RecipeError):
recipes.recipe_from_toml('name = "a failed toml string"\n')
def recipe_to_toml_test(self):
"""Test converting a Recipe object to a TOML string"""
# In order to avoid problems from matching strings we convert to TOML and
# then back so compare the Recipes.
for (toml_str, _recipe_dict) in self.input_toml:
# This is tested in toml_to_recipe
recipe_1 = recipes.recipe_from_toml(toml_str)
# Convert the Recipe to TOML and then back to a Recipe
toml_2 = recipe_1.toml()
recipe_2 = recipes.recipe_from_toml(toml_2)
self.assertEqual(recipe_1, recipe_2)
def recipe_bump_version_test(self):
"""Test the Recipe's version bump function"""
# Neither have a version
recipe = recipes.Recipe("test-recipe", "A recipe used for testing", None, None, None)
new_version = recipe.bump_version(None)
self.assertEqual(new_version, "0.0.1")
# Original has a version, new does not
recipe = recipes.Recipe("test-recipe", "A recipe used for testing", None, None, None)
new_version = recipe.bump_version("0.0.1")
self.assertEqual(new_version, "0.0.2")
# Original has no version, new does
recipe = recipes.Recipe("test-recipe", "A recipe used for testing", "0.1.0", None, None)
new_version = recipe.bump_version(None)
self.assertEqual(new_version, "0.1.0")
# New and Original are the same
recipe = recipes.Recipe("test-recipe", "A recipe used for testing", "0.0.1", None, None)
new_version = recipe.bump_version("0.0.1")
self.assertEqual(new_version, "0.0.2")
# New is different from Original
recipe = recipes.Recipe("test-recipe", "A recipe used for testing", "0.1.1", None, None)
new_version = recipe.bump_version("0.0.1")
self.assertEqual(new_version, "0.1.1")
def find_name_test(self):
"""Test the find_name function"""
test_list = [{"name":"dog"}, {"name":"cat"}, {"name":"squirrel"}]
self.assertEqual(recipes.find_name("dog", test_list), {"name":"dog"})
self.assertEqual(recipes.find_name("cat", test_list), {"name":"cat"})
self.assertEqual(recipes.find_name("squirrel", test_list), {"name":"squirrel"})
self.assertIsNone(recipes.find_name("alien", test_list))
def diff_items_test(self):
"""Test the diff_items function"""
self.assertEqual(recipes.diff_items("Modules", self.old_modules, self.new_modules), self.modules_result)
self.assertEqual(recipes.diff_items("Packages", self.old_packages, self.new_packages), self.packages_result)
def recipe_diff_test(self):
"""Test the recipe_diff function"""
old_recipe = recipes.Recipe("test-recipe", "A recipe used for testing", "0.1.1", self.old_modules, self.old_packages)
new_recipe = recipes.Recipe("test-recipe", "A recipe used for testing", "0.3.1", self.new_modules, self.new_packages)
result = [{'new': {'Version': '0.3.1'}, 'old': {'Version': '0.1.1'}},
{'new': {'Module': {'name': 'openssh', 'version': '2.8.1'}}, 'old': None},
{'new': None, 'old': {'Module': {'name': 'bash', 'version': '4.*'}}},
{'new': {'Module': {'name': 'httpd', 'version': '3.8.*'}},
'old': {'Module': {'name': 'httpd', 'version': '3.7.*'}}},
{'new': {'Package': {'name': 'git', 'version': '2.13.*'}}, 'old': None}]
self.assertEqual(recipes.recipe_diff(old_recipe, new_recipe), result)
class GitRecipesTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.repo_dir = tempfile.mkdtemp(prefix="lorax.test.repo.")
self.repo = recipes.open_or_create_repo(self.repo_dir)
self.results_path = "./tests/pylorax/results/"
self.examples_path = "./tests/pylorax/blueprints/"
self.new_recipe = os.path.join(self.examples_path, 'python-testing.toml')
@classmethod
def tearDownClass(self):
if self.repo is not None:
del self.repo
shutil.rmtree(self.repo_dir)
def tearDown(self):
if os.path.exists(self.new_recipe):
os.remove(self.new_recipe)
def _create_another_recipe(self):
open(self.new_recipe, 'w').write("""name = "python-testing"
description = "A recipe used during testing."
version = "0.0.1"
[[packages]]
name = "python"
version = "2.7.*"
""")
def test_01_repo_creation(self):
"""Test that creating the repository succeeded"""
self.assertNotEqual(self.repo, None)
def test_02_commit_recipe(self):
"""Test committing a Recipe object"""
recipe = recipes.Recipe("test-recipe", "A recipe used for testing", None, None, None)
oid = recipes.commit_recipe(self.repo, "master", recipe)
self.assertNotEqual(oid, None)
def test_03_list_recipe(self):
"""Test listing recipe commits"""
commits = recipes.list_commits(self.repo, "master", "test-recipe.toml")
self.assertEqual(len(commits), 1, "Wrong number of commits.")
self.assertEqual(commits[0].message, "Recipe test-recipe, version 0.0.1 saved.")
self.assertNotEqual(commits[0].timestamp, None, "Timestamp is None")
self.assertEqual(len(commits[0].commit), 40, "Commit hash isn't 40 characters")
self.assertEqual(commits[0].revision, None, "revision is not None")
def test_03_list_commits_commit_time_val_error(self):
"""Test listing recipe commits which raise CommitTimeValError"""
with mock.patch('pylorax.api.recipes.GLib.DateTime.to_timeval', return_value=False):
commits = recipes.list_commits(self.repo, "master", "test-recipe.toml")
self.assertEqual(len(commits), 0, "Wrong number of commits.")
def test_04_commit_recipe_file(self):
"""Test committing a TOML file"""
recipe_path = joinpaths(self.results_path, "full-recipe.toml")
oid = recipes.commit_recipe_file(self.repo, "master", recipe_path)
self.assertNotEqual(oid, None)
commits = recipes.list_commits(self.repo, "master", "http-server.toml")
self.assertEqual(len(commits), 1, "Wrong number of commits: %s" % commits)
def test_04_commit_recipe_file_handles_internal_ioerror(self):
"""Test committing a TOML raises RecipeFileError on internal IOError"""
recipe_path = joinpaths(self.results_path, "non-existing-file.toml")
with self.assertRaises(recipes.RecipeFileError):
recipes.commit_recipe_file(self.repo, "master", recipe_path)
def test_05_commit_toml_dir(self):
"""Test committing a directory of TOML files"""
# first verify that the newly created file isn't present
old_commits = recipes.list_commits(self.repo, "master", "python-testing.toml")
self.assertEqual(len(old_commits), 0, "Wrong number of commits: %s" % old_commits)
# then create it and commit the entire directory
self._create_another_recipe()
recipes.commit_recipe_directory(self.repo, "master", self.examples_path)
# verify that the newly created file is already in the repository
new_commits = recipes.list_commits(self.repo, "master", "python-testing.toml")
self.assertEqual(len(new_commits), 1, "Wrong number of commits: %s" % new_commits)
# again make sure new_commits != old_commits
self.assertGreater(len(new_commits), len(old_commits),
"New commits shoud differ from old commits")
def test_05_commit_recipe_directory_handling_internal_exceptions(self):
"""Test committing a directory of TOML files while handling internal exceptions"""
# first verify that the newly created file isn't present
old_commits = recipes.list_commits(self.repo, "master", "python-testing.toml")
self.assertEqual(len(old_commits), 0, "Wrong number of commits: %s" % old_commits)
# then create it and commit the entire directory
self._create_another_recipe()
# try to commit while raising RecipeFileError
with mock.patch('pylorax.api.recipes.commit_recipe_file', side_effect=recipes.RecipeFileError('TESTING')):
recipes.commit_recipe_directory(self.repo, "master", self.examples_path)
# try to commit while raising TomlError
with mock.patch('pylorax.api.recipes.commit_recipe_file', side_effect=TomlError('TESTING', 0, 0, '__test__')):
recipes.commit_recipe_directory(self.repo, "master", self.examples_path)
# verify again that the newly created file isn't present b/c we raised an exception
new_commits = recipes.list_commits(self.repo, "master", "python-testing.toml")
self.assertEqual(len(new_commits), 0, "Wrong number of commits: %s" % new_commits)
def test_06_read_recipe(self):
"""Test reading a recipe from a commit"""
commits = recipes.list_commits(self.repo, "master", "http-server.toml")
self.assertEqual(len(commits), 1, "Wrong number of commits: %s" % commits)
recipe = recipes.read_recipe_commit(self.repo, "master", "http-server")
self.assertNotEqual(recipe, None)
self.assertEqual(recipe["name"], "http-server")
# Read by commit id
recipe = recipes.read_recipe_commit(self.repo, "master", "http-server", commits[0].commit)
self.assertNotEqual(recipe, None)
self.assertEqual(recipe["name"], "http-server")
# Read the recipe and its commit id
(commit_id, recipe) = recipes.read_recipe_and_id(self.repo, "master", "http-server", commits[0].commit)
self.assertEqual(commit_id, commits[0].commit)
def test_07_tag_commit(self):
"""Test tagging the most recent commit of a recipe"""
result = recipes.tag_file_commit(self.repo, "master", "not-a-file")
self.assertEqual(result, None)
result = recipes.tag_recipe_commit(self.repo, "master", "http-server")
self.assertNotEqual(result, None)
commits = recipes.list_commits(self.repo, "master", "http-server.toml")
self.assertEqual(len(commits), 1, "Wrong number of commits: %s" % commits)
self.assertEqual(commits[0].revision, 1)
def test_08_delete_recipe(self):
"""Test deleting a file from a branch"""
oid = recipes.delete_recipe(self.repo, "master", "http-server")
self.assertNotEqual(oid, None)
master_files = recipes.list_branch_files(self.repo, "master")
self.assertEqual("http-server.toml" in master_files, False)
def test_09_revert_commit(self):
"""Test reverting a file on a branch"""
commits = recipes.list_commits(self.repo, "master", "http-server.toml")
revert_to = commits[0].commit
oid = recipes.revert_recipe(self.repo, "master", "http-server", revert_to)
self.assertNotEqual(oid, None)
commits = recipes.list_commits(self.repo, "master", "http-server.toml")
self.assertEqual(len(commits), 2, "Wrong number of commits: %s" % commits)
self.assertEqual(commits[0].message, "http-server.toml reverted to commit %s" % revert_to)
def test_10_tag_new_commit(self):
"""Test tagging a newer commit of a recipe"""
recipe = recipes.read_recipe_commit(self.repo, "master", "http-server")
recipe["description"] = "A modified description"
oid = recipes.commit_recipe(self.repo, "master", recipe)
self.assertNotEqual(oid, None)
# Tag the new commit
result = recipes.tag_recipe_commit(self.repo, "master", "http-server")
self.assertNotEqual(result, None)
commits = recipes.list_commits(self.repo, "master", "http-server.toml")
self.assertEqual(len(commits), 3, "Wrong number of commits: %s" % commits)
self.assertEqual(commits[0].revision, 2)
class ExistingGitRepoRecipesTest(GitRecipesTest):
@classmethod
def setUpClass(self):
# will initialize the git repository in the parent class
super(ExistingGitRepoRecipesTest, self).setUpClass()
# reopen the repository again so that tests are executed
# against the existing repo one more time.
self.repo = recipes.open_or_create_repo(self.repo_dir)
class GetRevisionFromTagTests(unittest.TestCase):
def test_01_valid_tag(self):
revision = recipes.get_revision_from_tag('branch/filename/r123')
self.assertEqual(123, revision)
def test_02_invalid_tag_not_a_number(self):
revision = recipes.get_revision_from_tag('branch/filename/rABC')
self.assertIsNone(revision)
def test_02_invalid_tag_missing_revision_string(self):
revision = recipes.get_revision_from_tag('branch/filename/mybranch')
self.assertIsNone(revision)

810
tests/pylorax/test_server.py Normal file
View File

@ -0,0 +1,810 @@
#
# Copyright (C) 2017 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os
from glob import glob
import shutil
import tempfile
import time
from threading import Lock
import unittest
from flask import json
import pytoml as toml
from pylorax.api.config import configure, make_yum_dirs, make_queue_dirs
from pylorax.api.queue import start_queue_monitor
from pylorax.api.recipes import open_or_create_repo, commit_recipe_directory
from pylorax.api.server import server, GitLock, YumLock
from pylorax.api.yumbase import get_base_object
from pylorax.sysutils import joinpaths
class ServerTestCase(unittest.TestCase):
@classmethod
def setUpClass(self):
repo_dir = tempfile.mkdtemp(prefix="lorax.test.repo.")
server.config["REPO_DIR"] = repo_dir
repo = open_or_create_repo(server.config["REPO_DIR"])
server.config["GITLOCK"] = GitLock(repo=repo, lock=Lock(), dir=repo_dir)
server.config["COMPOSER_CFG"] = configure(root_dir=repo_dir, test_config=True)
os.makedirs(joinpaths(server.config["COMPOSER_CFG"].get("composer", "share_dir"), "composer"))
errors = make_queue_dirs(server.config["COMPOSER_CFG"], 0)
if errors:
raise RuntimeError("\n".join(errors))
make_yum_dirs(server.config["COMPOSER_CFG"])
yb = get_base_object(server.config["COMPOSER_CFG"])
server.config["YUMLOCK"] = YumLock(yb=yb, lock=Lock())
server.config['TESTING'] = True
self.server = server.test_client()
self.examples_path = "./tests/pylorax/blueprints/"
# Copy the shared files over to the directory tree we are using
share_path = "./share/composer/"
for f in glob(joinpaths(share_path, "*")):
shutil.copy(f, joinpaths(server.config["COMPOSER_CFG"].get("composer", "share_dir"), "composer"))
# Import the example blueprints
commit_recipe_directory(server.config["GITLOCK"].repo, "master", self.examples_path)
start_queue_monitor(server.config["COMPOSER_CFG"], 0, 0)
@classmethod
def tearDownClass(self):
shutil.rmtree(server.config["REPO_DIR"])
def test_01_status(self):
"""Test the /api/status route"""
status_fields = ["build", "api", "db_version", "schema_version", "db_supported", "backend"]
resp = self.server.get("/api/status")
data = json.loads(resp.data)
# Just make sure the fields are present
self.assertEqual(sorted(data.keys()), sorted(status_fields))
def test_02_blueprints_list(self):
"""Test the /api/v0/blueprints/list route"""
list_dict = {"blueprints":["atlas", "development", "glusterfs", "http-server", "jboss", "kubernetes"],
"limit":20, "offset":0, "total":6}
resp = self.server.get("/api/v0/blueprints/list")
data = json.loads(resp.data)
self.assertEqual(data, list_dict)
def test_03_blueprints_info(self):
"""Test the /api/v0/blueprints/info route"""
info_dict_1 = {"changes":[{"changed":False, "name":"http-server"}],
"errors":[],
"blueprints":[{"description":"An example http server with PHP and MySQL support.",
"modules":[{"name":"httpd", "version":"2.4.*"},
{"name":"mod_auth_kerb", "version":"5.4"},
{"name":"mod_ssl", "version":"2.4.*"},
{"name":"php", "version":"5.4.*"},
{"name": "php-mysql", "version":"5.4.*"}],
"name":"http-server",
"packages": [{"name":"openssh-server", "version": "6.6.*"},
{"name": "rsync", "version": "3.0.*"},
{"name": "tmux", "version": "2.2"}],
"version": "0.0.1"}]}
resp = self.server.get("/api/v0/blueprints/info/http-server")
data = json.loads(resp.data)
self.assertEqual(data, info_dict_1)
info_dict_2 = {"changes":[{"changed":False, "name":"glusterfs"},
{"changed":False, "name":"http-server"}],
"errors":[],
"blueprints":[{"description": "An example GlusterFS server with samba",
"modules":[{"name":"glusterfs", "version":"3.7.*"},
{"name":"glusterfs-cli", "version":"3.7.*"}],
"name":"glusterfs",
"packages":[{"name":"samba", "version":"4.2.*"}],
"version": "0.0.1"},
{"description":"An example http server with PHP and MySQL support.",
"modules":[{"name":"httpd", "version":"2.4.*"},
{"name":"mod_auth_kerb", "version":"5.4"},
{"name":"mod_ssl", "version":"2.4.*"},
{"name":"php", "version":"5.4.*"},
{"name": "php-mysql", "version":"5.4.*"}],
"name":"http-server",
"packages": [{"name":"openssh-server", "version": "6.6.*"},
{"name": "rsync", "version": "3.0.*"},
{"name": "tmux", "version": "2.2"}],
"version": "0.0.1"},
]}
resp = self.server.get("/api/v0/blueprints/info/http-server,glusterfs")
data = json.loads(resp.data)
self.assertEqual(data, info_dict_2)
info_dict_3 = {"changes":[],
"errors":["missing-blueprint: No commits for missing-blueprint.toml on the master branch."],
"blueprints":[]
}
resp = self.server.get("/api/v0/blueprints/info/missing-blueprint")
data = json.loads(resp.data)
self.assertEqual(data, info_dict_3)
def test_04_blueprints_changes(self):
"""Test the /api/v0/blueprints/changes route"""
resp = self.server.get("/api/v0/blueprints/changes/http-server")
data = json.loads(resp.data)
# Can't compare a whole dict since commit hash and timestamps will change.
# Should have 1 commit (for now), with a matching message.
self.assertEqual(data["limit"], 20)
self.assertEqual(data["offset"], 0)
self.assertEqual(len(data["errors"]), 0)
self.assertEqual(len(data["blueprints"]), 1)
self.assertEqual(data["blueprints"][0]["name"], "http-server")
self.assertEqual(len(data["blueprints"][0]["changes"]), 1)
def test_04a_blueprints_diff_empty_ws(self):
"""Test the /api/v0/diff/NEWEST/WORKSPACE with empty workspace"""
resp = self.server.get("/api/v0/blueprints/diff/glusterfs/NEWEST/WORKSPACE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data, {"diff": []})
def test_05_blueprints_new_json(self):
"""Test the /api/v0/blueprints/new route with json blueprint"""
test_blueprint = {"description": "An example GlusterFS server with samba",
"name":"glusterfs",
"version": "0.2.0",
"modules":[{"name":"glusterfs", "version":"3.7.*"},
{"name":"glusterfs-cli", "version":"3.7.*"}],
"packages":[{"name":"samba", "version":"4.2.*"},
{"name":"tmux", "version":"2.2"}]}
resp = self.server.post("/api/v0/blueprints/new",
data=json.dumps(test_blueprint),
content_type="application/json")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/info/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0], test_blueprint)
def test_06_blueprints_new_toml(self):
"""Test the /api/v0/blueprints/new route with toml blueprint"""
test_blueprint = open(joinpaths(self.examples_path, "glusterfs.toml"), "rb").read()
resp = self.server.post("/api/v0/blueprints/new",
data=test_blueprint,
content_type="text/x-toml")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/info/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertEqual(len(blueprints), 1)
# Returned blueprint has had its version bumped to 0.2.1
test_blueprint = toml.loads(test_blueprint)
test_blueprint["version"] = "0.2.1"
self.assertEqual(blueprints[0], test_blueprint)
def test_07_blueprints_ws_json(self):
"""Test the /api/v0/blueprints/workspace route with json blueprint"""
test_blueprint = {"description": "An example GlusterFS server with samba, ws version",
"name":"glusterfs",
"version": "0.3.0",
"modules":[{"name":"glusterfs", "version":"3.7.*"},
{"name":"glusterfs-cli", "version":"3.7.*"}],
"packages":[{"name":"samba", "version":"4.2.*"},
{"name":"tmux", "version":"2.2"}]}
resp = self.server.post("/api/v0/blueprints/workspace",
data=json.dumps(test_blueprint),
content_type="application/json")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/info/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0], test_blueprint)
changes = data.get("changes")
self.assertEqual(len(changes), 1)
self.assertEqual(changes[0], {"name":"glusterfs", "changed":True})
def test_08_blueprints_ws_toml(self):
"""Test the /api/v0/blueprints/workspace route with toml blueprint"""
test_blueprint = {"description": "An example GlusterFS server with samba, ws version",
"name":"glusterfs",
"version": "0.4.0",
"modules":[{"name":"glusterfs", "version":"3.7.*"},
{"name":"glusterfs-cli", "version":"3.7.*"}],
"packages":[{"name":"samba", "version":"4.2.*"},
{"name":"tmux", "version":"2.2"}]}
resp = self.server.post("/api/v0/blueprints/workspace",
data=json.dumps(test_blueprint),
content_type="application/json")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/info/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0], test_blueprint)
changes = data.get("changes")
self.assertEqual(len(changes), 1)
self.assertEqual(changes[0], {"name":"glusterfs", "changed":True})
def test_09_blueprints_ws_delete(self):
"""Test DELETE /api/v0/blueprints/workspace/<blueprint_name>"""
# Write to the workspace first, just use the test_blueprints_ws_json test for this
self.test_07_blueprints_ws_json()
# Delete it
resp = self.server.delete("/api/v0/blueprints/workspace/glusterfs")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
# Make sure it isn't the workspace copy and that changed is False
resp = self.server.get("/api/v0/blueprints/info/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0]["version"], "0.2.1")
changes = data.get("changes")
self.assertEqual(len(changes), 1)
self.assertEqual(changes[0], {"name":"glusterfs", "changed":False})
def test_10_blueprints_delete(self):
"""Test DELETE /api/v0/blueprints/delete/<blueprint_name>"""
resp = self.server.delete("/api/v0/blueprints/delete/glusterfs")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
# Make sure glusterfs is no longer in the list of blueprints
resp = self.server.get("/api/v0/blueprints/list")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertEqual("glusterfs" in blueprints, False)
def test_11_blueprints_undo(self):
"""Test POST /api/v0/blueprints/undo/<blueprint_name>/<commit>"""
resp = self.server.get("/api/v0/blueprints/changes/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
# Revert it to the first commit
blueprints = data.get("blueprints")
self.assertNotEqual(blueprints, None)
changes = blueprints[0].get("changes")
self.assertEqual(len(changes) > 1, True)
# Revert it to the first commit
commit = changes[-1]["commit"]
resp = self.server.post("/api/v0/blueprints/undo/glusterfs/%s" % commit)
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/changes/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertNotEqual(blueprints, None)
changes = blueprints[0].get("changes")
self.assertEqual(len(changes) > 1, True)
expected_msg = "glusterfs.toml reverted to commit %s" % commit
self.assertEqual(changes[0]["message"], expected_msg)
def test_12_blueprints_tag(self):
"""Test POST /api/v0/blueprints/tag/<blueprint_name>"""
resp = self.server.post("/api/v0/blueprints/tag/glusterfs")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/changes/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
# Revert it to the first commit
blueprints = data.get("blueprints")
self.assertNotEqual(blueprints, None)
changes = blueprints[0].get("changes")
self.assertEqual(len(changes) > 1, True)
self.assertEqual(changes[0]["revision"], 1)
def test_13_blueprints_diff(self):
"""Test /api/v0/blueprints/diff/<blueprint_name>/<from_commit>/<to_commit>"""
resp = self.server.get("/api/v0/blueprints/changes/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertNotEqual(blueprints, None)
changes = blueprints[0].get("changes")
self.assertEqual(len(changes) >= 2, True)
from_commit = changes[1].get("commit")
self.assertNotEqual(from_commit, None)
to_commit = changes[0].get("commit")
self.assertNotEqual(to_commit, None)
# Get the differences between the two commits
resp = self.server.get("/api/v0/blueprints/diff/glusterfs/%s/%s" % (from_commit, to_commit))
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data, {"diff": [{"new": {"Version": "0.0.1"}, "old": {"Version": "0.2.1"}}]})
# Write to the workspace and check the diff
test_blueprint = {"description": "An example GlusterFS server with samba, ws version",
"name":"glusterfs",
"version": "0.3.0",
"modules":[{"name":"glusterfs", "version":"3.7.*"},
{"name":"glusterfs-cli", "version":"3.7.*"}],
"packages":[{"name":"samba", "version":"4.2.*"},
{"name":"tmux", "version":"2.2"}]}
resp = self.server.post("/api/v0/blueprints/workspace",
data=json.dumps(test_blueprint),
content_type="application/json")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
# Get the differences between the newest commit and the workspace
resp = self.server.get("/api/v0/blueprints/diff/glusterfs/NEWEST/WORKSPACE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
result = {"diff": [{"new": {"Description": "An example GlusterFS server with samba, ws version"},
"old": {"Description": "An example GlusterFS server with samba"}},
{"new": {"Version": "0.3.0"},
"old": {"Version": "0.0.1"}},
{"new": {"Package": {"version": "2.2", "name": "tmux"}},
"old": None}]}
self.assertEqual(data, result)
def test_14_blueprints_depsolve(self):
"""Test /api/v0/blueprints/depsolve/<blueprint_names>"""
resp = self.server.get("/api/v0/blueprints/depsolve/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertNotEqual(blueprints, None)
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0]["blueprint"]["name"], "glusterfs")
self.assertEqual(len(blueprints[0]["dependencies"]) > 10, True)
self.assertFalse(data.get("errors"))
def test_14_blueprints_depsolve_empty(self):
"""Test /api/v0/blueprints/depsolve/<blueprint_names> on empty blueprint"""
test_blueprint = {"description": "An empty blueprint",
"name":"void",
"version": "0.1.0"}
resp = self.server.post("/api/v0/blueprints/new",
data=json.dumps(test_blueprint),
content_type="application/json")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/depsolve/void")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertNotEqual(blueprints, None)
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0]["blueprint"]["name"], "void")
self.assertEqual(blueprints[0]["blueprint"]["packages"], [])
self.assertEqual(blueprints[0]["blueprint"]["modules"], [])
self.assertEqual(blueprints[0]["dependencies"], [])
self.assertFalse(data.get("errors"))
def test_15_blueprints_freeze(self):
"""Test /api/v0/blueprints/freeze/<blueprint_names>"""
resp = self.server.get("/api/v0/blueprints/freeze/glusterfs")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertNotEqual(blueprints, None)
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0]["blueprint"]["name"], "glusterfs")
evra = blueprints[0]["blueprint"]["modules"][0]["version"]
self.assertEqual(len(evra) > 10, True)
def test_projects_list(self):
"""Test /api/v0/projects/list"""
resp = self.server.get("/api/v0/projects/list")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
projects = data.get("projects")
self.assertEqual(len(projects) > 10, True)
def test_projects_info(self):
"""Test /api/v0/projects/info/<project_names>"""
resp = self.server.get("/api/v0/projects/info/bash")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
projects = data.get("projects")
self.assertEqual(len(projects), 1)
self.assertEqual(projects[0]["name"], "bash")
self.assertEqual(projects[0]["builds"][0]["source"]["license"], "GPLv3+")
def test_projects_depsolve(self):
"""Test /api/v0/projects/depsolve/<project_names>"""
resp = self.server.get("/api/v0/projects/depsolve/bash")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
deps = data.get("projects")
self.assertEqual(len(deps) > 10, True)
self.assertEqual(deps[0]["name"], "basesystem")
def test_modules_list(self):
"""Test /api/v0/modules/list"""
resp = self.server.get("/api/v0/modules/list")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
modules = data.get("modules")
self.assertEqual(len(modules) > 10, True)
self.assertEqual(modules[0]["group_type"], "rpm")
resp = self.server.get("/api/v0/modules/list/d*")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
modules = data.get("modules")
self.assertEqual(len(modules) > 0, True)
self.assertEqual(modules[0]["name"].startswith("d"), True)
def test_modules_info(self):
"""Test /api/v0/modules/info"""
resp = self.server.get("/api/v0/modules/info/bash")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
modules = data.get("modules")
self.assertEqual(len(modules), 1)
self.assertEqual(modules[0]["name"], "bash")
self.assertEqual(modules[0]["dependencies"][0]["name"], "basesystem")
def test_blueprint_new_branch(self):
"""Test the /api/v0/blueprints/new route with a new branch"""
test_blueprint = {"description": "An example GlusterFS server with samba",
"name":"glusterfs",
"version": "0.2.0",
"modules":[{"name":"glusterfs", "version":"3.7.*"},
{"name":"glusterfs-cli", "version":"3.7.*"}],
"packages":[{"name":"samba", "version":"4.2.*"},
{"name":"tmux", "version":"2.2"}]}
resp = self.server.post("/api/v0/blueprints/new?branch=test",
data=json.dumps(test_blueprint),
content_type="application/json")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
resp = self.server.get("/api/v0/blueprints/info/glusterfs?branch=test")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
blueprints = data.get("blueprints")
self.assertEqual(len(blueprints), 1)
self.assertEqual(blueprints[0], test_blueprint)
def assert_documentation(self, response):
"""
Assert response containing documentation from /api/doc/ is
valid *without* comparing to the actual file on disk.
"""
self.assertEqual(200, response.status_code)
self.assertTrue(len(response.data) > 1024)
# look for some well known strings inside the documentation
self.assertRegexpMatches(response.data, r"Lorax [\d.]+ documentation")
self.assertRegexpMatches(response.data, r"Copyright \d+, Red Hat, Inc.")
def test_api_docs(self):
"""Test the /api/docs/"""
resp = self.server.get("/api/docs/")
self.assert_documentation(resp)
def test_api_docs_with_existing_path(self):
"""Test the /api/docs/modules.html"""
resp = self.server.get("/api/docs/modules.html")
self.assert_documentation(resp)
def wait_for_status(self, uuid, wait_status):
"""Helper function that waits for a status
:param uuid: UUID of the build to check
:type uuid: str
:param wait_status: List of statuses to exit on
:type wait_status: list of str
:returns: True if status was found, False if it timed out
:rtype: bool
This will time out after 60 seconds
"""
start = time.time()
while True:
resp = self.server.get("/api/v0/compose/info/%s" % uuid)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
queue_status = data.get("queue_status")
if queue_status in wait_status:
return True
if time.time() > start + 60:
return False
time.sleep(5)
def test_compose_01_types(self):
"""Test the /api/v0/compose/types route"""
resp = self.server.get("/api/v0/compose/types")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual({"name": "tar", "enabled": True} in data["types"], True)
def test_compose_02_bad_type(self):
"""Test that using an unsupported image type failes"""
test_compose = {"blueprint_name": "glusterfs",
"compose_type": "snakes",
"branch": "master"}
resp = self.server.post("/api/v0/compose?test=1",
data=json.dumps(test_compose),
content_type="application/json")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False, "Failed to fail to start test compose: %s" % data)
self.assertEqual(data["errors"], ["Invalid compose type (snakes), must be one of ['ext4-filesystem', 'live-iso', 'partitioned-disk', 'qcow2', 'tar']"],
"Failed to get errors: %s" % data)
def test_compose_03_status_fail(self):
"""Test that requesting a status for a bad uuid is empty"""
resp = self.server.get("/api/v0/compose/status/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["uuids"], [], "Failed to get empty result bad uuid: %s" % data)
def test_compose_04_cancel_fail(self):
"""Test that requesting a cancel for a bad uuid fails."""
resp = self.server.delete("/api/v0/compose/cancel/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data)
self.assertEqual(data["errors"], ["NO-UUID-TO-SEE-HERE is not a valid build uuid"], "Failed to get errors: %s" % data)
def test_compose_05_delete_fail(self):
"""Test that requesting a delete for a bad uuid fails."""
resp = self.server.delete("/api/v0/compose/delete/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["errors"], ["no-uuid-to-see-here is not a valid build uuid"],
"Failed to get an error for a bad uuid: %s" % data)
def test_compose_06_info_fail(self):
"""Test that requesting info for a bad uuid fails."""
resp = self.server.get("/api/v0/compose/info/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data)
self.assertEqual(data["errors"], ["NO-UUID-TO-SEE-HERE is not a valid build_id"],
"Failed to get errors: %s" % data)
def test_compose_07_metadata_fail(self):
"""Test that requesting metadata for a bad uuid fails."""
resp = self.server.get("/api/v0/compose/metadata/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data)
self.assertEqual(data["errors"], ["NO-UUID-TO-SEE-HERE is not a valid build uuid"],
"Failed to get errors: %s" % data)
def test_compose_08_results_fail(self):
"""Test that requesting results for a bad uuid fails."""
resp = self.server.get("/api/v0/compose/results/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data)
self.assertEqual(data["errors"], ["NO-UUID-TO-SEE-HERE is not a valid build uuid"], "Failed to get errors: %s" % data)
def test_compose_09_logs_fail(self):
"""Test that requesting logs for a bad uuid fails."""
resp = self.server.get("/api/v0/compose/logs/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data)
self.assertEqual(data["errors"], ["NO-UUID-TO-SEE-HERE is not a valid build uuid"],
"Failed to get errors: %s" % data)
def test_compose_10_log_fail(self):
"""Test that requesting log for a bad uuid fails."""
resp = self.server.get("/api/v0/compose/log/NO-UUID-TO-SEE-HERE")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False, "Failed to get an error for a bad uuid: %s" % data)
self.assertEqual(data["errors"], ["NO-UUID-TO-SEE-HERE is not a valid build uuid"],
"Failed to get errors: %s" % data)
def test_compose_11_create_failed(self):
"""Test the /api/v0/compose routes with a failed test compose"""
test_compose = {"blueprint_name": "glusterfs",
"compose_type": "tar",
"branch": "master"}
resp = self.server.post("/api/v0/compose?test=1",
data=json.dumps(test_compose),
content_type="application/json")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data)
build_id = data["build_id"]
# Is it in the queue list (either new or run is fine, based on timing)
resp = self.server.get("/api/v0/compose/queue")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [e["id"] for e in data["new"] + data["run"]]
self.assertEqual(build_id in ids, True, "Failed to add build to the queue")
# Wait for it to start
self.assertEqual(self.wait_for_status(build_id, ["RUNNING"]), True, "Failed to start test compose")
# Wait for it to finish
self.assertEqual(self.wait_for_status(build_id, ["FAILED"]), True, "Failed to finish test compose")
resp = self.server.get("/api/v0/compose/info/%s" % build_id)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["queue_status"], "FAILED", "Build not in FAILED state")
# Test the /api/v0/compose/failed route
resp = self.server.get("/api/v0/compose/failed")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [e["id"] for e in data["failed"]]
self.assertEqual(build_id in ids, True, "Failed build not listed by /compose/failed")
# Test the /api/v0/compose/finished route
resp = self.server.get("/api/v0/compose/finished")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["finished"], [], "Finished build not listed by /compose/finished")
# Test the /api/v0/compose/status/<uuid> route
resp = self.server.get("/api/v0/compose/status/%s" % build_id)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [(e["id"], e["queue_status"]) for e in data["uuids"]]
self.assertEqual((build_id, "FAILED") in ids, True, "Failed build not listed by /compose/status")
# Test the /api/v0/compose/cancel/<uuid> route
resp = self.server.post("/api/v0/compose?test=1",
data=json.dumps(test_compose),
content_type="application/json")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data)
cancel_id = data["build_id"]
# Wait for it to start
self.assertEqual(self.wait_for_status(cancel_id, ["RUNNING"]), True, "Failed to start test compose")
# Cancel the build
resp = self.server.delete("/api/v0/compose/cancel/%s" % cancel_id)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], True, "Failed to cancel test compose: %s" % data)
# Delete the failed build
# Test the /api/v0/compose/delete/<uuid> route
resp = self.server.delete("/api/v0/compose/delete/%s" % build_id)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [(e["uuid"], e["status"]) for e in data["uuids"]]
self.assertEqual((build_id, True) in ids, True, "Failed to delete test compose: %s" % data)
# Make sure the failed list is empty
resp = self.server.get("/api/v0/compose/failed")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["failed"], [], "Failed to delete the failed build: %s" % data)
def test_compose_12_create_finished(self):
"""Test the /api/v0/compose routes with a finished test compose"""
test_compose = {"blueprint_name": "glusterfs",
"compose_type": "tar",
"branch": "master"}
resp = self.server.post("/api/v0/compose?test=2",
data=json.dumps(test_compose),
content_type="application/json")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], True, "Failed to start test compose: %s" % data)
build_id = data["build_id"]
# Is it in the queue list (either new or run is fine, based on timing)
resp = self.server.get("/api/v0/compose/queue")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [e["id"] for e in data["new"] + data["run"]]
self.assertEqual(build_id in ids, True, "Failed to add build to the queue")
# Wait for it to start
self.assertEqual(self.wait_for_status(build_id, ["RUNNING"]), True, "Failed to start test compose")
# Wait for it to finish
self.assertEqual(self.wait_for_status(build_id, ["FINISHED"]), True, "Failed to finish test compose")
resp = self.server.get("/api/v0/compose/info/%s" % build_id)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["queue_status"], "FINISHED", "Build not in FINISHED state")
# Test the /api/v0/compose/finished route
resp = self.server.get("/api/v0/compose/finished")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [e["id"] for e in data["finished"]]
self.assertEqual(build_id in ids, True, "Finished build not listed by /compose/finished")
# Test the /api/v0/compose/failed route
resp = self.server.get("/api/v0/compose/failed")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["failed"], [], "Failed build not listed by /compose/failed")
# Test the /api/v0/compose/status/<uuid> route
resp = self.server.get("/api/v0/compose/status/%s" % build_id)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [(e["id"], e["queue_status"]) for e in data["uuids"]]
self.assertEqual((build_id, "FINISHED") in ids, True, "Finished build not listed by /compose/status")
# Test the /api/v0/compose/metadata/<uuid> route
resp = self.server.get("/api/v0/compose/metadata/%s" % build_id)
self.assertEqual(resp.status_code, 200)
self.assertEqual(len(resp.data) > 1024, True)
# Test the /api/v0/compose/results/<uuid> route
resp = self.server.get("/api/v0/compose/results/%s" % build_id)
self.assertEqual(resp.status_code, 200)
self.assertEqual(len(resp.data) > 1024, True)
# Test the /api/v0/compose/image/<uuid> route
resp = self.server.get("/api/v0/compose/image/%s" % build_id)
self.assertEqual(resp.status_code, 200)
self.assertEqual(len(resp.data) > 0, True)
self.assertEqual(resp.data, "TEST IMAGE")
# Delete the finished build
# Test the /api/v0/compose/delete/<uuid> route
resp = self.server.delete("/api/v0/compose/delete/%s" % build_id)
data = json.loads(resp.data)
self.assertNotEqual(data, None)
ids = [(e["uuid"], e["status"]) for e in data["uuids"]]
self.assertEqual((build_id, True) in ids, True, "Failed to delete test compose: %s" % data)
# Make sure the finished list is empty
resp = self.server.get("/api/v0/compose/finished")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["finished"], [], "Failed to delete the failed build: %s" % data)

90
tests/pylorax/test_workspace.py Normal file
View File

@ -0,0 +1,90 @@
#
# Copyright (C) 2017 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os
import mock
import shutil
import tempfile
import unittest
import pylorax.api.recipes as recipes
from pylorax.api.workspace import workspace_dir, workspace_read, workspace_write, workspace_delete
from pylorax.sysutils import joinpaths
class WorkspaceTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.repo_dir = tempfile.mkdtemp(prefix="lorax.test.repo.")
self.repo = recipes.open_or_create_repo(self.repo_dir)
self.results_path = "./tests/pylorax/results/"
self.examples_path = "./tests/pylorax/blueprints/"
recipe_path = joinpaths(self.examples_path, "http-server.toml")
f = open(recipe_path, 'rb')
self.example_recipe = recipes.recipe_from_toml(f.read())
@classmethod
def tearDownClass(self):
if self.repo is not None:
del self.repo
shutil.rmtree(self.repo_dir)
def test_01_repo_creation(self):
"""Test that creating the repository succeeded"""
self.assertNotEqual(self.repo, None)
def test_02_workspace_dir(self):
"""Test the workspace_dir function"""
ws_dir = workspace_dir(self.repo, "master")
self.assertEqual(ws_dir, joinpaths(self.repo_dir, "git", "workspace", "master"))
def test_03_workspace_write(self):
"""Test the workspace_write function"""
# Use an example recipe
workspace_write(self.repo, "master", self.example_recipe)
# The file should have ended up here
ws_recipe_path = joinpaths(self.repo_dir, "git", "workspace", "master", "http-server.toml")
self.assertEqual(os.path.exists(ws_recipe_path), True)
def test_04_workspace_read(self):
"""Test the workspace_read function"""
# The recipe was written by the workspace_write test. Read it and compare with the source recipe.
recipe = workspace_read(self.repo, "master", "http-server")
self.assertEqual(self.example_recipe, recipe)
def test_04_workspace_read_ioerror(self):
"""Test the workspace_read function dealing with internal IOError"""
# The recipe was written by the workspace_write test.
with self.assertRaises(recipes.RecipeFileError):
with mock.patch('pylorax.api.workspace.recipe_from_toml', side_effect=IOError('TESTING')):
workspace_read(self.repo, "master", "http-server")
def test_05_workspace_delete(self):
"""Test the workspace_delete function"""
ws_recipe_path = joinpaths(self.repo_dir, "git", "workspace", "master", "http-server.toml")
self.assertEqual(os.path.exists(ws_recipe_path), True)
workspace_delete(self.repo, "master", "http-server")
self.assertEqual(os.path.exists(ws_recipe_path), False)
def test_05_workspace_delete_non_existing(self):
"""Test the workspace_delete function"""
ws_recipe_path = joinpaths(self.repo_dir, "git", "workspace", "master", "non-existing.toml")
workspace_delete(self.repo, "master", "non-existing")
self.assertFalse(os.path.exists(ws_recipe_path))

88
tests/pylorax/test_yumbase.py Normal file
View File

@ -0,0 +1,88 @@
#
# Copyright (C) 2018 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import os
import shutil
import tempfile
import unittest
import ConfigParser
from pylorax.api.config import configure, make_yum_dirs
from pylorax.api.yumbase import get_base_object
class YumbaseTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.tmp_dir = tempfile.mkdtemp(prefix="lorax.test.yumbase.")
conf_file = os.path.join(self.tmp_dir, 'test.conf')
open(conf_file, 'w').write("""[composer]
# releasever different from the current default
releasever = 6
[yum]
proxy = https://proxy.example.com
sslverify = False
[repos]
use_system_repos = False
""")
# will read the above configuration
config = configure(conf_file=conf_file, root_dir=self.tmp_dir)
make_yum_dirs(config)
# will read composer config and store a yum config file
self.yb = get_base_object(config)
# will read the stored yum config file
self.yumconf = ConfigParser.ConfigParser()
self.yumconf.read([config.get("composer", "yum_conf")])
@classmethod
def tearDownClass(self):
shutil.rmtree(self.tmp_dir)
def test_stores_yum_proxy_from_composer_config(self):
self.assertEqual('https://proxy.example.com', self.yumconf.get('main', 'proxy'))
def test_disables_sslverify_if_composer_disables_it(self):
self.assertEqual('0', self.yumconf.get('main', 'sslverify'))
def test_sets_releasever_from_composer(self):
self.assertEqual('6', self.yb.conf.yumvar['releasever'])
def test_doesnt_use_system_repos(self):
# no other repos defined for this test
self.assertEqual({}, self.yb._repos.repos)
class CreateYumDirsTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.tmp_dir = tempfile.mkdtemp(prefix="lorax.test.yumbase.")
@classmethod
def tearDownClass(self):
shutil.rmtree(self.tmp_dir)
def test_creates_missing_yum_root_directory(self):
config = configure(test_config=True, root_dir=self.tmp_dir)
# will create the above directory if missing
make_yum_dirs(config)
_ = get_base_object(config)
self.assertTrue(os.path.exists(self.tmp_dir + '/var/tmp/composer/yum/root'))