From 9784f0ec903ed18af0e0a8f056ee6ef23c87d41f Mon Sep 17 00:00:00 2001 From: Lars Wirzenius Date: Fri, 30 Mar 2018 11:21:04 +0300 Subject: Add: classes for pipeline actions --- ick2/actions_tests.py | 230 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 230 insertions(+) create mode 100644 ick2/actions_tests.py (limited to 'ick2/actions_tests.py') diff --git a/ick2/actions_tests.py b/ick2/actions_tests.py new file mode 100644 index 0000000..8220fe2 --- /dev/null +++ b/ick2/actions_tests.py @@ -0,0 +1,230 @@ +# Copyright (C) 2018 Lars Wirzenius +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 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 Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . + + +import os +import shutil +import tempfile +import unittest + + +import ick2 + + +class ActionFactoryTests(unittest.TestCase): + + def setUp(self): + self.tempdir = tempfile.mkdtemp() + self.workspace_area = self.tempdir + self.project = 'magic' + self.af = ick2.ActionFactory('systree', self.workspace_area, None) + + def tearDown(self): + shutil.rmtree(self.tempdir) + + def test_sets_token(self): + token = 'this is a token' + self.af.set_token(token) + self.assertEqual(self.af.get_token(), token) + + def test_lists_allows_environments(self): + allowed = self.af.get_allowed_environments() + self.assertTrue(isinstance(allowed, list)) + self.assertNotEqual(allowed, []) + + def test_creates_host_environment_by_default(self): + spec = { + 'shell': 'echo hello world', + } + env = self.af.create_environment(spec, self.project) + self.assertTrue(isinstance(env, ick2.HostEnvironment)) + + self.assertEqual( + env.get_workspace_directory(), + os.path.join(self.workspace_area, self.project)) + + def test_creates_host_environment_when_asked_explicitly(self): + spec = { + 'where': 'host', + } + env = self.af.create_environment(spec, self.project) + self.assertTrue(isinstance(env, ick2.HostEnvironment)) + + def test_creates_chroot_environment_by_default(self): + spec = { + 'where': 'chroot', + } + env = self.af.create_environment(spec, self.project) + self.assertTrue(isinstance(env, ick2.ChrootEnvironment)) + + def test_creates_container_environment_by_default(self): + spec = { + 'where': 'container', + } + env = self.af.create_environment(spec, self.project) + self.assertTrue(isinstance(env, ick2.ContainerEnvironment)) + + def test_creates_shell_action_on_host(self): + def url(blob_name): + return blob_name + + spec = { + 'shell': 'echo hello, world', + } + token = 'secret token' + self.af.set_token(token) + self.af.set_blob_url_func(url) + action = self.af.create_action(spec, self.project) + self.assertTrue(isinstance(action, ick2.ShellAction)) + self.assertTrue(isinstance(action.get_env(), ick2.HostEnvironment)) + self.assertEqual(action.get_token(), token) + self.assertEqual(action.get_blob_upload_url('foo'), 'foo') + + def test_creates_python_action_on_host(self): + spec = { + 'python': 'pass', + } + action = self.af.create_action(spec, self.project) + self.assertTrue(isinstance(action, ick2.PythonAction)) + self.assertTrue(isinstance(action.get_env(), ick2.HostEnvironment)) + + def test_creates_debootstrap_action_on_host(self): + spec = { + 'debootstrap': 'auto', + } + action = self.af.create_action(spec, self.project) + self.assertTrue(isinstance(action, ick2.DebootstrapAction)) + self.assertTrue(isinstance(action.get_env(), ick2.HostEnvironment)) + + def test_creates_create_workspace_action_on_host(self): + spec = { + 'action': 'create_workspace', + } + action = self.af.create_action(spec, self.project) + self.assertTrue(isinstance(action, ick2.CreateWorkspaceAction)) + self.assertTrue(isinstance(action.get_env(), ick2.HostEnvironment)) + + def test_creates_archive_workspace_action_on_host(self): + spec = { + 'archive': 'workspace', + } + action = self.af.create_action(spec, self.project) + self.assertTrue(isinstance(action, ick2.ArchiveWorkspaceAction)) + self.assertTrue(isinstance(action.get_env(), ick2.HostEnvironment)) + + def test_creates_populate_sysgtre_action_on_host(self): + spec = { + 'action': 'populate_systree', + } + action = self.af.create_action(spec, self.project) + self.assertTrue(isinstance(action, ick2.PopulateSystreeAction)) + self.assertTrue(isinstance(action.get_env(), ick2.HostEnvironment)) + + def test_raises_exception_for_unknown_step(self): + with self.assertRaises(ick2.UnknownStepError): + self.af.create_action({}, self.project) + + +class ShellActionTests(unittest.TestCase): + + def test_encodes_parameters(self): + params = { + 'foo': 'bar', + } + action = ick2.ShellAction(None) + encoded = action.encode64(params) + decoded = action.decode64(encoded) + self.assertEqual(params, decoded) + + def test_runs_argv(self): + params = {} + step = { + 'shell': 'echo hello, world', + } + env = DummyEnvironment() + env.exit_code = 42 + action = ick2.ShellAction(env) + exit_code = action.execute(params, step) + self.assertEqual(exit_code, env.exit_code) + + +class PythonActionTests(unittest.TestCase): + + def test_runs_argv(self): + params = {} + step = { + 'python': 'pass', + } + env = DummyEnvironment() + env.exit_code = 42 + action = ick2.PythonAction(env) + exit_code = action.execute(params, step) + self.assertEqual(exit_code, env.exit_code) + + +class DebootstrapActionTests(unittest.TestCase): + + def test_runs_argv(self): + params = { + 'debian_codename': 'stretch', + } + step = { + 'debootstrap': 'auto', + 'mirror': 'http://deb.debian.org/debian', + } + env = DummyEnvironment() + env.exit_code = 42 + action = ick2.DebootstrapAction(env) + exit_code = action.execute(params, step) + self.assertEqual(exit_code, env.exit_code) + + +class CreateWorkspaceActionTests(unittest.TestCase): + + def setUp(self): + self.tempdir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.tempdir) + + def test_runs_argv(self): + params = {} + step = { + 'action': 'create_workspace', + } + env = DummyEnvironment(tempdir=self.tempdir) + action = ick2.CreateWorkspaceAction(env) + exit_code = action.execute(params, step) + self.assertEqual(exit_code, 0) + + +class DummyEnvironment: + + def __init__(self, tempdir=None): + self.exit_code = None + self.argv = None + self.tempdir = tempdir or tempfile.mkdtemp() + + def report(self, exit_code, msg): + pass + + def host_runcmd(self, argv, **kwargs): + self.argv = argv + return self.exit_code + + def runcmd(self, argv, **kwargs): + return self.host_runcmd(argv) + + def get_workspace_directory(self): + return self.tempdir -- cgit v1.2.1