blob: 3256f9e72f74c681dc685f9ba2e611a28a8cced8 [file] [log] [blame]
#!/usr/bin/env python3.8
# 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 os
import shutil
import subprocess
import time
import unittest
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("foo/bar.o"), "foo/bar.o.tmp")
def test_transform_temp_dir_only(self):
self.assertEqual(
output_cacher.TempFileTransform(
temp_dir="t/m/p").transform("foo/bar.o"), "t/m/p/foo/bar.o")
def test_transform_suffix_and_temp_dir(self):
self.assertEqual(
output_cacher.TempFileTransform(
temp_dir="/t/m/p", suffix=".foo").transform("foo/bar.o"),
"/t/m/p/foo/bar.o.foo")
def test_transform_basename_prefix_only(self):
self.assertEqual(
output_cacher.TempFileTransform(
basename_prefix="fake.").transform("bar.o"), "fake.bar.o")
self.assertEqual(
output_cacher.TempFileTransform(
basename_prefix="fake.").transform("foo/bar.o"),
"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("foo/bar.o"),
"/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(os.path, "exists",
return_value=True) as mock_exists:
output_cacher.ensure_file_exists("some/file.txt")
mock_exists.assert_called()
# just make sure no exception raised
def test_file_not_exists(self):
with mock.patch.object(os.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("not/a/file.txt")
mock_exists.assert_called()
def test_file_exists_later(self):
with mock.patch.object(os.path, "exists",
side_effect=[False, True]) as mock_exists:
with mock.patch.object(time, "sleep") as mock_sleep:
output_cacher.ensure_file_exists("late/file.txt")
mock_exists.assert_called()
mock_sleep.assert_called()
# just make sure no exception raised
class MoveIfDifferentTests(unittest.TestCase):
def test_nonexistent_source(self):
with mock.patch.object(os.path, "exists",
return_value=False) as mock_exists:
with mock.patch.object(shutil, "move") as mock_move:
with mock.patch.object(os, "remove") as mock_remove:
with mock.patch.object(time, "sleep") as mock_sleep:
with self.assertRaises(FileNotFoundError):
output_cacher.move_if_different(
"source.txt", "dest.txt")
mock_exists.assert_called()
mock_move.assert_not_called()
mock_remove.assert_not_called()
def test_new_output(self):
def fake_exists(path):
if path == "source.txt":
return True
elif path == "dest.txt":
return False
with mock.patch.object(os.path, "exists",
wraps=fake_exists) as mock_exists:
with mock.patch.object(output_cacher, "files_match") as mock_diff:
with mock.patch.object(shutil, "move") as mock_move:
with mock.patch.object(os, "remove") as mock_remove:
moved = output_cacher.move_if_different(
"source.txt", "dest.txt")
self.assertTrue(moved)
mock_exists.assert_called()
mock_diff.assert_not_called()
mock_move.assert_called_with("source.txt", "dest.txt")
mock_remove.assert_not_called()
def test_updating_output(self):
with mock.patch.object(os.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(shutil, "move") as mock_move:
with mock.patch.object(os, "remove") as mock_remove:
moved = output_cacher.move_if_different(
"source.txt", "dest.txt")
self.assertTrue(moved)
mock_exists.assert_called()
mock_diff.assert_called()
mock_move.assert_called_with("source.txt", "dest.txt")
mock_remove.assert_not_called()
def test_cached_output(self):
with mock.patch.object(os.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(shutil, "move") as mock_move:
with mock.patch.object(os, "remove") as mock_remove:
with mock.patch.object(
output_cacher,
"ensure_file_exists") as mock_exists:
moved = output_cacher.move_if_different(
"source.txt", "dest.txt")
self.assertFalse(moved)
mock_exists.assert_called()
mock_diff.assert_called()
mock_move.assert_not_called()
mock_remove.assert_called_with("source.txt")
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, {"out.txt": "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, {"out.txt": "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, {"out.txt": "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, {"out.txt": "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(subprocess, "call") 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(subprocess, "call", 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(subprocess, "call", 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="out.put.tmp", dest="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(subprocess, "call", 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(os, "makedirs") as mock_mkdir:
self.assertEqual(action.run_cached(transform), 0)
mock_mkdir.assert_called_with("temp/temp/foo", exist_ok=True)
mock_call.assert_called_with(
["run.sh", "in.put", "temp/temp/foo/out.put"])
mock_update.assert_called_with(
src="temp/temp/foo/out.put", dest="foo/out.put", verbose=False)
class RunTwiceCompareTests(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(subprocess, "call", 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:
self.assertEqual(
action.run_twice_and_compare_outputs(transform),
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(subprocess, "call", return_value=0) as mock_call:
with mock.patch.object(output_cacher, "files_match",
return_value=True) as mock_match:
with mock.patch.object(os.path, "isfile",
return_value=True) as mock_isfile:
with mock.patch.object(os, "remove") as mock_remove:
with mock.patch.object(os, "makedirs") as mock_mkdir:
with mock.patch.object(shutil,
"copy2") as mock_copy:
self.assertEqual(
action.run_twice_and_compare_outputs(
transform), 0)
mock_call.assert_has_calls(
[
mock.call(["run.sh", "in.put", "out.put"]),
mock.call(["run.sh", "in.put", "out.put"]),
],
any_order=True)
mock_match.assert_called_with("out.put", "out.put.tmp")
mock_remove.assert_called_with("out.put.tmp")
mock_isfile.assert_called()
mock_mkdir.assert_not_called() # using suffix, not temp_dir
mock_copy.assert_called_once_with(
"out.put", "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(subprocess, "call", 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.path, "isfile",
return_value=True) as mock_isfile:
with mock.patch.object(os, "remove") as mock_remove:
with mock.patch.object(os, "makedirs") as mock_mkdir:
with mock.patch.object(shutil,
"copy2") as mock_copy:
self.assertEqual(
action.run_twice_and_compare_outputs(
transform), 1)
mock_call.assert_has_calls(
[
mock.call(["run.sh", "in.put", "out.put"]),
mock.call(["run.sh", "in.put", "out.put"]),
],
any_order=True)
mock_match.assert_called_with("out.put", "out.put.tmp")
mock_remove.assert_not_called()
mock_isfile.assert_called()
mock_mkdir.assert_not_called() # using suffix, not temp_dir
mock_copy.assert_called_once_with(
"out.put", "out.put.tmp", follow_symlinks=False)
def test_command_passed_and_some_outptus_differ(self):
def fake_match(file1: str, file2: str) -> bool:
if file1 == "out.put":
return True
elif file1 == "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(subprocess, "call", return_value=0) as mock_call:
with mock.patch.object(output_cacher, "files_match",
wraps=fake_match) as mock_match:
with mock.patch.object(os.path, "isfile",
return_value=True) as mock_isfile:
with mock.patch.object(os, "remove") as mock_remove:
with mock.patch.object(os, "makedirs") as mock_mkdir:
with mock.patch.object(shutil,
"copy2") as mock_copy:
self.assertEqual(
action.run_twice_and_compare_outputs(
transform), 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("out.put", "out.put.tmp"),
mock.call("out2.put", "out2.put.tmp"),
],
any_order=True)
mock_remove.assert_has_calls([mock.call("out.put.tmp")])
mock_isfile.assert_called()
mock_mkdir.assert_not_called() # using suffix, not temp_dir
mock_copy.assert_has_calls(
[
mock.call("out.put", "out.put.tmp", follow_symlinks=False),
mock.call("out2.put", "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(subprocess, "call", 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(
transform), 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(subprocess, "call", return_value=0) as mock_call:
with mock.patch.object(output_cacher, "files_match",
return_value=True) as mock_match:
with mock.patch.object(os, "remove") as mock_remove:
self.assertEqual(
action.run_twice_with_substitution_and_compare_outputs(
transform), 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("out.put", "out.put.tmp")
mock_remove.assert_called_with("out.put.tmp")
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(subprocess, "call", 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(
transform), 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("out.put", "out.put.tmp")
mock_remove.assert_not_called()
def test_command_passed_and_some_outptus_differ(self):
def fake_match(file1: str, file2: str) -> bool:
if file1 == "out.put":
return True
elif file1 == "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(subprocess, "call", return_value=0) as mock_call:
with mock.patch.object(output_cacher, "files_match",
wraps=fake_match) as mock_match:
with mock.patch.object(os, "remove") as mock_remove:
self.assertEqual(
action.run_twice_with_substitution_and_compare_outputs(
transform), 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("out.put", "out.put.tmp"),
mock.call("out2.put", "out2.put.tmp"),
],
any_order=True)
mock_remove.assert_has_calls([mock.call("out.put.tmp")])
if __name__ == '__main__':
unittest.main()