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