| #!/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 textwrap |
| import unittest |
| from unittest import mock |
| |
| from typing import AbstractSet, Iterable |
| |
| import action_tracer |
| |
| |
| class ToolCommandTests(unittest.TestCase): |
| |
| def test_empty_command(self): |
| with self.assertRaises(IndexError): |
| action_tracer.ToolCommand().tool |
| |
| def test_command_no_args(self): |
| command = action_tracer.ToolCommand(tokens=['echo']) |
| self.assertEqual(command.env_tokens, []) |
| self.assertEqual(command.tool, 'echo') |
| self.assertEqual(command.args, []) |
| |
| def test_command_with_env(self): |
| command = action_tracer.ToolCommand(tokens=['TMPDIR=/my/tmp', 'echo']) |
| self.assertEqual(command.env_tokens, ['TMPDIR=/my/tmp']) |
| self.assertEqual(command.tool, 'echo') |
| self.assertEqual(command.args, []) |
| |
| def test_command_with_args(self): |
| command = action_tracer.ToolCommand( |
| tokens=['ln', '-f', '-s', 'foo', 'bar']) |
| self.assertEqual(command.env_tokens, []) |
| self.assertEqual(command.tool, 'ln') |
| self.assertEqual(command.args, ['-f', '-s', 'foo', 'bar']) |
| |
| def test_unwrap_no_change(self): |
| tokens = ['ln', '-f', '-s', 'foo', 'bar'] |
| command = action_tracer.ToolCommand(tokens=tokens) |
| self.assertEqual(command.unwrap().tokens, tokens) |
| |
| def test_unwrap_one_level(self): |
| tokens = ['wrapper', '--opt', 'foo', '--', 'bar.sh', 'arg'] |
| command = action_tracer.ToolCommand(tokens=tokens) |
| self.assertEqual(command.unwrap().tokens, ['bar.sh', 'arg']) |
| |
| def test_unwrap_one_of_many_level(self): |
| tokens = [ |
| 'wrapper', '--opt', 'foo', '--', 'bar.sh', 'arg', '--', 'inner.sh' |
| ] |
| command = action_tracer.ToolCommand(tokens=tokens) |
| command2 = command.unwrap() |
| self.assertEqual(command2.tokens, ['bar.sh', 'arg', '--', 'inner.sh']) |
| command3 = command2.unwrap() |
| self.assertEqual(command3.tokens, ['inner.sh']) |
| |
| |
| class IsKnownWrapperTests(unittest.TestCase): |
| |
| def test_action_tracer_is_not_wrapper(self): |
| command = action_tracer.ToolCommand( |
| tokens=[ |
| 'path/to/python3.x', 'path/to/not_a_wrapper.py', '--opt1', |
| 'arg1' |
| ]) |
| self.assertFalse(action_tracer.is_known_wrapper(command)) |
| |
| def test_action_tracer_is_not_wrapper_implicit_interpreter(self): |
| command = action_tracer.ToolCommand( |
| tokens=['path/to/not_a_wrapper.py', '--opt1', 'arg1']) |
| self.assertFalse(action_tracer.is_known_wrapper(command)) |
| |
| def test_action_tracer_is_wrapper(self): |
| command = action_tracer.ToolCommand( |
| tokens=[ |
| 'path/to/python3.x', 'path/to/action_tracer.py', '--', 'foo.sh', |
| 'arg1', 'arg2' |
| ]) |
| self.assertTrue(action_tracer.is_known_wrapper(command)) |
| |
| def test_action_tracer_is_wrapper_extra_python_flag(self): |
| command = action_tracer.ToolCommand( |
| tokens=[ |
| 'path/to/python3.x', '-S', 'path/to/action_tracer.py', '--', |
| 'foo.sh', 'arg1', 'arg2' |
| ]) |
| self.assertTrue(action_tracer.is_known_wrapper(command)) |
| |
| def test_action_tracer_is_wrapper_implicit_interpreter(self): |
| command = action_tracer.ToolCommand( |
| tokens=['path/to/action_tracer.py', '--', 'foo.sh', 'arg1', 'arg2']) |
| self.assertTrue(action_tracer.is_known_wrapper(command)) |
| |
| |
| class DepEdgesParseTests(unittest.TestCase): |
| |
| def test_invalid_input(self): |
| with self.assertRaises(ValueError): |
| action_tracer.parse_dep_edges( |
| "output.txt input1.txt") # missing ":" |
| |
| def test_output_only(self): |
| dep = action_tracer.parse_dep_edges("output.txt:") |
| self.assertEqual(dep.ins, set()) |
| self.assertEqual(dep.outs, {"output.txt"}) |
| |
| def test_multipl_outputs_only(self): |
| dep = action_tracer.parse_dep_edges("output.txt output2.txt :") |
| self.assertEqual(dep.ins, set()) |
| self.assertEqual(dep.outs, {"output.txt", "output2.txt"}) |
| |
| def test_output_with_one_input(self): |
| dep = action_tracer.parse_dep_edges("output.txt:input.cc") |
| self.assertEqual(dep.ins, {"input.cc"}) |
| self.assertEqual(dep.outs, {"output.txt"}) |
| |
| def test_output_with_multiple_inputs(self): |
| dep = action_tracer.parse_dep_edges( |
| "output.txt:input.cc includes/header.h") |
| self.assertEqual(dep.ins, {"input.cc", "includes/header.h"}) |
| self.assertEqual(dep.outs, {"output.txt"}) |
| |
| def test_output_with_multiple_inputs_unusual_spacing(self): |
| dep = action_tracer.parse_dep_edges( |
| " output.txt : input.cc includes/header.h ") |
| self.assertEqual(dep.ins, {"input.cc", "includes/header.h"}) |
| self.assertEqual(dep.outs, {"output.txt"}) |
| |
| def test_file_name_with_escaped_space(self): |
| dep = action_tracer.parse_dep_edges( |
| "output.txt: source\\ input.cc includes/header.h") |
| self.assertEqual(dep.ins, {"source input.cc", "includes/header.h"}) |
| self.assertEqual(dep.outs, {"output.txt"}) |
| |
| |
| class ParseDepFileTests(unittest.TestCase): |
| |
| def test_empty(self): |
| depfile = action_tracer.parse_depfile([]) |
| self.assertEqual(depfile.deps, []) |
| self.assertEqual(depfile.all_ins, set()) |
| self.assertEqual(depfile.all_outs, set()) |
| |
| def test_multiple_ins_multiple_outs(self): |
| depfile = action_tracer.parse_depfile(["a b: c d"]) |
| self.assertEqual( |
| depfile.deps, [ |
| action_tracer.DepEdges(ins={"c", "d"}, outs={"a", "b"}), |
| ]) |
| self.assertEqual(depfile.all_ins, {"c", "d"}) |
| self.assertEqual(depfile.all_outs, {"a", "b"}) |
| |
| def test_two_deps(self): |
| depfile = action_tracer.parse_depfile([ |
| "A: B", |
| "C: D E", |
| ]) |
| self.assertEqual( |
| depfile.deps, [ |
| action_tracer.DepEdges(ins={"B"}, outs={"A"}), |
| action_tracer.DepEdges(ins={"D", "E"}, outs={"C"}), |
| ]) |
| self.assertEqual(depfile.all_ins, {"B", "D", "E"}) |
| self.assertEqual(depfile.all_outs, {"A", "C"}) |
| |
| def test_continuation(self): |
| depfile = action_tracer.parse_depfile( |
| [ |
| "a \\\n", |
| "b: \\\n", |
| "c \\\n", |
| "d", |
| ]) |
| self.assertEqual( |
| depfile.deps, [ |
| action_tracer.DepEdges(ins={"c", "d"}, outs={"a", "b"}), |
| ]) |
| self.assertEqual(depfile.all_ins, {"c", "d"}) |
| self.assertEqual(depfile.all_outs, {"a", "b"}) |
| |
| def test_carriage_continuation(self): |
| depfile = action_tracer.parse_depfile( |
| [ |
| "a \\\r\n", |
| "b: c \\\r\n", |
| "d e", |
| ]) |
| self.assertEqual( |
| depfile.deps, [ |
| action_tracer.DepEdges(ins={"c", "d", "e"}, outs={"a", "b"}), |
| ]) |
| self.assertEqual(depfile.all_ins, {"c", "d", "e"}) |
| self.assertEqual(depfile.all_outs, {"a", "b"}) |
| |
| def test_space_in_filename(self): |
| depfile = action_tracer.parse_depfile(["a\\ b: c d"]) |
| self.assertEqual( |
| depfile.deps, [ |
| action_tracer.DepEdges(ins={"c", "d"}, outs={"a b"}), |
| ]) |
| self.assertEqual(depfile.all_ins, {"c", "d"}) |
| self.assertEqual(depfile.all_outs, {"a b"}) |
| |
| def test_consecutive_backslashes(self): |
| with self.assertRaises(ValueError): |
| depfile = action_tracer.parse_depfile(["a\\\\ b: c"]) |
| |
| def test_trailing_escaped_whitespace(self): |
| with self.assertRaises(ValueError): |
| depfile = action_tracer.parse_depfile([ |
| "a \\ \r\n", |
| "b: c", |
| ]) |
| with self.assertRaises(ValueError): |
| depfile = action_tracer.parse_depfile([ |
| "a \\ \n", |
| "b: c", |
| ]) |
| with self.assertRaises(ValueError): |
| depfile = action_tracer.parse_depfile([ |
| "a \\ \n", |
| "b: c", |
| ]) |
| |
| def test_unfinished_line_continuation(self): |
| with self.assertRaises(ValueError): |
| depfile = action_tracer.parse_depfile([ |
| "a \\\n", |
| "b: c \\\n", |
| ]) |
| |
| def test_blank_line(self): |
| depfile = action_tracer.parse_depfile([ |
| "a:b", |
| " ", |
| "b:", |
| ]) |
| self.assertEqual( |
| depfile.deps, [ |
| action_tracer.DepEdges(ins={"b"}, outs={"a"}), |
| action_tracer.DepEdges(ins=set(), outs={"b"}), |
| ]) |
| self.assertEqual(depfile.all_ins, {"b"}) |
| self.assertEqual(depfile.all_outs, {"a", "b"}) |
| |
| def test_comment(self): |
| depfile = action_tracer.parse_depfile([ |
| " # a:b", |
| "b:", |
| ]) |
| self.assertEqual( |
| depfile.deps, [ |
| action_tracer.DepEdges(ins=set(), outs={"b"}), |
| ]) |
| self.assertEqual(depfile.all_ins, set()) |
| self.assertEqual(depfile.all_outs, {"b"}) |
| |
| def test_continuation_blank_line(self): |
| with self.assertRaises(ValueError): |
| depfile = action_tracer.parse_depfile([ |
| "a: \\\n", |
| "", |
| "b", |
| ]) |
| |
| def test_continuation_comment(self): |
| with self.assertRaises(ValueError): |
| depfile = action_tracer.parse_depfile( |
| [ |
| "a: \\\n", |
| "# comment", |
| "b", |
| ]) |
| |
| |
| class ParseFsatraceOutputTests(unittest.TestCase): |
| |
| def test_empty_stream(self): |
| self.assertEqual( |
| list(action_tracer.parse_fsatrace_output([])), |
| [], |
| ) |
| |
| def test_ignore_malformed_line(self): |
| self.assertEqual( |
| list(action_tracer.parse_fsatrace_output(["invalid_line"])), |
| [], |
| ) |
| |
| def test_read(self): |
| self.assertEqual( |
| list(action_tracer.parse_fsatrace_output(["r|README.md"])), |
| [action_tracer.Read("README.md")], |
| ) |
| |
| def test_write(self): |
| self.assertEqual( |
| list(action_tracer.parse_fsatrace_output(["w|main.o"])), |
| [action_tracer.Write("main.o")], |
| ) |
| |
| def test_touch(self): |
| self.assertEqual( |
| list(action_tracer.parse_fsatrace_output(["t|file.stamp"])), |
| [action_tracer.Write("file.stamp")], |
| ) |
| |
| def test_delete(self): |
| self.assertEqual( |
| list(action_tracer.parse_fsatrace_output(["d|remove-me.tmp"])), |
| [action_tracer.Delete("remove-me.tmp")], |
| ) |
| |
| def test_move(self): |
| self.assertEqual( |
| list( |
| action_tracer.parse_fsatrace_output(["m|dest.txt|source.txt"])), |
| [ |
| action_tracer.Delete("source.txt"), |
| action_tracer.Write("dest.txt"), |
| ], |
| ) |
| |
| def test_sequence(self): |
| self.assertEqual( |
| list( |
| action_tracer.parse_fsatrace_output( |
| [ |
| "m|dest.txt|source.txt", |
| "r|input.txt", |
| "w|output.log", |
| ])), |
| [ |
| action_tracer.Delete("source.txt"), |
| action_tracer.Write("dest.txt"), |
| action_tracer.Read("input.txt"), |
| action_tracer.Write("output.log"), |
| ], |
| ) |
| |
| |
| class MatchConditionsTests(unittest.TestCase): |
| |
| def test_no_conditions(self): |
| self.assertFalse(action_tracer.MatchConditions().matches("foo/bar")) |
| |
| def test_prefix_matches(self): |
| self.assertTrue( |
| action_tracer.MatchConditions(prefixes={"fo"}).matches("foo/bar")) |
| |
| def test_suffix_matches(self): |
| self.assertTrue( |
| action_tracer.MatchConditions(suffixes={"ar"}).matches("foo/bar")) |
| |
| def test_component_matches(self): |
| self.assertTrue( |
| action_tracer.MatchConditions( |
| components={"bar", "bq"}).matches("foo/bar/baz.txt")) |
| |
| |
| class AccessShouldCheckTests(unittest.TestCase): |
| |
| def test_no_required_prefix(self): |
| ignore_conditions = action_tracer.MatchConditions() |
| self.assertTrue( |
| action_tracer.Read("book").should_check( |
| ignore_conditions=ignore_conditions, required_path_prefix="")) |
| self.assertTrue( |
| action_tracer.Write("block").should_check( |
| ignore_conditions=ignore_conditions, required_path_prefix="")) |
| |
| def test_required_prefix_matches(self): |
| ignore_conditions = action_tracer.MatchConditions() |
| prefix = "/home/project" |
| self.assertTrue( |
| action_tracer.Read("/home/project/book").should_check( |
| ignore_conditions=ignore_conditions, |
| required_path_prefix=prefix)) |
| self.assertTrue( |
| action_tracer.Write("/home/project/out/block").should_check( |
| ignore_conditions=ignore_conditions, |
| required_path_prefix=prefix)) |
| |
| def test_required_prefix_no_match(self): |
| ignore_conditions = action_tracer.MatchConditions() |
| prefix = "/home/project" |
| self.assertFalse( |
| action_tracer.Read("book").should_check( |
| ignore_conditions=ignore_conditions, |
| required_path_prefix=prefix)) |
| self.assertFalse( |
| action_tracer.Write("output/log").should_check( |
| ignore_conditions=ignore_conditions, |
| required_path_prefix=prefix)) |
| |
| def test_no_ignored_prefix(self): |
| ignore_conditions = action_tracer.MatchConditions(prefixes={}) |
| self.assertTrue( |
| action_tracer.Read("book").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertTrue( |
| action_tracer.Write("output/log").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| def test_ignored_prefix_matches(self): |
| ignore_conditions = action_tracer.MatchConditions(prefixes={"/tmp"}) |
| self.assertFalse( |
| action_tracer.Read("/tmp/book").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertFalse( |
| action_tracer.Write("/tmp/log").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| def test_ignored_prefix_no_match(self): |
| ignore_conditions = action_tracer.MatchConditions( |
| prefixes={"/tmp", "/no/look/here"}) |
| self.assertTrue( |
| action_tracer.Read("book").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertTrue( |
| action_tracer.Write("out/log").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| def test_no_ignored_suffix(self): |
| ignore_conditions = action_tracer.MatchConditions(suffixes={}) |
| self.assertTrue( |
| action_tracer.Read("book").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertTrue( |
| action_tracer.Write("output/log").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| def test_ignored_suffix_matches(self): |
| # e.g. from compiler --save-temps |
| ignore_conditions = action_tracer.MatchConditions(suffixes={".ii"}) |
| self.assertFalse( |
| action_tracer.Read("book.ii").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertFalse( |
| action_tracer.Write("tmp/log.ii").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| def test_ignored_suffix_no_match(self): |
| # e.g. from compiler --save-temps |
| ignore_conditions = action_tracer.MatchConditions( |
| suffixes={".ii", ".S"}) |
| self.assertTrue( |
| action_tracer.Read("book.txt").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertTrue( |
| action_tracer.Write("out/process.log").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| def test_ignored_path_components_no_match(self): |
| ignore_conditions = action_tracer.MatchConditions( |
| components={"__auto__", ".generated"}) |
| self.assertTrue( |
| action_tracer.Read("book").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertTrue( |
| action_tracer.Write("out/log").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| def test_ignored_path_components_matches(self): |
| ignore_conditions = action_tracer.MatchConditions( |
| components={"__auto__", ".generated"}) |
| self.assertFalse( |
| action_tracer.Read("library/__auto__/book").should_check( |
| ignore_conditions=ignore_conditions)) |
| self.assertFalse( |
| action_tracer.Write(".generated/out/log").should_check( |
| ignore_conditions=ignore_conditions)) |
| |
| |
| class CheckAccessAllowedTests(unittest.TestCase): |
| |
| def test_allowed_read(self): |
| self.assertTrue( |
| action_tracer.Read("foo.txt").allowed( |
| allowed_reads={"foo.txt"}, allowed_writes={})) |
| |
| def test_forbiddden_read(self): |
| self.assertFalse( |
| action_tracer.Read("bar.txt").allowed( |
| allowed_reads={}, allowed_writes={})) |
| |
| def test_allowed_write(self): |
| self.assertTrue( |
| action_tracer.Write("foo.txt").allowed( |
| allowed_reads={}, allowed_writes={"foo.txt"})) |
| |
| def test_forbiddden_write(self): |
| self.assertFalse( |
| action_tracer.Write("baz.txt").allowed( |
| allowed_reads={}, allowed_writes={})) |
| |
| def test_allowed_delete(self): |
| self.assertTrue( |
| action_tracer.Delete("foo.txt").allowed( |
| allowed_reads={}, allowed_writes={"foo.txt"})) |
| |
| def test_forbiddden_delete(self): |
| self.assertFalse( |
| action_tracer.Delete("baz.txt").allowed( |
| allowed_reads={}, allowed_writes={})) |
| |
| |
| class FormatAccessSetTest(unittest.TestCase): |
| |
| def test_empty(self): |
| self.assertEqual(str(action_tracer.FSAccessSet()), "[empty accesses]") |
| |
| def test_reads(self): |
| self.assertEqual( |
| str(action_tracer.FSAccessSet(reads={"c", "a", "b"})), |
| textwrap.dedent( |
| """\ |
| Reads: |
| a |
| b |
| c""")) |
| |
| def test_writes(self): |
| self.assertEqual( |
| str(action_tracer.FSAccessSet(writes={"e", "f", "d"})), |
| textwrap.dedent( |
| """\ |
| Writes: |
| d |
| e |
| f""")) |
| |
| def test_deletes(self): |
| self.assertEqual( |
| str(action_tracer.FSAccessSet(deletes={"r", "q", "p"})), |
| textwrap.dedent( |
| """\ |
| Deletes: |
| p |
| q |
| r""")) |
| |
| def test_reads_writes(self): |
| files = {"c", "a", "b"} |
| self.assertEqual( |
| str(action_tracer.FSAccessSet(reads=files, writes=files)), |
| textwrap.dedent( |
| """\ |
| Reads: |
| a |
| b |
| c |
| Writes: |
| a |
| b |
| c""")) |
| |
| def test_writes_deletes(self): |
| files = {"c", "a", "b"} |
| self.assertEqual( |
| str(action_tracer.FSAccessSet(writes=files, deletes=files)), |
| textwrap.dedent( |
| """\ |
| Writes: |
| a |
| b |
| c |
| Deletes: |
| a |
| b |
| c""")) |
| |
| |
| class FinalizeFileSystemAccessesTest(unittest.TestCase): |
| |
| def test_no_accesses(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses([]), |
| action_tracer.FSAccessSet()) |
| |
| def test_reads(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Read("r1.txt"), |
| action_tracer.Read("r2.txt"), |
| ]), action_tracer.FSAccessSet(reads={"r1.txt", "r2.txt"})) |
| |
| def test_writes(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Write("wb.txt"), |
| action_tracer.Write("wa.txt"), |
| ]), action_tracer.FSAccessSet(writes={"wa.txt", "wb.txt"})) |
| |
| def test_reads_writes_no_deletes(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Read("r2.txt"), |
| action_tracer.Write("wb.txt"), |
| action_tracer.Write("wa.txt"), |
| action_tracer.Read("r1.txt"), |
| ]), |
| action_tracer.FSAccessSet( |
| reads={"r1.txt", "r2.txt"}, writes={"wa.txt", "wb.txt"})) |
| |
| def test_read_after_write(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Write("temp.txt"), |
| action_tracer.Read("temp.txt"), |
| ]), action_tracer.FSAccessSet(reads=set(), writes={"temp.txt"})) |
| |
| def test_delete(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Delete("d1.txt"), |
| action_tracer.Delete("d2.txt"), |
| ]), action_tracer.FSAccessSet(deletes={"d1.txt", "d2.txt"})) |
| |
| def test_delete_after_write(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Write("temp.txt"), |
| action_tracer.Delete("temp.txt"), |
| ]), action_tracer.FSAccessSet()) |
| |
| def test_write_after_delete(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Delete("temp.txt"), |
| action_tracer.Write("temp.txt"), |
| ]), action_tracer.FSAccessSet(writes={"temp.txt"})) |
| |
| def test_write_read_delete(self): |
| self.assertEqual( |
| action_tracer.finalize_filesystem_accesses( |
| [ |
| action_tracer.Write("temp.txt"), |
| action_tracer.Read("temp.txt"), |
| action_tracer.Delete("temp.txt"), |
| ]), action_tracer.FSAccessSet()) |
| |
| |
| class CheckAccessPermissionsTests(unittest.TestCase): |
| |
| def test_no_accesses(self): |
| self.assertEqual( |
| action_tracer.check_access_permissions( |
| action_tracer.FSAccessSet(), action_tracer.AccessConstraints()), |
| action_tracer.FSAccessSet(), |
| ) |
| |
| def test_ok_read(self): |
| self.assertEqual( |
| action_tracer.check_access_permissions( |
| action_tracer.FSAccessSet(reads={"readable.txt"}), |
| action_tracer.AccessConstraints( |
| allowed_reads={"readable.txt"})), |
| action_tracer.FSAccessSet(), |
| ) |
| |
| def test_forbidden_read(self): |
| read = "unreadable.txt" |
| self.assertEqual( |
| action_tracer.check_access_permissions( |
| action_tracer.FSAccessSet(reads={read}), |
| action_tracer.AccessConstraints()), |
| action_tracer.FSAccessSet(reads={read}), |
| ) |
| |
| def test_ok_write(self): |
| self.assertEqual( |
| action_tracer.check_access_permissions( |
| action_tracer.FSAccessSet(writes={"writeable.txt"}), |
| action_tracer.AccessConstraints( |
| allowed_writes={"writeable.txt"})), |
| action_tracer.FSAccessSet(), |
| ) |
| |
| def test_forbidden_writes(self): |
| # make sure multiple violations accumulate |
| bad_writes = { |
| "unwriteable.txt", |
| "you-shall-not-pass.txt", |
| } |
| self.assertEqual( |
| action_tracer.check_access_permissions( |
| action_tracer.FSAccessSet(writes=bad_writes), |
| action_tracer.AccessConstraints()), |
| action_tracer.FSAccessSet(writes=bad_writes), |
| ) |
| |
| def test_read_from_temporary_writes_ok(self): |
| temp_file = "__file.tmp" |
| reads = {temp_file} |
| writes = { |
| "unwriteable.txt", |
| temp_file, |
| } |
| self.assertEqual( |
| action_tracer.check_access_permissions( |
| action_tracer.FSAccessSet(reads=reads, writes=writes), |
| action_tracer.AccessConstraints()), |
| action_tracer.FSAccessSet(reads=set(), writes=writes), |
| ) |
| |
| |
| class CheckMissingWritesTests(unittest.TestCase): |
| |
| def test_no_accesses(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes([], {}), |
| {}, |
| ) |
| |
| def test_only_reads(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes( |
| [action_tracer.Read("newspaper.pdf")], |
| {}, |
| ), |
| {}, |
| ) |
| |
| def test_excess_write(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes( |
| [action_tracer.Write("side-effect.txt")], |
| {}, |
| ), |
| {}, |
| ) |
| |
| def test_fulfilled_write(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes( |
| [action_tracer.Write("compiled.o")], |
| {"compiled.o"}, |
| ), |
| set(), |
| ) |
| |
| def test_missing_write(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes( |
| [], |
| {"write-me.out"}, |
| ), |
| {"write-me.out"}, |
| ) |
| |
| def test_missing_and_fulfilled_write(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes( |
| [action_tracer.Write("compiled.o")], |
| { |
| "write-me.out", |
| "compiled.o", |
| }, |
| ), |
| {"write-me.out"}, |
| ) |
| |
| def test_written_then_deleted(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes( |
| [ |
| action_tracer.Write("compiled.o"), |
| action_tracer.Delete("compiled.o"), |
| ], |
| {"compiled.o"}, |
| ), |
| {"compiled.o"}, |
| ) |
| |
| def test_deleted_then_written(self): |
| self.assertEqual( |
| action_tracer.check_missing_writes( |
| [ |
| action_tracer.Delete("compiled.o"), |
| action_tracer.Write("compiled.o"), |
| ], |
| {"compiled.o"}, |
| ), |
| set(), |
| ) |
| |
| |
| def abspaths(container: Iterable[str]) -> AbstractSet[str]: |
| return {os.path.abspath(f) for f in container} |
| |
| |
| class AccessConstraintsTests(unittest.TestCase): |
| |
| def test_empty_action(self): |
| action = action_tracer.Action(inputs=["script.sh"]) |
| self.assertEqual( |
| action.access_constraints(), |
| action_tracer.AccessConstraints( |
| allowed_reads=abspaths({"script.sh"}))) |
| |
| def test_have_inputs(self): |
| action = action_tracer.Action( |
| inputs=["script.sh", "input.txt", "main.cc"]) |
| self.assertEqual( |
| action.access_constraints(), |
| action_tracer.AccessConstraints( |
| allowed_reads=abspaths({"script.sh", "input.txt", "main.cc"}))) |
| |
| def test_have_outputs(self): |
| action = action_tracer.Action(inputs=["script.sh"], outputs=["main.o"]) |
| self.assertEqual( |
| action.access_constraints(), |
| action_tracer.AccessConstraints( |
| allowed_reads=abspaths({"script.sh", "main.o"}), |
| allowed_writes=abspaths({"main.o"}), |
| required_writes=abspaths({"main.o"}))) |
| |
| def test_have_depfile_writeable_inputs(self): |
| action = action_tracer.Action(inputs=["script.sh"], depfile="foo.d") |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| with mock.patch("builtins.open", mock.mock_open( |
| read_data="foo.o: foo.cc foo.h\n")) as mock_file: |
| constraints = action.access_constraints( |
| writeable_depfile_inputs=True) |
| mock_exists.assert_called_once() |
| mock_file.assert_called_once() |
| |
| self.assertEqual( |
| constraints, |
| action_tracer.AccessConstraints( |
| allowed_reads=abspaths( |
| {"script.sh", "foo.d", "foo.o", "foo.cc", "foo.h"}), |
| allowed_writes=abspaths({"foo.d", "foo.o", "foo.cc", "foo.h"}))) |
| |
| def test_have_depfile_nonwritable_inputs(self): |
| action = action_tracer.Action(inputs=["script.sh"], depfile="foo.d") |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| with mock.patch("builtins.open", mock.mock_open( |
| read_data="foo.o: foo.cc foo.h\n")) as mock_file: |
| constraints = action.access_constraints( |
| writeable_depfile_inputs=False) |
| mock_exists.assert_called_once() |
| mock_file.assert_called_once() |
| |
| self.assertEqual( |
| constraints, |
| action_tracer.AccessConstraints( |
| allowed_reads=abspaths( |
| {"script.sh", "foo.d", "foo.o", "foo.cc", "foo.h"}), |
| allowed_writes=abspaths({"foo.d", "foo.o"}))) |
| |
| def test_links_are_followed(self): |
| |
| def fake_realpath(s: str) -> str: |
| return f'test/realpath/{s}' |
| |
| action = action_tracer.Action(inputs=["script.sh"], depfile="foo.d") |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| with mock.patch("builtins.open", mock.mock_open( |
| read_data="foo.o: foo.cc foo.h\n")) as mock_file: |
| with mock.patch.object(os.path, 'realpath', |
| wraps=fake_realpath) as mock_realpath: |
| constraints = action.access_constraints( |
| writeable_depfile_inputs=False) |
| mock_exists.assert_called_once() |
| mock_file.assert_called_once() |
| mock_realpath.assert_called() |
| |
| self.assertEqual( |
| constraints, |
| action_tracer.AccessConstraints( |
| allowed_reads=abspaths( |
| { |
| "test/realpath/script.sh", |
| "test/realpath/foo.d", |
| "test/realpath/foo.o", |
| "test/realpath/foo.cc", |
| "test/realpath/foo.h", |
| }), |
| allowed_writes=abspaths({"foo.d", "foo.o"}))) |
| |
| def test_have_nonexistent_depfile(self): |
| action = action_tracer.Action(depfile="foo.d") |
| with mock.patch.object(os.path, 'exists', |
| return_value=False) as mock_exists: |
| constraints = action.access_constraints() |
| mock_exists.assert_called() |
| self.assertEqual( |
| constraints, |
| action_tracer.AccessConstraints( |
| allowed_writes=abspaths({"foo.d"}), |
| allowed_reads=abspaths({"foo.d"}))) |
| |
| |
| class DiagnoseStaleOutputsTest(unittest.TestCase): |
| |
| def test_no_accesses_no_constraints(self): |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[], |
| access_constraints=action_tracer.AccessConstraints(), |
| ) |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics(), |
| ) |
| |
| def test_missing_write_no_inputs(self): |
| required_writes = {"write.me"} |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[], |
| access_constraints=action_tracer.AccessConstraints( |
| required_writes=required_writes), |
| ) |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics( |
| required_writes=required_writes, |
| nonexistent_outputs={"write.me"}), |
| ) |
| |
| def test_missing_write_with_used_input(self): |
| used_input = "read.me" |
| required_writes = {"write.me"} |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[action_tracer.Read(used_input)], |
| access_constraints=action_tracer.AccessConstraints( |
| allowed_reads={used_input}, |
| required_writes=required_writes, |
| ), |
| ) |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics( |
| required_writes=required_writes, |
| nonexistent_outputs={"write.me"}), |
| ) |
| |
| def test_stale_output_no_inputs(self): |
| required_writes = {"write.me"} |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[], |
| access_constraints=action_tracer.AccessConstraints( |
| required_writes=required_writes), |
| ) |
| mock_exists.assert_called_once() |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics(required_writes=required_writes), |
| ) |
| |
| def test_stale_output_with_used_input(self): |
| |
| def fake_read_ctime(path: str): |
| if path.startswith("read"): |
| return 200 |
| raise ValueError(f'Unexpected path: {path}') |
| |
| def fake_write_ctime(path: str): |
| if path.startswith("write"): |
| return 100 |
| raise ValueError(f'Unexpected path: {path}') |
| |
| used_input = "read.me" |
| required_writes = {"write.me"} |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| with mock.patch.object(os.path, 'getctime', |
| wraps=fake_read_ctime) as mock_read_ctime: |
| with mock.patch.object( |
| action_tracer, 'realpath_ctime', |
| wraps=fake_write_ctime) as mock_write_ctime: |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[action_tracer.Read(used_input)], |
| access_constraints=action_tracer.AccessConstraints( |
| allowed_reads={used_input}, |
| required_writes=required_writes), |
| ) |
| mock_exists.assert_called_once() |
| mock_read_ctime.assert_called() |
| mock_write_ctime.assert_called() |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics( |
| required_writes=required_writes, |
| newest_input=used_input, |
| stale_outputs={"write.me"}), |
| ) |
| |
| def test_stale_output_with_multiple_used_inputs(self): |
| |
| def fake_read_ctime(path: str): |
| if path == "read.me": |
| return 200 |
| if path == "read.me.newer": |
| return 300 |
| raise Exception(f'fake_read_ctime for unexpected path: {path}') |
| |
| def fake_write_ctime(path: str): |
| if path.startswith("write"): |
| return 250 |
| raise Exception(f'fake_write_ctime for unexpected path: {path}') |
| |
| used_input = "read.me" |
| # Make sure the timestamp of the newest input is used for comparison. |
| used_input_newer = "read.me.newer" |
| required_writes = {"write.me"} |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| with mock.patch.object(os.path, 'getctime', |
| wraps=fake_read_ctime) as mock_read_ctime: |
| with mock.patch.object( |
| action_tracer, 'realpath_ctime', |
| wraps=fake_write_ctime) as mock_write_ctime: |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[ |
| action_tracer.Read(used_input), |
| action_tracer.Read(used_input_newer), |
| ], |
| access_constraints=action_tracer.AccessConstraints( |
| allowed_reads={used_input, used_input_newer}, |
| required_writes=required_writes), |
| ) |
| mock_exists.assert_called_once() |
| mock_read_ctime.assert_called() |
| mock_write_ctime.assert_called() |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics( |
| required_writes=required_writes, |
| # newer input is used for comparison |
| newest_input=used_input_newer, |
| stale_outputs={"write.me"}), |
| ) |
| |
| def test_fresh_output_with_used_input(self): |
| |
| def fake_getctime(path: str): |
| if path.startswith("read"): |
| return 100 |
| if path.startswith("write"): |
| return 200 |
| return 0 |
| |
| used_input = "read.me" |
| written_output = "write.me" |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| with mock.patch.object(os.path, 'getctime', |
| wraps=fake_getctime) as mock_ctime: |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[ |
| action_tracer.Read(used_input), |
| action_tracer.Write(written_output), |
| ], |
| access_constraints=action_tracer.AccessConstraints( |
| allowed_reads={used_input}, |
| required_writes={written_output}), |
| ) |
| # There are no untouched outputs, so getctime is never called. |
| mock_exists.assert_not_called() |
| mock_ctime.assert_not_called() |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics( |
| required_writes={written_output}, |
| # newest_input is not evaluated |
| stale_outputs=set(), |
| ), |
| ) |
| |
| def test_fresh_output_with_used_input_readable_output(self): |
| |
| def fake_getctime(path: str): |
| if path.startswith("read"): |
| return 100 |
| if path.startswith("write"): |
| return 200 |
| return 0 |
| |
| used_input = "read.me" |
| written_output = "write.me" |
| with mock.patch.object(os.path, 'exists', |
| return_value=True) as mock_exists: |
| with mock.patch.object(os.path, 'getctime', |
| wraps=fake_getctime) as mock_ctime: |
| output_diagnostics = action_tracer.diagnose_stale_outputs( |
| accesses=[ |
| action_tracer.Read(used_input), |
| action_tracer.Read(written_output), |
| action_tracer.Write(written_output), |
| ], |
| access_constraints=action_tracer.AccessConstraints( |
| allowed_reads={used_input, written_output}, |
| required_writes={written_output}), |
| ) |
| # There are no untouched outputs, so getctime is never called. |
| mock_exists.assert_not_called() |
| mock_ctime.assert_not_called() |
| self.assertEqual( |
| output_diagnostics, |
| action_tracer.StalenessDiagnostics( |
| required_writes={written_output}, |
| # newest_input is not evaluated |
| stale_outputs=set(), |
| ), |
| ) |
| |
| |
| class AllParentDirsTests(unittest.TestCase): |
| |
| def test_empty_path(self): |
| dirs = action_tracer.all_parent_dirs("") |
| self.assertEqual(dirs, set()) |
| |
| def test_empty_path_slash(self): |
| dirs = action_tracer.all_parent_dirs("/") |
| self.assertEqual(dirs, set()) |
| |
| def test_one_path(self): |
| dirs = action_tracer.all_parent_dirs("foo") |
| self.assertEqual(dirs, set()) |
| |
| def test_one_path_absolute(self): |
| dirs = action_tracer.all_parent_dirs("/foo") |
| self.assertEqual(dirs, set()) |
| |
| def test_subdir(self): |
| dirs = action_tracer.all_parent_dirs("parent/child") |
| self.assertEqual(dirs, {"parent"}) |
| |
| def test_subdir_absolute(self): |
| dirs = action_tracer.all_parent_dirs("/parent/child") |
| self.assertEqual(dirs, {"/parent"}) |
| |
| def test_subsubdir(self): |
| dirs = action_tracer.all_parent_dirs("matron/parent/child") |
| self.assertEqual(dirs, {"matron", "matron/parent"}) |
| |
| def test_subsubdir_absolute(self): |
| dirs = action_tracer.all_parent_dirs("/boss/parent/child") |
| self.assertEqual(dirs, {"/boss", "/boss/parent"}) |
| |
| |
| class DetectAllDirs(unittest.TestCase): |
| |
| def test_empty(self): |
| dirs = action_tracer.detect_all_dirs([]) |
| self.assertEqual(dirs, set()) |
| |
| def test_one(self): |
| dirs = action_tracer.detect_all_dirs(["/x/y/z"]) |
| self.assertEqual(dirs, {"/x", "/x/y"}) |
| |
| def test_different_parents(self): |
| dirs = action_tracer.detect_all_dirs(["/x/y/z", "/a/b/c"]) |
| self.assertEqual(dirs, {"/x", "/x/y", "/a", "/a/b"}) |
| |
| def test_common_parent(self): |
| dirs = action_tracer.detect_all_dirs(["/x/w/z", "/x/w/c"]) |
| self.assertEqual(dirs, {"/x", "/x/w"}) |
| |
| def test_common_grandparent(self): |
| dirs = action_tracer.detect_all_dirs(["/x/y/z", "/x/w/c"]) |
| self.assertEqual(dirs, {"/x", "/x/y", "/x/w"}) |
| |
| |
| class MainArgParserTests(unittest.TestCase): |
| |
| # These args are required, and there's nothing interesting about them to test. |
| required_args = "--trace-output t.out --label //pkg:tgt " |
| |
| def test_only_required_args(self): |
| parser = action_tracer.main_arg_parser() |
| args = parser.parse_args(self.required_args.split()) |
| self.assertEqual(args.trace_output, "t.out") |
| self.assertEqual(args.label, "//pkg:tgt") |
| # Make sure some checks are enabled by default |
| self.assertTrue(args.check_access_permissions) |
| |
| def test_check_access_permissions(self): |
| parser = action_tracer.main_arg_parser() |
| args = parser.parse_args( |
| (self.required_args + "--check-access-permissions").split()) |
| self.assertTrue(args.check_access_permissions) |
| |
| def test_no_check_access_permissions(self): |
| parser = action_tracer.main_arg_parser() |
| args = parser.parse_args( |
| (self.required_args + "--no-check-access-permissions").split()) |
| self.assertFalse(args.check_access_permissions) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
