build/tracer/output_cacher_test.py

#!/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 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_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")


class ReplaceTokensTest(unittest.TestCase):

    def test_no_change(self):
        self.assertEqual(
            output_cacher.replace_tokens(['echo', 'xyzzy'], lambda x: x),
            (['echo', 'xyzzy'], {}))

    def test_with_substitution(self):

        def transform(x):
            return x + x if x.startswith('x') else x

        self.assertEqual(
            output_cacher.replace_tokens(['echo', 'xyz', 'bar'], transform),
            (['echo', 'xyzxyz', 'bar'], {
                'xyz': 'xyzxyz'
            }))


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:
                    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:
                        output_cacher.move_if_different(
                            "source.txt", "dest.txt")
        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:
                        output_cacher.move_if_different(
                            "source.txt", "dest.txt")
        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:
                        output_cacher.move_if_different(
                            "source.txt", "dest.txt")
        mock_exists.assert_called()
        mock_diff.assert_called()
        mock_move.assert_not_called()
        mock_remove.assert_called_with("source.txt")


class ReplaceOutputArgsTest(unittest.TestCase):

    def test_command_failed(self):
        transform = output_cacher.TempFileTransform(suffix=".tmp")
        action = output_cacher.Action(command=["run.sh"], outputs={})
        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"], outputs={"out.put"})
        with mock.patch.object(subprocess, "call", return_value=0) as mock_call:
            with mock.patch.object(output_cacher,
                                   "move_if_different") 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"],
            outputs={"foo/out.put"})
        with mock.patch.object(subprocess, "call", return_value=0) as mock_call:
            with mock.patch.object(output_cacher,
                                   "move_if_different") 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"], outputs={})
        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"], outputs={"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"], outputs={"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"],
            outputs={"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"], outputs={})
        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"], outputs={"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"], outputs={"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"],
            outputs={"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()