diff options
Diffstat (limited to 'testing/mozharness/test/test_base_script.py')
| -rw-r--r-- | testing/mozharness/test/test_base_script.py | 960 |
1 files changed, 960 insertions, 0 deletions
diff --git a/testing/mozharness/test/test_base_script.py b/testing/mozharness/test/test_base_script.py new file mode 100644 index 0000000000..75175dc794 --- /dev/null +++ b/testing/mozharness/test/test_base_script.py @@ -0,0 +1,960 @@ +import gc +import os +import re +import shutil +import tempfile +import types +import unittest +from unittest import mock + +PYWIN32 = False +if os.name == "nt": + try: + import win32file + + PYWIN32 = True + except ImportError: + pass + + +import mozharness.base.errors as errors +import mozharness.base.log as log +import mozharness.base.script as script +from mozharness.base.config import parse_config_file +from mozharness.base.log import CRITICAL, DEBUG, ERROR, FATAL, IGNORE, INFO, WARNING + +here = os.path.dirname(os.path.abspath(__file__)) + +test_string = """foo +bar +baz""" + + +class CleanupObj(script.ScriptMixin, log.LogMixin): + def __init__(self): + super(CleanupObj, self).__init__() + self.log_obj = None + self.config = {"log_level": ERROR} + + +def cleanup(files=None): + files = files or [] + files.extend(("test_logs", "test_dir", "tmpfile_stdout", "tmpfile_stderr")) + gc.collect() + c = CleanupObj() + for f in files: + c.rmtree(f) + + +def get_debug_script_obj(): + s = script.BaseScript( + config={"log_type": "multi", "log_level": DEBUG}, + initial_config_file="test/test.json", + ) + return s + + +def _post_fatal(self, **kwargs): + fh = open("tmpfile_stdout", "w") + print(test_string, file=fh) + fh.close() + + +# TestScript {{{1 +class TestScript(unittest.TestCase): + def setUp(self): + cleanup() + self.s = None + self.tmpdir = tempfile.mkdtemp(suffix=".mozharness") + + def tearDown(self): + # Close the logfile handles, or windows can't remove the logs + if hasattr(self, "s") and isinstance(self.s, object): + del self.s + cleanup([self.tmpdir]) + + # test _dump_config_hierarchy() when --dump-config-hierarchy is passed + def test_dump_config_hierarchy_valid_files_len(self): + try: + self.s = script.BaseScript( + initial_config_file="test/test.json", + option_args=["--cfg", "test/test_override.py,test/test_override2.py"], + config={"dump_config_hierarchy": True}, + ) + except SystemExit: + local_cfg_files = parse_config_file("test_logs/localconfigfiles.json") + # first let's see if the correct number of config files were + # realized + self.assertEqual( + len(local_cfg_files), + 4, + msg="--dump-config-hierarchy dumped wrong number of config files", + ) + + def test_dump_config_hierarchy_keys_unique_and_valid(self): + try: + self.s = script.BaseScript( + initial_config_file="test/test.json", + option_args=["--cfg", "test/test_override.py,test/test_override2.py"], + config={"dump_config_hierarchy": True}, + ) + except SystemExit: + local_cfg_files = parse_config_file("test_logs/localconfigfiles.json") + # now let's see if only unique items were added from each config + t_override = local_cfg_files.get("test/test_override.py", {}) + self.assertTrue( + t_override.get("keep_string") == "don't change me" + and len(t_override.keys()) == 1, + msg="--dump-config-hierarchy dumped wrong keys/value for " + "`test/test_override.py`. There should only be one " + "item and it should be unique to all the other " + "items in test_log/localconfigfiles.json.", + ) + + def test_dump_config_hierarchy_matches_self_config(self): + try: + ###### + # we need temp_cfg because self.s will be gcollected (NoneType) by + # the time we get to SystemExit exception + # temp_cfg will differ from self.s.config because of + # 'dump_config_hierarchy'. we have to make a deepcopy because + # config is a locked dict + temp_s = script.BaseScript( + initial_config_file="test/test.json", + option_args=["--cfg", "test/test_override.py,test/test_override2.py"], + ) + from copy import deepcopy + + temp_cfg = deepcopy(temp_s.config) + temp_cfg.update({"dump_config_hierarchy": True}) + ###### + self.s = script.BaseScript( + initial_config_file="test/test.json", + option_args=["--cfg", "test/test_override.py,test/test_override2.py"], + config={"dump_config_hierarchy": True}, + ) + except SystemExit: + local_cfg_files = parse_config_file("test_logs/localconfigfiles.json") + # finally let's just make sure that all the items added up, equals + # what we started with: self.config + target_cfg = {} + for cfg_file in local_cfg_files: + target_cfg.update(local_cfg_files[cfg_file]) + self.assertEqual( + target_cfg, + temp_cfg, + msg="all of the items (combined) in each cfg file dumped via " + "--dump-config-hierarchy does not equal self.config ", + ) + + # test _dump_config() when --dump-config is passed + def test_dump_config_equals_self_config(self): + try: + ###### + # we need temp_cfg because self.s will be gcollected (NoneType) by + # the time we get to SystemExit exception + # temp_cfg will differ from self.s.config because of + # 'dump_config_hierarchy'. we have to make a deepcopy because + # config is a locked dict + temp_s = script.BaseScript( + initial_config_file="test/test.json", + option_args=["--cfg", "test/test_override.py,test/test_override2.py"], + ) + from copy import deepcopy + + temp_cfg = deepcopy(temp_s.config) + temp_cfg.update({"dump_config": True}) + ###### + self.s = script.BaseScript( + initial_config_file="test/test.json", + option_args=["--cfg", "test/test_override.py,test/test_override2.py"], + config={"dump_config": True}, + ) + except SystemExit: + target_cfg = parse_config_file("test_logs/localconfig.json") + self.assertEqual( + target_cfg, + temp_cfg, + msg="all of the items (combined) in each cfg file dumped via " + "--dump-config does not equal self.config ", + ) + + def test_nonexistent_mkdir_p(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + self.s.mkdir_p("test_dir/foo/bar/baz") + self.assertTrue(os.path.isdir("test_dir/foo/bar/baz"), msg="mkdir_p error") + + def test_existing_mkdir_p(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + os.makedirs("test_dir/foo/bar/baz") + self.s.mkdir_p("test_dir/foo/bar/baz") + self.assertTrue( + os.path.isdir("test_dir/foo/bar/baz"), msg="mkdir_p error when dir exists" + ) + + def test_chdir(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + cwd = os.getcwd() + self.s.chdir("test_logs") + self.assertEqual(os.path.join(cwd, "test_logs"), os.getcwd(), msg="chdir error") + self.s.chdir(cwd) + + def _test_log_helper(self, obj): + obj.debug("Testing DEBUG") + obj.warning("Testing WARNING") + obj.error("Testing ERROR") + obj.critical("Testing CRITICAL") + try: + obj.fatal("Testing FATAL") + except SystemExit: + pass + else: + self.assertTrue(False, msg="fatal() didn't SystemExit!") + + def test_log(self): + self.s = get_debug_script_obj() + self.s.log_obj = None + self._test_log_helper(self.s) + del self.s + self.s = script.BaseScript(initial_config_file="test/test.json") + self._test_log_helper(self.s) + + def test_run_nonexistent_command(self): + self.s = get_debug_script_obj() + self.s.run_command( + command="this_cmd_should_not_exist --help", + env={"GARBLE": "FARG"}, + error_list=errors.PythonErrorList, + ) + error_logsize = os.path.getsize("test_logs/test_error.log") + self.assertTrue(error_logsize > 0, msg="command not found error not hit") + + def test_run_command_in_bad_dir(self): + self.s = get_debug_script_obj() + self.s.run_command( + command="ls", + cwd="/this_dir_should_not_exist", + error_list=errors.PythonErrorList, + ) + error_logsize = os.path.getsize("test_logs/test_error.log") + self.assertTrue(error_logsize > 0, msg="bad dir error not hit") + + def test_get_output_from_command_in_bad_dir(self): + self.s = get_debug_script_obj() + self.s.get_output_from_command(command="ls", cwd="/this_dir_should_not_exist") + error_logsize = os.path.getsize("test_logs/test_error.log") + self.assertTrue(error_logsize > 0, msg="bad dir error not hit") + + def test_get_output_from_command_with_missing_file(self): + self.s = get_debug_script_obj() + self.s.get_output_from_command(command="ls /this_file_should_not_exist") + error_logsize = os.path.getsize("test_logs/test_error.log") + self.assertTrue(error_logsize > 0, msg="bad file error not hit") + + def test_get_output_from_command_with_missing_file2(self): + self.s = get_debug_script_obj() + self.s.run_command( + command="cat mozharness/base/errors.py", + error_list=[ + {"substr": "error", "level": ERROR}, + { + "regex": re.compile(",$"), + "level": IGNORE, + }, + { + "substr": "]$", + "level": WARNING, + }, + ], + ) + error_logsize = os.path.getsize("test_logs/test_error.log") + self.assertTrue(error_logsize > 0, msg="error list not working properly") + + def test_download_unpack(self): + # NOTE: The action is called *download*, however, it can work for files in disk + self.s = get_debug_script_obj() + + archives_path = os.path.join(here, "helper_files", "archives") + + # Test basic decompression + for archive in ( + "archive.tar", + "archive.tar.bz2", + "archive.tar.gz", + "archive.zip", + ): + self.s.download_unpack( + url=os.path.join(archives_path, archive), extract_to=self.tmpdir + ) + self.assertIn("script.sh", os.listdir(os.path.join(self.tmpdir, "bin"))) + self.assertIn("lorem.txt", os.listdir(self.tmpdir)) + shutil.rmtree(self.tmpdir) + + # Test permissions for extracted entries from zip archive + self.s.download_unpack( + url=os.path.join(archives_path, "archive.zip"), + extract_to=self.tmpdir, + ) + file_stats = os.stat(os.path.join(self.tmpdir, "bin", "script.sh")) + orig_fstats = os.stat( + os.path.join(archives_path, "reference", "bin", "script.sh") + ) + self.assertEqual(file_stats.st_mode, orig_fstats.st_mode) + shutil.rmtree(self.tmpdir) + + # Test unzip specific dirs only + self.s.download_unpack( + url=os.path.join(archives_path, "archive.zip"), + extract_to=self.tmpdir, + extract_dirs=["bin/*"], + ) + self.assertIn("bin", os.listdir(self.tmpdir)) + self.assertNotIn("lorem.txt", os.listdir(self.tmpdir)) + shutil.rmtree(self.tmpdir) + + # Test for invalid filenames (Windows only) + if PYWIN32: + with self.assertRaises(IOError): + self.s.download_unpack( + url=os.path.join(archives_path, "archive_invalid_filename.zip"), + extract_to=self.tmpdir, + ) + + def test_unpack(self): + self.s = get_debug_script_obj() + + archives_path = os.path.join(here, "helper_files", "archives") + + # Test basic decompression + for archive in ( + "archive.tar", + "archive.tar.bz2", + "archive.tar.gz", + "archive.zip", + ): + self.s.unpack(os.path.join(archives_path, archive), self.tmpdir) + self.assertIn("script.sh", os.listdir(os.path.join(self.tmpdir, "bin"))) + self.assertIn("lorem.txt", os.listdir(self.tmpdir)) + shutil.rmtree(self.tmpdir) + + # Test permissions for extracted entries from zip archive + self.s.unpack(os.path.join(archives_path, "archive.zip"), self.tmpdir) + file_stats = os.stat(os.path.join(self.tmpdir, "bin", "script.sh")) + orig_fstats = os.stat( + os.path.join(archives_path, "reference", "bin", "script.sh") + ) + self.assertEqual(file_stats.st_mode, orig_fstats.st_mode) + shutil.rmtree(self.tmpdir) + + # Test extract specific dirs only + self.s.unpack( + os.path.join(archives_path, "archive.zip"), + self.tmpdir, + extract_dirs=["bin/*"], + ) + self.assertIn("bin", os.listdir(self.tmpdir)) + self.assertNotIn("lorem.txt", os.listdir(self.tmpdir)) + shutil.rmtree(self.tmpdir) + + # Test for invalid filenames (Windows only) + if PYWIN32: + with self.assertRaises(IOError): + self.s.unpack( + os.path.join(archives_path, "archive_invalid_filename.zip"), + self.tmpdir, + ) + + +# TestHelperFunctions {{{1 +class TestHelperFunctions(unittest.TestCase): + temp_file = "test_dir/mozilla" + + def setUp(self): + cleanup() + self.s = None + + def tearDown(self): + # Close the logfile handles, or windows can't remove the logs + if hasattr(self, "s") and isinstance(self.s, object): + del self.s + cleanup() + + def _create_temp_file(self, contents=test_string): + os.mkdir("test_dir") + fh = open(self.temp_file, "w+") + fh.write(contents) + fh.close + + def test_mkdir_p(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + self.s.mkdir_p("test_dir") + self.assertTrue(os.path.isdir("test_dir"), msg="mkdir_p error") + + def test_get_output_from_command(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + contents = self.s.get_output_from_command( + ["bash", "-c", "cat %s" % self.temp_file] + ) + self.assertEqual( + test_string, + contents, + msg="get_output_from_command('cat file') differs from fh.write", + ) + + def test_run_command(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + temp_file_name = os.path.basename(self.temp_file) + self.assertEqual( + self.s.run_command("cat %s" % temp_file_name, cwd="test_dir"), + 0, + msg="run_command('cat file') did not exit 0", + ) + + def test_move1(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + temp_file2 = "%s2" % self.temp_file + self.s.move(self.temp_file, temp_file2) + self.assertFalse( + os.path.exists(self.temp_file), + msg="%s still exists after move()" % self.temp_file, + ) + + def test_move2(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + temp_file2 = "%s2" % self.temp_file + self.s.move(self.temp_file, temp_file2) + self.assertTrue( + os.path.exists(temp_file2), msg="%s doesn't exist after move()" % temp_file2 + ) + + def test_copyfile(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + temp_file2 = "%s2" % self.temp_file + self.s.copyfile(self.temp_file, temp_file2) + self.assertEqual( + os.path.getsize(self.temp_file), + os.path.getsize(temp_file2), + msg="%s and %s are different sizes after copyfile()" + % (self.temp_file, temp_file2), + ) + + def test_existing_rmtree(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + self.s.mkdir_p("test_dir/foo/bar/baz") + self.s.rmtree("test_dir") + self.assertFalse(os.path.exists("test_dir"), msg="rmtree unsuccessful") + + def test_nonexistent_rmtree(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + status = self.s.rmtree("test_dir") + self.assertFalse(status, msg="nonexistent rmtree error") + + @unittest.skipUnless(PYWIN32, "PyWin32 specific") + def test_long_dir_rmtree(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + # create a very long path that the command-prompt cannot delete + # by using unicode format (max path length 32000) + path = u"\\\\?\\%s\\test_dir" % os.getcwd() + win32file.CreateDirectoryExW(u".", path) + + for x in range(0, 20): + print("path=%s" % path) + path = path + u"\\%sxxxxxxxxxxxxxxxxxxxx" % x + win32file.CreateDirectoryExW(u".", path) + self.s.rmtree("test_dir") + self.assertFalse(os.path.exists("test_dir"), msg="rmtree unsuccessful") + + @unittest.skipUnless(PYWIN32, "PyWin32 specific") + def test_chmod_rmtree(self): + self._create_temp_file() + win32file.SetFileAttributesW(self.temp_file, win32file.FILE_ATTRIBUTE_READONLY) + self.s = script.BaseScript(initial_config_file="test/test.json") + self.s.rmtree("test_dir") + self.assertFalse(os.path.exists("test_dir"), msg="rmtree unsuccessful") + + @unittest.skipIf(os.name == "nt", "Not for Windows") + def test_chmod(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + self.s.chmod(self.temp_file, 0o100700) + self.assertEqual(os.stat(self.temp_file)[0], 33216, msg="chmod unsuccessful") + + def test_env_normal(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + script_env = self.s.query_env() + self.assertEqual( + script_env, + os.environ, + msg="query_env() != env\n%s\n%s" % (script_env, os.environ), + ) + + def test_env_normal2(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + self.s.query_env() + script_env = self.s.query_env() + self.assertEqual( + script_env, + os.environ, + msg="Second query_env() != env\n%s\n%s" % (script_env, os.environ), + ) + + def test_env_partial(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + script_env = self.s.query_env(partial_env={"foo": "bar"}) + self.assertTrue("foo" in script_env and script_env["foo"] == "bar") + + def test_env_path(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + partial_path = "yaddayadda:%(PATH)s" + full_path = partial_path % {"PATH": os.environ["PATH"]} + script_env = self.s.query_env(partial_env={"PATH": partial_path}) + self.assertEqual(script_env["PATH"], full_path) + + def test_query_exe(self): + self.s = script.BaseScript( + initial_config_file="test/test.json", + config={"exes": {"foo": "bar"}}, + ) + path = self.s.query_exe("foo") + self.assertEqual(path, "bar") + + def test_query_exe_string_replacement(self): + self.s = script.BaseScript( + initial_config_file="test/test.json", + config={ + "base_work_dir": "foo", + "work_dir": "bar", + "exes": {"foo": os.path.join("%(abs_work_dir)s", "baz")}, + }, + ) + path = self.s.query_exe("foo") + self.assertEqual(path, os.path.join("foo", "bar", "baz")) + + def test_read_from_file(self): + self._create_temp_file() + self.s = script.BaseScript(initial_config_file="test/test.json") + contents = self.s.read_from_file(self.temp_file) + self.assertEqual(contents, test_string) + + def test_read_from_nonexistent_file(self): + self.s = script.BaseScript(initial_config_file="test/test.json") + contents = self.s.read_from_file("nonexistent_file!!!") + self.assertEqual(contents, None) + + +# TestScriptLogging {{{1 +class TestScriptLogging(unittest.TestCase): + # I need a log watcher helper function, here and in test_log. + def setUp(self): + cleanup() + self.s = None + + def tearDown(self): + # Close the logfile handles, or windows can't remove the logs + if hasattr(self, "s") and isinstance(self.s, object): + del self.s + cleanup() + + def test_info_logsize(self): + self.s = script.BaseScript( + config={"log_type": "multi"}, initial_config_file="test/test.json" + ) + info_logsize = os.path.getsize("test_logs/test_info.log") + self.assertTrue(info_logsize > 0, msg="initial info logfile missing/size 0") + + def test_add_summary_info(self): + self.s = script.BaseScript( + config={"log_type": "multi"}, initial_config_file="test/test.json" + ) + info_logsize = os.path.getsize("test_logs/test_info.log") + self.s.add_summary("one") + info_logsize2 = os.path.getsize("test_logs/test_info.log") + self.assertTrue( + info_logsize < info_logsize2, msg="add_summary() info not logged" + ) + + def test_add_summary_warning(self): + self.s = script.BaseScript( + config={"log_type": "multi"}, initial_config_file="test/test.json" + ) + warning_logsize = os.path.getsize("test_logs/test_warning.log") + self.s.add_summary("two", level=WARNING) + warning_logsize2 = os.path.getsize("test_logs/test_warning.log") + self.assertTrue( + warning_logsize < warning_logsize2, + msg="add_summary(level=%s) not logged in warning log" % WARNING, + ) + + def test_summary(self): + self.s = script.BaseScript( + config={"log_type": "multi"}, initial_config_file="test/test.json" + ) + self.s.add_summary("one") + self.s.add_summary("two", level=WARNING) + info_logsize = os.path.getsize("test_logs/test_info.log") + warning_logsize = os.path.getsize("test_logs/test_warning.log") + self.s.summary() + info_logsize2 = os.path.getsize("test_logs/test_info.log") + warning_logsize2 = os.path.getsize("test_logs/test_warning.log") + msg = "" + if info_logsize >= info_logsize2: + msg += "summary() didn't log to info!\n" + if warning_logsize >= warning_logsize2: + msg += "summary() didn't log to warning!\n" + self.assertEqual(msg, "", msg=msg) + + def _test_log_level(self, log_level, log_level_file_list): + self.s = script.BaseScript( + config={"log_type": "multi"}, initial_config_file="test/test.json" + ) + if log_level != FATAL: + self.s.log("testing", level=log_level) + else: + self.s._post_fatal = types.MethodType(_post_fatal, self.s) + try: + self.s.fatal("testing") + except SystemExit: + contents = None + if os.path.exists("tmpfile_stdout"): + fh = open("tmpfile_stdout") + contents = fh.read() + fh.close() + self.assertEqual(contents.rstrip(), test_string, "_post_fatal failed!") + del self.s + msg = "" + for level in log_level_file_list: + log_path = "test_logs/test_%s.log" % level + if not os.path.exists(log_path): + msg += "%s doesn't exist!\n" % log_path + else: + filesize = os.path.getsize(log_path) + if not filesize > 0: + msg += "%s is size 0!\n" % log_path + self.assertEqual(msg, "", msg=msg) + + def test_debug(self): + self._test_log_level(DEBUG, []) + + def test_ignore(self): + self._test_log_level(IGNORE, []) + + def test_info(self): + self._test_log_level(INFO, [INFO]) + + def test_warning(self): + self._test_log_level(WARNING, [INFO, WARNING]) + + def test_error(self): + self._test_log_level(ERROR, [INFO, WARNING, ERROR]) + + def test_critical(self): + self._test_log_level(CRITICAL, [INFO, WARNING, ERROR, CRITICAL]) + + def test_fatal(self): + self._test_log_level(FATAL, [INFO, WARNING, ERROR, CRITICAL, FATAL]) + + +# TestRetry {{{1 +class NewError(Exception): + pass + + +class OtherError(Exception): + pass + + +class TestRetry(unittest.TestCase): + def setUp(self): + self.ATTEMPT_N = 1 + self.s = script.BaseScript(initial_config_file="test/test.json") + + def tearDown(self): + # Close the logfile handles, or windows can't remove the logs + if hasattr(self, "s") and isinstance(self.s, object): + del self.s + cleanup() + + def _succeedOnSecondAttempt(self, foo=None, exception=Exception): + if self.ATTEMPT_N == 2: + self.ATTEMPT_N += 1 + return + self.ATTEMPT_N += 1 + raise exception("Fail") + + def _raiseCustomException(self): + return self._succeedOnSecondAttempt(exception=NewError) + + def _alwaysPass(self): + self.ATTEMPT_N += 1 + return True + + def _mirrorArgs(self, *args, **kwargs): + return args, kwargs + + def _alwaysFail(self): + raise Exception("Fail") + + def testRetrySucceed(self): + # Will raise if anything goes wrong + self.s.retry(self._succeedOnSecondAttempt, attempts=2, sleeptime=0) + + def testRetryFailWithoutCatching(self): + self.assertRaises( + Exception, self.s.retry, self._alwaysFail, sleeptime=0, exceptions=() + ) + + def testRetryFailEnsureRaisesLastException(self): + self.assertRaises( + SystemExit, self.s.retry, self._alwaysFail, sleeptime=0, error_level=FATAL + ) + + def testRetrySelectiveExceptionSucceed(self): + self.s.retry( + self._raiseCustomException, + attempts=2, + sleeptime=0, + retry_exceptions=(NewError,), + ) + + def testRetrySelectiveExceptionFail(self): + self.assertRaises( + NewError, + self.s.retry, + self._raiseCustomException, + attempts=2, + sleeptime=0, + retry_exceptions=(OtherError,), + ) + + # TODO: figure out a way to test that the sleep actually happened + def testRetryWithSleep(self): + self.s.retry(self._succeedOnSecondAttempt, attempts=2, sleeptime=1) + + def testRetryOnlyRunOnce(self): + """Tests that retry() doesn't call the action again after success""" + self.s.retry(self._alwaysPass, attempts=3, sleeptime=0) + # self.ATTEMPT_N gets increased regardless of pass/fail + self.assertEqual(2, self.ATTEMPT_N) + + def testRetryReturns(self): + ret = self.s.retry(self._alwaysPass, sleeptime=0) + self.assertEqual(ret, True) + + def testRetryCleanupIsCalled(self): + cleanup = mock.Mock() + self.s.retry(self._succeedOnSecondAttempt, cleanup=cleanup, sleeptime=0) + self.assertEqual(cleanup.call_count, 1) + + def testRetryArgsPassed(self): + args = (1, "two", 3) + kwargs = dict(foo="a", bar=7) + ret = self.s.retry( + self._mirrorArgs, args=args, kwargs=kwargs.copy(), sleeptime=0 + ) + print(ret) + self.assertEqual(ret[0], args) + self.assertEqual(ret[1], kwargs) + + +class BaseScriptWithDecorators(script.BaseScript): + def __init__(self, *args, **kwargs): + super(BaseScriptWithDecorators, self).__init__(*args, **kwargs) + + self.pre_run_1_args = [] + self.raise_during_pre_run_1 = False + self.pre_action_1_args = [] + self.raise_during_pre_action_1 = False + self.pre_action_2_args = [] + self.pre_action_3_args = [] + self.post_action_1_args = [] + self.raise_during_post_action_1 = False + self.post_action_2_args = [] + self.post_action_3_args = [] + self.post_run_1_args = [] + self.raise_during_post_run_1 = False + self.post_run_2_args = [] + self.raise_during_build = False + + @script.PreScriptRun + def pre_run_1(self, *args, **kwargs): + self.pre_run_1_args.append((args, kwargs)) + + if self.raise_during_pre_run_1: + raise Exception(self.raise_during_pre_run_1) + + @script.PreScriptAction + def pre_action_1(self, *args, **kwargs): + self.pre_action_1_args.append((args, kwargs)) + + if self.raise_during_pre_action_1: + raise Exception(self.raise_during_pre_action_1) + + @script.PreScriptAction + def pre_action_2(self, *args, **kwargs): + self.pre_action_2_args.append((args, kwargs)) + + @script.PreScriptAction("clobber") + def pre_action_3(self, *args, **kwargs): + self.pre_action_3_args.append((args, kwargs)) + + @script.PostScriptAction + def post_action_1(self, *args, **kwargs): + self.post_action_1_args.append((args, kwargs)) + + if self.raise_during_post_action_1: + raise Exception(self.raise_during_post_action_1) + + @script.PostScriptAction + def post_action_2(self, *args, **kwargs): + self.post_action_2_args.append((args, kwargs)) + + @script.PostScriptAction("build") + def post_action_3(self, *args, **kwargs): + self.post_action_3_args.append((args, kwargs)) + + @script.PostScriptRun + def post_run_1(self, *args, **kwargs): + self.post_run_1_args.append((args, kwargs)) + + if self.raise_during_post_run_1: + raise Exception(self.raise_during_post_run_1) + + @script.PostScriptRun + def post_run_2(self, *args, **kwargs): + self.post_run_2_args.append((args, kwargs)) + + def build(self): + if self.raise_during_build: + raise Exception(self.raise_during_build) + + +class TestScriptDecorators(unittest.TestCase): + def setUp(self): + cleanup() + self.s = None + + def tearDown(self): + if hasattr(self, "s") and isinstance(self.s, object): + del self.s + + cleanup() + + def test_decorators_registered(self): + self.s = BaseScriptWithDecorators(initial_config_file="test/test.json") + + self.assertEqual(len(self.s._listeners["pre_run"]), 1) + self.assertEqual(len(self.s._listeners["pre_action"]), 3) + self.assertEqual(len(self.s._listeners["post_action"]), 3) + self.assertEqual(len(self.s._listeners["post_run"]), 2) + + def test_pre_post_fired(self): + self.s = BaseScriptWithDecorators(initial_config_file="test/test.json") + self.s.run() + + self.assertEqual(len(self.s.pre_run_1_args), 1) + self.assertEqual(len(self.s.pre_action_1_args), 2) + self.assertEqual(len(self.s.pre_action_2_args), 2) + self.assertEqual(len(self.s.pre_action_3_args), 1) + self.assertEqual(len(self.s.post_action_1_args), 2) + self.assertEqual(len(self.s.post_action_2_args), 2) + self.assertEqual(len(self.s.post_action_3_args), 1) + self.assertEqual(len(self.s.post_run_1_args), 1) + + self.assertEqual(self.s.pre_run_1_args[0], ((), {})) + + self.assertEqual(self.s.pre_action_1_args[0], (("clobber",), {})) + self.assertEqual(self.s.pre_action_1_args[1], (("build",), {})) + + # pre_action_3 should only get called for the action it is registered + # with. + self.assertEqual(self.s.pre_action_3_args[0], (("clobber",), {})) + + self.assertEqual(self.s.post_action_1_args[0][0], ("clobber",)) + self.assertEqual(self.s.post_action_1_args[0][1], dict(success=True)) + self.assertEqual(self.s.post_action_1_args[1][0], ("build",)) + self.assertEqual(self.s.post_action_1_args[1][1], dict(success=True)) + + # post_action_3 should only get called for the action it is registered + # with. + self.assertEqual(self.s.post_action_3_args[0], (("build",), dict(success=True))) + + self.assertEqual(self.s.post_run_1_args[0], ((), {})) + + def test_post_always_fired(self): + self.s = BaseScriptWithDecorators(initial_config_file="test/test.json") + self.s.raise_during_build = "Testing post always fired." + + with self.assertRaises(SystemExit): + self.s.run() + + self.assertEqual(len(self.s.pre_run_1_args), 1) + self.assertEqual(len(self.s.pre_action_1_args), 2) + self.assertEqual(len(self.s.post_action_1_args), 2) + self.assertEqual(len(self.s.post_action_2_args), 2) + self.assertEqual(len(self.s.post_run_1_args), 1) + self.assertEqual(len(self.s.post_run_2_args), 1) + + self.assertEqual(self.s.post_action_1_args[0][1], dict(success=True)) + self.assertEqual(self.s.post_action_1_args[1][1], dict(success=False)) + self.assertEqual(self.s.post_action_2_args[1][1], dict(success=False)) + + def test_pre_run_exception(self): + self.s = BaseScriptWithDecorators(initial_config_file="test/test.json") + self.s.raise_during_pre_run_1 = "Error during pre run 1" + + with self.assertRaises(SystemExit): + self.s.run() + + self.assertEqual(len(self.s.pre_run_1_args), 1) + self.assertEqual(len(self.s.pre_action_1_args), 0) + self.assertEqual(len(self.s.post_run_1_args), 1) + self.assertEqual(len(self.s.post_run_2_args), 1) + + def test_pre_action_exception(self): + self.s = BaseScriptWithDecorators(initial_config_file="test/test.json") + self.s.raise_during_pre_action_1 = "Error during pre 1" + + with self.assertRaises(SystemExit): + self.s.run() + + self.assertEqual(len(self.s.pre_run_1_args), 1) + self.assertEqual(len(self.s.pre_action_1_args), 1) + self.assertEqual(len(self.s.pre_action_2_args), 0) + self.assertEqual(len(self.s.post_action_1_args), 1) + self.assertEqual(len(self.s.post_action_2_args), 1) + self.assertEqual(len(self.s.post_run_1_args), 1) + self.assertEqual(len(self.s.post_run_2_args), 1) + + def test_post_action_exception(self): + self.s = BaseScriptWithDecorators(initial_config_file="test/test.json") + self.s.raise_during_post_action_1 = "Error during post 1" + + with self.assertRaises(SystemExit): + self.s.run() + + self.assertEqual(len(self.s.pre_run_1_args), 1) + self.assertEqual(len(self.s.post_action_1_args), 1) + self.assertEqual(len(self.s.post_action_2_args), 1) + self.assertEqual(len(self.s.post_run_1_args), 1) + self.assertEqual(len(self.s.post_run_2_args), 1) + + def test_post_run_exception(self): + self.s = BaseScriptWithDecorators(initial_config_file="test/test.json") + self.s.raise_during_post_run_1 = "Error during post run 1" + + with self.assertRaises(SystemExit): + self.s.run() + + self.assertEqual(len(self.s.post_run_1_args), 1) + self.assertEqual(len(self.s.post_run_2_args), 1) + + +# main {{{1 +if __name__ == "__main__": + unittest.main() |