| #!/usr/bin/env fuchsia-vendored-python |
| # Copyright 2020 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import contextlib |
| import filecmp |
| import os |
| import shutil |
| import tempfile |
| import time |
| import unittest |
| from pathlib import Path |
| from unittest import mock |
| |
| import output_cacher |
| |
| |
| class TempFileTransformTests(unittest.TestCase): |
| |
| def test_invalid(self): |
| self.assertFalse(output_cacher.TempFileTransform().valid) |
| |
| def test_temp_dir_only(self): |
| self.assertTrue( |
| output_cacher.TempFileTransform(temp_dir="my_tmp").valid) |
| self.assertTrue( |
| output_cacher.TempFileTransform(temp_dir="/my_tmp").valid) |
| |
| def test_basename_prefix_only(self): |
| self.assertTrue( |
| output_cacher.TempFileTransform(basename_prefix="temp-").valid) |
| |
| def test_suffix_only(self): |
| self.assertTrue(output_cacher.TempFileTransform(suffix=".tmp").valid) |
| |
| def test_temp_dir_and_suffix(self): |
| self.assertTrue( |
| output_cacher.TempFileTransform( |
| temp_dir="throw/away", suffix=".tmp").valid) |
| |
| def test_transform_suffix_only(self): |
| self.assertEqual( |
| output_cacher.TempFileTransform(suffix=".tmp").transform( |
| Path("foo/bar.o")), |
| Path("foo/bar.o.tmp"), |
| ) |
| |
| def test_transform_temp_dir_only(self): |
| self.assertEqual( |
| output_cacher.TempFileTransform(temp_dir=Path("t/m/p")).transform( |
| Path("foo/bar.o")), |
| Path("t/m/p/foo/bar.o"), |
| ) |
| |
| def test_transform_suffix_and_temp_dir(self): |
| self.assertEqual( |
| output_cacher.TempFileTransform( |
| temp_dir=Path("/t/m/p"), |
| suffix=".foo").transform(Path("foo/bar.o")), |
| Path("/t/m/p/foo/bar.o.foo"), |
| ) |
| |
| def test_transform_basename_prefix_only(self): |
| self.assertEqual( |
| output_cacher.TempFileTransform(basename_prefix="fake.").transform( |
| Path("bar.o")), |
| Path("fake.bar.o"), |
| ) |
| self.assertEqual( |
| output_cacher.TempFileTransform(basename_prefix="fake.").transform( |
| Path("foo/bar.o")), |
| Path("foo/fake.bar.o"), |
| ) |
| |
| def test_transform_basename_prefix_and_temp_dir(self): |
| self.assertEqual( |
| output_cacher.TempFileTransform( |
| temp_dir="/t/m/p", |
| basename_prefix="xyz-").transform(Path("foo/bar.o")), |
| Path("/t/m/p/foo/xyz-bar.o"), |
| ) |
| |
| |
| class SplitTransformJoinTest(unittest.TestCase): |
| |
| def test_no_change(self): |
| self.assertEqual( |
| output_cacher.split_transform_join('text', '=', lambda x: x), |
| 'text') |
| |
| def test_repeat(self): |
| self.assertEqual( |
| output_cacher.split_transform_join('text', '=', lambda x: x + x), |
| 'texttext') |
| |
| def test_with_split(self): |
| self.assertEqual( |
| output_cacher.split_transform_join('a=b', '=', lambda x: x + x), |
| 'aa=bb') |
| |
| def test_with_split_recorded(self): |
| renamed_tokens = {} |
| |
| def recorded_transform(x): |
| new_text = x + x |
| renamed_tokens[x] = new_text |
| return new_text |
| |
| self.assertEqual( |
| output_cacher.split_transform_join('a=b', '=', recorded_transform), |
| 'aa=bb') |
| self.assertEqual(renamed_tokens, {'a': 'aa', 'b': 'bb'}) |
| |
| |
| class OutputSubstitutionTest(unittest.TestCase): |
| |
| def test_no_prev_opt(self): |
| subst = output_cacher.OutputSubstitution('foo') |
| self.assertEqual(subst.output_name, 'foo') |
| self.assertEqual(subst.match_previous_option, '') |
| |
| def test_with_prev_opt(self): |
| subst = output_cacher.OutputSubstitution('substitute_after:--out:foo') |
| self.assertEqual(subst.output_name, 'foo') |
| self.assertEqual(subst.match_previous_option, '--out') |
| |
| def test_malformed_prev_opt(self): |
| with self.assertRaises(ValueError): |
| output_cacher.OutputSubstitution('substitute_after:--outfoo') |
| |
| |
| class LexicallyRewriteTokenTest(unittest.TestCase): |
| |
| def test_repeat_text(self): |
| self.assertEqual( |
| output_cacher.lexically_rewrite_token('foo', lambda x: x + x), |
| 'foofoo') |
| |
| def test_delimters_only(self): |
| self.assertEqual( |
| output_cacher.lexically_rewrite_token(',,==,=,=,', lambda x: x + x), |
| ',,==,=,=,') |
| |
| def test_flag_with_value(self): |
| |
| def transform(x): |
| if x.startswith('file'): |
| return 'tmp-' + x |
| else: |
| return x |
| |
| self.assertEqual( |
| output_cacher.lexically_rewrite_token('--foo=file1', transform), |
| '--foo=tmp-file1') |
| self.assertEqual( |
| output_cacher.lexically_rewrite_token( |
| 'notfile,file1,file2,notfile', transform), |
| 'notfile,tmp-file1,tmp-file2,notfile') |
| self.assertEqual( |
| output_cacher.lexically_rewrite_token( |
| '--foo=file1,file2', transform), '--foo=tmp-file1,tmp-file2') |
| |
| |
| class EnsureFileExists(unittest.TestCase): |
| |
| def test_file_exists(self): |
| with mock.patch.object(Path, "exists", |
| return_value=True) as mock_exists: |
| output_cacher.ensure_file_exists(Path("some/file.txt")) |
| mock_exists.assert_called() |
| # just make sure no exception raised |
| |
| def test_file_not_exists(self): |
| with mock.patch.object(Path, "exists", |
| return_value=False) as mock_exists: |
| with mock.patch.object(time, "sleep") as mock_sleep: |
| with self.assertRaises(FileNotFoundError): |
| output_cacher.ensure_file_exists(Path("not/a/file.txt")) |
| mock_exists.assert_called() |
| |
| def test_file_exists_later(self): |
| with mock.patch.object(Path, "exists", |
| side_effect=[False, True]) as mock_exists: |
| with mock.patch.object(time, "sleep") as mock_sleep: |
| output_cacher.ensure_file_exists(Path("late/file.txt")) |
| mock_exists.assert_called() |
| mock_sleep.assert_called() |
| # just make sure no exception raised |
| |
| |
| @contextlib.contextmanager |
| def chdir_cm(d: Path): |
| """FIXME: replace with contextlib.chdir(), once Python 3.11 is default.""" |
| save_dir = os.getcwd() |
| os.chdir(d) # could raise OSError |
| try: |
| yield |
| finally: |
| os.chdir(save_dir) |
| |
| |
| class CopyPreserveSubpathTests(unittest.TestCase): |
| |
| def test_subpath(self): |
| with tempfile.TemporaryDirectory() as td1: |
| tdp1 = Path(td1) |
| dest_dir = tdp1 / 'backups' |
| with chdir_cm(tdp1): # working directory |
| srcdir = Path('aa/bb') |
| srcdir.mkdir(parents=True, exist_ok=True) |
| src_file = srcdir / 'c.txt' |
| src_file.write_text('hello\n') |
| output_cacher.copy_preserve_subpath(src_file, dest_dir) |
| dest_file = dest_dir / src_file |
| self.assertTrue(filecmp.cmp(src_file, dest_file, shallow=False)) |
| |
| |
| class DefaultDiffActionTests(unittest.TestCase): |
| |
| def test_no_copy(self): |
| file1 = Path('srcdir-x/file1.bin') |
| file2 = Path('srcdir-y/file2.bin') |
| with mock.patch.object(output_cacher, 'detail_diff') as mock_diff: |
| with mock.patch.object(output_cacher, |
| 'copy_preserve_subpath') as mock_copy: |
| output_cacher.default_diff_action(file1, file2) |
| |
| mock_diff.assert_called_with(file1, file2) |
| mock_copy.assert_not_called() |
| |
| def test_with_copy(self): |
| file1 = Path('srcdir-x/file1.bin') |
| file2 = Path('srcdir-y/file2.bin') |
| export_dir = Path('bad/diffs') |
| with mock.patch.object(output_cacher, 'detail_diff') as mock_diff: |
| with mock.patch.object(output_cacher, |
| 'copy_preserve_subpath') as mock_copy: |
| output_cacher.default_diff_action( |
| file1, file2, miscomparison_export_dir=export_dir) |
| |
| mock_diff.assert_called_with(file1, file2) |
| mock_copy.assert_has_calls( |
| [ |
| mock.call(file1, export_dir), |
| mock.call(file2, export_dir), |
| ], |
| any_order=True) |
| |
| |
| class MoveIfDifferentTests(unittest.TestCase): |
| |
| def test_nonexistent_source(self): |
| with mock.patch.object(Path, "exists", |
| return_value=False) as mock_exists: |
| with mock.patch.object(Path, "rename") as mock_move: |
| with mock.patch.object(Path, "unlink") as mock_remove: |
| with mock.patch.object(time, "sleep") as mock_sleep: |
| with self.assertRaises(FileNotFoundError): |
| output_cacher.move_if_different( |
| Path("source.txt"), Path("dest.txt")) |
| mock_exists.assert_called_with() |
| mock_move.assert_not_called() |
| mock_remove.assert_not_called() |
| |
| def test_new_output(self): |
| |
| with mock.patch.object( |
| Path, |
| "exists", |
| side_effect=[True, False], # source.txt, dest.txt |
| ) as mock_exists: |
| with mock.patch.object(output_cacher, "files_match") as mock_diff: |
| with mock.patch.object(Path, "rename") as mock_move: |
| with mock.patch.object(Path, "unlink") as mock_remove: |
| moved = output_cacher.move_if_different( |
| Path("source.txt"), Path("dest.txt")) |
| self.assertTrue(moved) |
| mock_exists.assert_called_with() |
| mock_diff.assert_not_called() |
| mock_move.assert_called_with(Path("dest.txt")) |
| mock_remove.assert_not_called() |
| |
| def test_updating_output(self): |
| with mock.patch.object(Path, "exists", |
| return_value=True) as mock_exists: |
| with mock.patch.object(output_cacher, "files_match", |
| return_value=False) as mock_diff: |
| with mock.patch.object(Path, "rename") as mock_move: |
| with mock.patch.object(Path, "unlink") as mock_remove: |
| moved = output_cacher.move_if_different( |
| Path("source.txt"), Path("dest.txt")) |
| self.assertTrue(moved) |
| mock_exists.assert_called_with() |
| mock_diff.assert_called_with(Path("dest.txt"), Path("source.txt")) |
| mock_move.assert_called_with(Path("dest.txt")) |
| mock_remove.assert_not_called() |
| |
| def test_cached_output(self): |
| with mock.patch.object(Path, "exists", |
| return_value=True) as mock_exists: |
| with mock.patch.object(output_cacher, "files_match", |
| return_value=True) as mock_diff: |
| with mock.patch.object(Path, "rename") as mock_move: |
| with mock.patch.object(Path, "unlink") as mock_remove: |
| with mock.patch.object( |
| output_cacher, |
| "ensure_file_exists") as mock_exists: |
| moved = output_cacher.move_if_different( |
| Path("source.txt"), Path("dest.txt")) |
| self.assertFalse(moved) |
| mock_exists.assert_called() |
| mock_diff.assert_called() |
| mock_move.assert_not_called() |
| mock_remove.assert_called_with() |
| |
| |
| class SubstituteCommandTest(unittest.TestCase): |
| |
| def test_no_substitutions(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action(command=["run.sh"], substitutions={}) |
| repl, renamed = action.substitute_command(transform) |
| self.assertEqual(repl, ["run.sh"]) |
| self.assertEqual(renamed, {}) |
| |
| def test_one_suffix(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "out.txt"], substitutions={"out.txt": ""}) |
| repl, renamed = action.substitute_command(transform) |
| self.assertEqual(repl, ["run.sh", "out.txt.tmp"]) |
| self.assertEqual(renamed, {Path("out.txt"): Path("out.txt.tmp")}) |
| |
| def test_require_flag_match_missing_flag(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "out.txt"], substitutions={"out.txt": "-f"}) |
| repl, renamed = action.substitute_command(transform) |
| self.assertEqual(repl, ["run.sh", "out.txt"]) |
| self.assertEqual(renamed, {}) |
| |
| def test_require_flag_match_wrong_flag(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "-g", "out.txt"], |
| substitutions={"out.txt": "-f"}) |
| repl, renamed = action.substitute_command(transform) |
| self.assertEqual(repl, ["run.sh", "-g", "out.txt"]) |
| self.assertEqual(renamed, {}) |
| |
| def test_require_flag_match_have_flag(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "-f", "out.txt"], |
| substitutions={"out.txt": "-f"}) |
| repl, renamed = action.substitute_command(transform) |
| self.assertEqual(repl, ["run.sh", "-f", "out.txt.tmp"]) |
| self.assertEqual(renamed, {Path("out.txt"): Path("out.txt.tmp")}) |
| |
| def test_match_only_one_flag(self): |
| transform = output_cacher.TempFileTransform(basename_prefix="tmp-") |
| action = output_cacher.Action( |
| command=["run.sh", "-x", "out.txt", "-f", "out.txt"], |
| substitutions={"out.txt": "-f"}) |
| repl, renamed = action.substitute_command(transform) |
| self.assertEqual(repl, ["run.sh", "-x", "out.txt", "-f", "tmp-out.txt"]) |
| self.assertEqual(renamed, {Path("out.txt"): Path("tmp-out.txt")}) |
| |
| def test_match_only_one_flag_first_occurrence(self): |
| transform = output_cacher.TempFileTransform(basename_prefix="tmp-") |
| action = output_cacher.Action( |
| command=["run.sh", "out.txt", "-f", "out.txt", "out.txt"], |
| substitutions={"out.txt": "-f"}) |
| repl, renamed = action.substitute_command(transform) |
| self.assertEqual( |
| repl, ["run.sh", "out.txt", "-f", "tmp-out.txt", "out.txt"]) |
| self.assertEqual(renamed, {Path("out.txt"): Path("tmp-out.txt")}) |
| |
| |
| class ReplaceOutputArgsTest(unittest.TestCase): |
| |
| def test_dry_run(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action(command=["run.sh"], substitutions={}) |
| with mock.patch.object(output_cacher, |
| "execute_main_command") as mock_call: |
| self.assertEqual(action.run_cached(transform, dry_run=True), 0) |
| mock_call.assert_not_called() |
| |
| def test_command_failed(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action(command=["run.sh"], substitutions={}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=1) as mock_call: |
| with mock.patch.object(output_cacher, |
| "move_if_different") as mock_update: |
| self.assertEqual(action.run_cached(transform), 1) |
| mock_call.assert_called_with(["run.sh"]) |
| mock_update.assert_not_called() |
| |
| def test_command_passed_using_suffix(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "out.put"], |
| substitutions={"out.put": ""}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| # This test doesn't care if move happened. |
| with mock.patch.object(output_cacher, "move_if_different", |
| return_value=False) as mock_update: |
| self.assertEqual(action.run_cached(transform), 0) |
| mock_call.assert_called_with(["run.sh", "in.put", "out.put.tmp"]) |
| mock_update.assert_called_with( |
| src=Path("out.put.tmp"), dest=Path("out.put"), verbose=False) |
| |
| def test_command_passed_using_relative_temp_dir(self): |
| transform = output_cacher.TempFileTransform(temp_dir="temp/temp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "foo/out.put"], |
| substitutions={"foo/out.put": ""}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| # This test doesn't care if move happened. |
| with mock.patch.object(output_cacher, "move_if_different", |
| return_value=False) as mock_update: |
| with mock.patch.object(Path, "mkdir") as mock_mkdir: |
| self.assertEqual(action.run_cached(transform), 0) |
| mock_mkdir.assert_called_with(parents=True, exist_ok=True) |
| mock_call.assert_called_with( |
| ["run.sh", "in.put", "temp/temp/foo/out.put"]) |
| mock_update.assert_called_with( |
| src=Path("temp/temp/foo/out.put"), |
| dest=Path("foo/out.put"), |
| verbose=False) |
| |
| |
| def fake_detail_diff(left: Path, right: Path): |
| # Don't print anything, or attempt to call a diff script |
| pass |
| |
| |
| class RunRepeatedlyCompareTests(unittest.TestCase): |
| |
| def test_command_failed(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action(command=["run.sh"], substitutions={}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=1) as mock_call: |
| with mock.patch.object(output_cacher, "files_match") as mock_match: |
| with mock.patch.object(os.path, "exists", |
| return_value=True) as mock_exists: |
| with mock.patch.object(shutil, "copy2") as mock_copy: |
| with mock.patch.object(os, "makedirs") as mock_mkdir: |
| with mock.patch.object(time, "sleep") as mock_zzz: |
| self.assertEqual( |
| action.run_repeatedly_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| ), 1) |
| mock_call.assert_called_once_with(["run.sh"]) |
| mock_match.assert_not_called() |
| mock_exists.assert_not_called() |
| mock_mkdir.assert_not_called() |
| mock_copy.assert_not_called() |
| |
| def test_command_passed_and_rerun_matches(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "out.put"], |
| substitutions={"out.put": ""}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| with mock.patch.object(output_cacher, "files_match", |
| return_value=True) as mock_match: |
| with mock.patch.object(Path, "is_file", |
| return_value=True) as mock_isfile: |
| with mock.patch.object(output_cacher, |
| "remove_if_exists") as mock_remove: |
| with mock.patch.object(Path, "mkdir") as mock_mkdir: |
| with mock.patch.object(shutil, |
| "copy2") as mock_copy: |
| with mock.patch.object(time, |
| "sleep") as mock_zzz: |
| self.assertEqual( |
| action. |
| run_repeatedly_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| max_attempts=2, |
| ), 0) |
| mock_call.assert_has_calls( |
| [ |
| mock.call(["run.sh", "in.put", "out.put"]), # first run |
| mock.call(["run.sh", "in.put", "out.put"]), # re-run 1 |
| mock.call(["run.sh", "in.put", "out.put"]), # re-run 2 |
| ], |
| any_order=True) |
| mock_match.assert_called_with(Path("out.put"), Path("out.put.tmp")) |
| mock_remove.assert_called_with(Path("out.put.tmp")) |
| mock_isfile.assert_called() |
| mock_mkdir.assert_not_called() # using suffix, not temp_dir |
| mock_copy.assert_called_once_with( |
| Path("out.put"), Path("out.put.tmp"), follow_symlinks=False) |
| |
| def test_command_passed_and_rerun_differs(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "out.put"], |
| substitutions={"out.put": ""}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| with mock.patch.object( |
| output_cacher, |
| "files_match", |
| # mismatch on 2nd attempt |
| side_effect=[True, False]) as mock_match: |
| with mock.patch.object(Path, "is_file", |
| return_value=True) as mock_isfile: |
| with mock.patch.object(output_cacher, |
| "remove_if_exists") as mock_remove: |
| with mock.patch.object(Path, "mkdir") as mock_mkdir: |
| with mock.patch.object(shutil, |
| "copy2") as mock_copy: |
| with mock.patch.object(time, |
| "sleep") as mock_zzz: |
| self.assertEqual( |
| action. |
| run_repeatedly_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| max_attempts=10, |
| ), 1) |
| mock_call.assert_has_calls( |
| [ |
| mock.call(["run.sh", "in.put", "out.put"]), |
| mock.call(["run.sh", "in.put", "out.put"]), |
| mock.call(["run.sh", "in.put", "out.put"]), |
| # Stop after first miscomparison, before max_attempts reached. |
| ], |
| any_order=True) |
| mock_match.assert_called_with(Path("out.put"), Path("out.put.tmp")) |
| mock_remove.assert_not_called() # mismatches kept |
| mock_isfile.assert_called() |
| mock_mkdir.assert_not_called() # using suffix, not temp_dir |
| mock_copy.assert_called_once_with( |
| Path("out.put"), Path("out.put.tmp"), follow_symlinks=False) |
| |
| def test_command_passed_and_some_outputs_differ(self): |
| |
| def fake_match(file1: Path, file2: Path) -> bool: |
| if file1 == Path("out.put"): |
| return True |
| elif file1 == Path("out2.put"): |
| return False |
| raise ValueError(f"Unhandled file name: {file1}") |
| |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "out.put", "out2.put"], |
| substitutions={ |
| "out.put": "", |
| "out2.put": "" |
| }) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| with mock.patch.object(output_cacher, "files_match", |
| wraps=fake_match) as mock_match: |
| with mock.patch.object(Path, "is_file", |
| return_value=True) as mock_isfile: |
| with mock.patch.object(output_cacher, |
| "remove_if_exists") as mock_remove: |
| with mock.patch.object(Path, "mkdir") as mock_mkdir: |
| with mock.patch.object(shutil, |
| "copy2") as mock_copy: |
| with mock.patch.object(time, |
| "sleep") as mock_zzz: |
| self.assertEqual( |
| action. |
| run_repeatedly_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| ), 1) |
| mock_call.assert_has_calls( |
| [ |
| mock.call(["run.sh", "in.put", "out.put", "out2.put"]), |
| mock.call(["run.sh", "in.put", "out.put", "out2.put"]), |
| ], |
| any_order=True) |
| mock_match.assert_has_calls( |
| [ |
| mock.call(Path("out.put"), Path("out.put.tmp")), |
| mock.call(Path("out2.put"), Path("out2.put.tmp")), |
| ], |
| any_order=True) |
| mock_remove.assert_not_called() # keep around mismatches |
| mock_isfile.assert_called() |
| mock_mkdir.assert_not_called() # using suffix, not temp_dir |
| mock_copy.assert_has_calls( |
| [ |
| mock.call( |
| Path("out.put"), Path("out.put.tmp"), |
| follow_symlinks=False), |
| mock.call( |
| Path("out2.put"), |
| Path("out2.put.tmp"), |
| follow_symlinks=False), |
| ], |
| any_order=True) |
| |
| |
| class RunTwiceWithSubstitutionCompareTests(unittest.TestCase): |
| |
| def test_command_failed(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action(command=["run.sh"], substitutions={}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=1) as mock_call: |
| with mock.patch.object(output_cacher, "files_match") as mock_match: |
| self.assertEqual( |
| action.run_twice_with_substitution_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| ), 1) |
| mock_call.assert_called_once_with(["run.sh"]) |
| mock_match.assert_not_called() |
| |
| def test_command_passed_and_rerun_matches(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "out.put"], |
| substitutions={"out.put": ""}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| with mock.patch.object(output_cacher, "files_match", |
| return_value=True) as mock_match: |
| with mock.patch.object(Path, "unlink") as mock_remove: |
| self.assertEqual( |
| action.run_twice_with_substitution_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| ), 0) |
| mock_call.assert_has_calls( |
| [ |
| mock.call(["run.sh", "in.put", "out.put"]), |
| mock.call(["run.sh", "in.put", "out.put.tmp"]), |
| ], |
| any_order=True) |
| mock_match.assert_called_with(Path("out.put"), Path("out.put.tmp")) |
| mock_remove.assert_called_with() |
| |
| def test_command_passed_and_rerun_differs(self): |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "out.put"], |
| substitutions={"out.put": ""}) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| with mock.patch.object(output_cacher, "files_match", |
| return_value=False) as mock_match: |
| with mock.patch.object(os, "remove") as mock_remove: |
| self.assertEqual( |
| action.run_twice_with_substitution_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| ), 1) |
| mock_call.assert_has_calls( |
| [ |
| mock.call(["run.sh", "in.put", "out.put"]), |
| mock.call(["run.sh", "in.put", "out.put.tmp"]), |
| ], |
| any_order=True) |
| mock_match.assert_called_with(Path("out.put"), Path("out.put.tmp")) |
| mock_remove.assert_not_called() |
| |
| def test_command_passed_and_some_outptus_differ(self): |
| |
| def fake_match(file1: Path, file2: Path) -> bool: |
| if file1 == Path("out.put"): |
| return True |
| elif file1 == Path("out2.put"): |
| return False |
| raise ValueError(f"Unhandled file name: {file1}") |
| |
| transform = output_cacher.TempFileTransform(suffix=".tmp") |
| action = output_cacher.Action( |
| command=["run.sh", "in.put", "out.put", "out2.put"], |
| substitutions={ |
| "out.put": "", |
| "out2.put": "" |
| }) |
| with mock.patch.object(output_cacher, "execute_main_command", |
| return_value=0) as mock_call: |
| with mock.patch.object(output_cacher, "files_match", |
| wraps=fake_match) as mock_match: |
| with mock.patch.object(Path, "unlink") as mock_remove: |
| self.assertEqual( |
| action.run_twice_with_substitution_and_compare_outputs( |
| tempfile_transform=transform, |
| diff_action=fake_detail_diff, |
| ), 1) |
| mock_call.assert_has_calls( |
| [ |
| mock.call(["run.sh", "in.put", "out.put", "out2.put"]), |
| mock.call(["run.sh", "in.put", "out.put.tmp", "out2.put.tmp"]), |
| ], |
| any_order=True) |
| mock_match.assert_has_calls( |
| [ |
| mock.call(Path("out.put"), Path("out.put.tmp")), |
| mock.call(Path("out2.put"), Path("out2.put.tmp")), |
| ], |
| any_order=True) |
| mock_remove.assert_has_calls([mock.call()]) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
