diff options
Diffstat (limited to 'testing/mozharness/test/test_base_script.py')
| -rw-r--r-- | testing/mozharness/test/test_base_script.py | 898 |
1 files changed, 898 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 000000000..c069a82f3 --- /dev/null +++ b/testing/mozharness/test/test_base_script.py @@ -0,0 +1,898 @@ +import gc +import mock +import os +import re +import shutil +import tempfile +import types +import unittest +PYWIN32 = False +if os.name == 'nt': + try: + import win32file + PYWIN32 = True + except: + pass + + +import mozharness.base.errors as errors +import mozharness.base.log as log +from mozharness.base.log import DEBUG, INFO, WARNING, ERROR, CRITICAL, FATAL, IGNORE +import mozharness.base.script as script +from mozharness.base.config import parse_config_file + + +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 >>fh, test_string + 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, 0100700) + 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.assertEquals(2, self.ATTEMPT_N) + + def testRetryReturns(self): + ret = self.s.retry(self._alwaysPass, sleeptime=0) + self.assertEquals(ret, True) + + def testRetryCleanupIsCalled(self): + cleanup = mock.Mock() + self.s.retry(self._succeedOnSecondAttempt, cleanup=cleanup, sleeptime=0) + self.assertEquals(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']), 3) + + 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() |