# Copyright 2022 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Toolchain service tests."""


from collections import defaultdict
import os
from pathlib import Path
from typing import Dict, List, NamedTuple, Text
from unittest import mock

from chromite.lib import cros_test_lib
from chromite.lib.parser import package_info
from chromite.scripts import tricium_clang_tidy
from chromite.service import toolchain


class MockArtifact(NamedTuple):
    """Data for a Mocked Artifact."""

    linter: Text
    package: Text
    file_name: Text
    contents: Text


class MockBuildLinter(toolchain.BuildLinter):
    """Mocked version of Build Linters class."""

    def __init__(self, tempdir: Text, packages: List[Text] = None):
        super().__init__([], "")
        self.tempdir = tempdir
        self.packages = [] if packages is None else packages
        self.package_atoms = packages

        self.artifacts = defaultdict(lambda: defaultdict(list))
        self.artifacts_base = os.path.join(self.tempdir, "artifacts")

    def add_artifact(self, artifact: MockArtifact):
        """Adds a mock artifact and writes it to tempdir."""
        tmp_path = os.path.join(
            self.artifacts_base,
            artifact.linter,
            artifact.package,
            artifact.file_name,
        )

        os.makedirs(os.path.dirname(tmp_path), exist_ok=True)
        with open(tmp_path, "w", encoding="utf-8") as tmp_artifact_file:
            tmp_artifact_file.write(artifact.contents)
        self.artifacts[artifact.linter][artifact.package].append(tmp_path)

    def _fetch_from_linting_artifacts(self, subdir) -> Dict[Text, List[Text]]:
        """Get file from emerge artifact directory."""
        artifacts = {}
        for package, package_artifacts in self.artifacts[subdir].items():
            if not self.packages or package in self.package_atoms:
                if package not in artifacts:
                    artifacts[package] = []
                artifacts[package].extend(package_artifacts)
        return artifacts


class BuildLinterTests(cros_test_lib.MockTempDirTestCase):
    """Unit tests for Build Linter Class."""

    def checkArtifacts(
        self,
        expected_artifacts: List[MockArtifact],
        retrieved_artifact_paths: Dict[Text, List[Text]],
    ):
        """Asserts that artifact paths match the list of expected results."""

        actual_artifacts = []
        for paths in retrieved_artifact_paths.values():
            actual_artifacts.extend(paths)
        self.assertEqual(len(actual_artifacts), len(expected_artifacts))

        for artifact in expected_artifacts:
            for artifact_path in retrieved_artifact_paths:
                if artifact_path.endswith(
                    f"{artifact.linter}/{artifact.package}/{artifact.file_name}"
                ):
                    with open(
                        artifact_path, "r", encoding="utf-8"
                    ) as artifact_file:
                        contents = artifact_file.read()
                    self.assertEqual(contents, artifact.contents)

    def testMockBuildLinter(self):
        mbl = MockBuildLinter(self.tempdir, ["pkg_1", "pkg_2", "pkg_3"])
        relevant_artifacts = [
            MockArtifact("linter_1", "pkg_1", "out.txt", "Hello world"),
            MockArtifact("linter_1", "pkg_2", "out2.txt", "Hello world 2"),
            MockArtifact("linter_1", "pkg_3", "out.txt", "Hello\nWorld"),
        ]
        irrelevant_artifacts = [
            MockArtifact("linter_2", "pkg_2", "out.txt", "Goodbye World"),
            MockArtifact("linter_1", "pkg_4", "out.txt", "Goodbye world"),
            MockArtifact("linter_2", "pkg_5", "out.txt", "Goodbye world 2"),
            MockArtifact("linter_3", "pkg_6", "out.txt", "Goodbye\nWorld"),
            MockArtifact("linter_4", "pkg_1", "out.txt", "Goodbye\nWorld"),
        ]
        for artifact in relevant_artifacts + irrelevant_artifacts:
            mbl.add_artifact(artifact)

        # pylint: disable=protected-access
        retrieved_artifact_paths = mbl._fetch_from_linting_artifacts("linter_1")

        self.checkArtifacts(relevant_artifacts, retrieved_artifact_paths)

    def testMockBuildLinterNoPackages(self):
        mbl = MockBuildLinter(self.tempdir, [])
        relevant_artifacts = [
            MockArtifact("linter_1", "pkg_1", "out.txt", "Hello world"),
            MockArtifact("linter_1", "pkg_2", "out2.txt", "Hello world 2"),
            MockArtifact("linter_1", "pkg_3", "out.txt", "Hello\nWorld"),
            MockArtifact("linter_1", "pkg_4", "out.txt", "Hello world"),
        ]
        irrelevant_artifacts = [
            MockArtifact("linter_2", "pkg_2", "out.txt", "Goodbye World"),
            MockArtifact("linter_2", "pkg_5", "out.txt", "Goodbye world 2"),
            MockArtifact("linter_3", "pkg_6", "out.txt", "Goodbye\nWorld"),
            MockArtifact("linter_4", "pkg_1", "out.txt", "Goodbye\nWorld"),
        ]
        for artifact in relevant_artifacts + irrelevant_artifacts:
            mbl.add_artifact(artifact)

        # pylint: disable=protected-access
        retrieved_artifact_paths = mbl._fetch_from_linting_artifacts("linter_1")

        self.checkArtifacts(relevant_artifacts, retrieved_artifact_paths)

    def testFetchFromLintingArtifacts(self):
        bl = toolchain.BuildLinter(
            [
                package_info.parse("category0/package0"),
                package_info.parse("category1/package1"),
                package_info.parse("category1/package2"),
                package_info.parse("category2/package3"),
            ],
            self.tempdir,
        )

        bl_no_pkg = toolchain.BuildLinter([], self.tempdir)

        relevant_cases = [
            "category1/package1/linting-output/linter1/a.out",
            "category1/package1/linting-output/linter1/b.json",
            "category1/package1/linting-output/linter1/c",
            "category1/package2/linting-output/linter1/a.out",
            "category1/package2/linting-output/linter1/b.json",
            "category1/package2/linting-output/linter1/c",
            "category2/package3/linting-output/linter1/a.out",
            "category2/package3/linting-output/linter1/b.json",
            "category2/package3/linting-output/linter1/c",
        ]

        irrelevant_cases = [
            "category3/package1/linting-output/linter1/a.out",
            "category4/package2/linting-output/linter1/a.out",
            "category2/package5/linting-output/linter1/a.out",
            "category2/package6/linting-output/linter1/a.out",
            "category1/package1/linting-output/linter2/a.out",
            "category1/package2/linting-output/linter2/a.out",
            "category2/package3/linting-output/linter3/a.out",
            "category2/package4/linting-output/linter4/a.out",
        ]

        root = Path(self.tempdir) / "var/lib/chromeos/package-artifacts"

        expected_results = {
            "category1/package1": [
                f"{str(root)}/category1/package1/linting-output/linter1/a.out",
                f"{str(root)}/category1/package1/linting-output/linter1/b.json",
                f"{str(root)}/category1/package1/linting-output/linter1/c",
            ],
            "category1/package2": [
                f"{str(root)}/category1/package2/linting-output/linter1/a.out",
                f"{str(root)}/category1/package2/linting-output/linter1/b.json",
                f"{str(root)}/category1/package2/linting-output/linter1/c",
            ],
            "category2/package3": [
                f"{str(root)}/category2/package3/linting-output/linter1/a.out",
                f"{str(root)}/category2/package3/linting-output/linter1/b.json",
                f"{str(root)}/category2/package3/linting-output/linter1/c",
            ],
        }

        additional_no_pkg_results = {
            "category3/package1": [
                f"{str(root)}/category3/package1/linting-output/linter1/a.out"
            ],
            "category4/package2": [
                f"{str(root)}/category4/package2/linting-output/linter1/a.out"
            ],
            "category2/package5": [
                f"{str(root)}/category2/package5/linting-output/linter1/a.out"
            ],
            "category2/package6": [
                f"{str(root)}/category2/package6/linting-output/linter1/a.out"
            ],
        }

        expected_no_pkg_results = dict(
            expected_results, **additional_no_pkg_results
        )

        for case in relevant_cases + irrelevant_cases:
            test_path = root / case
            test_path.parent.mkdir(parents=True, exist_ok=True)
            test_path.touch()

        # pylint: disable=protected-access
        actual_results = bl._fetch_from_linting_artifacts("linter1")
        self.assertDictEqual(expected_results, actual_results)

        no_pkg_results = bl_no_pkg._fetch_from_linting_artifacts("linter1")
        self.assertDictEqual(expected_no_pkg_results, no_pkg_results)

    def testGetPackageForArtifactDir(self):
        bl = toolchain.BuildLinter([], "")

        test_cases = [
            ("category1", "package1", "linter1", "category1/package1"),
            ("category1", "package2", "linter2", "category1/package2"),
            ("category2", "package1", "linter3", "category2/package1"),
            ("category3", "package3", "linter4", "category3/package3"),
        ]

        root = Path(self.tempdir) / "var/lib/chromeos/package-artifacts"
        for case in test_cases:
            test_path = root / case[0] / case[1] / "linting-output" / case[2]
            expected_result = case[3]
            # pylint: disable=protected-access
            actual_result = bl._get_package_for_artifact_dir(test_path)
            self.assertEqual(expected_result, actual_result)

    def testFetchTidyLints(self):
        mock_calls = 0
        diagnostics = [
            tricium_clang_tidy.TidyDiagnostic("", 0, "lint1", "body1", [], []),
            tricium_clang_tidy.TidyDiagnostic("", 0, "lint2", "body2", [], []),
            tricium_clang_tidy.TidyDiagnostic("", 0, "lint3", "body3", [], []),
            tricium_clang_tidy.TidyDiagnostic("", 0, "lint4", "body4", [], []),
        ]
        expected_findings = [
            toolchain.LinterFinding(
                name="lint1",
                message="body1",
                locations=tuple(
                    [toolchain.CodeLocation("", "", 0, 0, None, None)]
                ),
                linter="clang_tidy",
                suggested_fixes=tuple(),
                package=package_info.parse("cat/pkg"),
            ),
            toolchain.LinterFinding(
                name="lint2",
                message="body2",
                locations=tuple(
                    [toolchain.CodeLocation("", "", 0, 0, None, None)]
                ),
                linter="clang_tidy",
                suggested_fixes=tuple(),
                package=package_info.parse("cat/pkg"),
            ),
            toolchain.LinterFinding(
                name="lint3",
                message="body3",
                locations=tuple(
                    [toolchain.CodeLocation("", "", 0, 0, None, None)]
                ),
                linter="clang_tidy",
                suggested_fixes=tuple(),
                package=package_info.parse("cat/pkg"),
            ),
            toolchain.LinterFinding(
                name="lint4",
                message="body4",
                locations=tuple(
                    [toolchain.CodeLocation("", "", 0, 0, None, None)]
                ),
                linter="clang_tidy",
                suggested_fixes=tuple(),
                package=package_info.parse("cat/pkg"),
            ),
        ]

        def mock_parse_tidy_invocation(_):
            nonlocal mock_calls
            mock_calls += 1
            if mock_calls == 1:
                meta = tricium_clang_tidy.InvocationMetadata(0, [], "", "", "")
                return meta, diagnostics[:2]
            if mock_calls == 2:
                meta = tricium_clang_tidy.InvocationMetadata(0, [], "", "", "")
                return meta, diagnostics[2:]
            if mock_calls == 3:
                meta = tricium_clang_tidy.InvocationMetadata(0, [], "", "", "")
                return meta, []
            elif mock_calls == 4:
                return tricium_clang_tidy.ExceptionData()
            elif mock_calls == 5:
                meta = tricium_clang_tidy.InvocationMetadata(1, [], "", "", "")
                return meta, diagnostics
            else:
                self.fail("Too many calls to parse_tidy_invocation")

        def mock_filter_findings(_, __, findings):
            if mock_calls == 1:
                self.assertCountEqual(findings, diagnostics[:2])
            elif mock_calls == 2:
                self.assertCountEqual(findings, diagnostics[2:])
            elif mock_calls == 3:
                self.assertEqual(findings, [])
            return findings

        mbl = MockBuildLinter(self.tempdir)

        artifacts = [
            MockArtifact("clang-tidy", "cat/pkg", "a.json", "content"),
            MockArtifact("clang-tidy", "cat/pkg", "b.json", "content"),
            MockArtifact("clang-tidy", "cat/pkg", "c.json", "content"),
            MockArtifact("clang-tidy", "cat/pkg", "d.json", "content"),
            MockArtifact("clang-tidy", "cat/pkg", "e.out", "content"),
        ]

        for artifact in artifacts:
            mbl.add_artifact(artifact)

        tmp_parse_tidy_invocation = tricium_clang_tidy.parse_tidy_invocation
        tricium_clang_tidy.parse_tidy_invocation = mock_parse_tidy_invocation
        tmp_filter_tidy_lints = tricium_clang_tidy.filter_tidy_lints
        tricium_clang_tidy.filter_tidy_lints = mock_filter_findings

        # pylint: disable=protected-access
        lints = mbl._fetch_tidy_lints()

        tricium_clang_tidy.parse_tidy_invocation = tmp_parse_tidy_invocation
        tricium_clang_tidy.filter_tidy_lints = tmp_filter_tidy_lints

        self.assertCountEqual(lints, expected_findings)

    def testParseIWYUFiles(self):
        mbl = MockBuildLinter(self.tempdir)

        artifacts = [
            MockArtifact(
                "iwyu",
                "cat/pkg",
                "iwyu.out",
                (
                    "\n".join(
                        [
                            "(/a/good/file.c has correct #includes/fwd-decls)",
                            "",
                            "/path/to/some/file.c should add these lines:",
                            "#include <missing1.h>  // for func1, func2, func3",
                            "#include <missing2.h>  // for func4, func5, func6",
                            "",
                            "/path/to/some/file.c should remove these lines:",
                            "- #include <unwanted.h>  // lines 11-11",
                            "",
                            "The full include-list for /path/to/some/file.c:",
                            "#include <missing1.h>  // for func1, func2, func3",
                            "#include <missing2.h>  // for func4, func5, func6",
                            "#include <stdint.h>",
                            "#include <stdlib.h>",
                            '#include "lib/foo.h"  // for func7, func8',
                            "---",
                        ]
                    )
                ),
            ),
            MockArtifact(
                "iwyu",
                "cat/pkg",
                "duplicate.out",
                (
                    "\n".join(
                        [
                            "/path/to/some/file.c should add these lines:",
                            "#include <missing1.h>  // for func1, func2, func3",
                            "#include <missing2.h>  // for func4, func5, func6",
                            "",
                            "/path/to/some/file.c should remove these lines:",
                            "- #include <unwanted.h>  // lines 11-11",
                            "",
                            "The full include-list for /path/to/some/file.c:",
                            "#include <missing1.h>  // for func1, func2, func3",
                            "#include <missing2.h>  // for func4, func5, func6",
                            "#include <stdint.h>",
                            "#include <stdlib.h>",
                            '#include "lib/foo.h"  // for func7, func8',
                            "---",
                        ]
                    )
                ),
            ),
            MockArtifact("IWYU", "cat/pkg", "empty.out", ""),
        ]

        expected_findings = [
            toolchain.LinterFinding(
                name="add",
                message="\n".join(
                    [
                        "Include list is missing:",
                        "\t#include <missing1.h>",
                        "Which is required for func1, func2, func3",
                        "",
                        "The full suggested include-list for this file:",
                        "#include <missing1.h>  // for func1, func2, func3",
                        "#include <missing2.h>  // for func4, func5, func6",
                        "#include <stdint.h>",
                        "#include <stdlib.h>",
                        '#include "lib/foo.h"  // for func7, func8',
                        "",
                        "Note: Suggestions from IWYU are not always correct and"
                        " thus require human supervision.",
                    ]
                ),
                locations=(
                    toolchain.CodeLocation(
                        "/path/to/some/file.c",
                        "",
                        line_start=1,
                        line_end=1,
                        col_start=None,
                        col_end=None,
                    ),
                ),
                linter="iwyu",
                suggested_fixes=tuple(),
                package=package_info.parse("cat/pkg"),
            ),
            toolchain.LinterFinding(
                name="add",
                message="\n".join(
                    [
                        "Include list is missing:",
                        "\t#include <missing2.h>",
                        "Which is required for func4, func5, func6",
                        "",
                        "The full suggested include-list for this file:",
                        "#include <missing1.h>  // for func1, func2, func3",
                        "#include <missing2.h>  // for func4, func5, func6",
                        "#include <stdint.h>",
                        "#include <stdlib.h>",
                        '#include "lib/foo.h"  // for func7, func8',
                        "",
                        "Note: Suggestions from IWYU are not always correct and"
                        " thus require human supervision.",
                    ]
                ),
                locations=(
                    toolchain.CodeLocation(
                        "/path/to/some/file.c",
                        "",
                        line_start=1,
                        line_end=1,
                        col_start=None,
                        col_end=None,
                    ),
                ),
                linter="iwyu",
                suggested_fixes=tuple(),
                package=package_info.parse("cat/pkg"),
            ),
            toolchain.LinterFinding(
                name="remove",
                message="\n".join(
                    [
                        "Remove from include list:",
                        "\t#include <unwanted.h>",
                        "",
                        "The full suggested include-list for this file:",
                        "#include <missing1.h>  // for func1, func2, func3",
                        "#include <missing2.h>  // for func4, func5, func6",
                        "#include <stdint.h>",
                        "#include <stdlib.h>",
                        '#include "lib/foo.h"  // for func7, func8',
                        "",
                        "Note: Suggestions from IWYU are not always correct and"
                        " thus require human supervision.",
                    ]
                ),
                locations=(
                    toolchain.CodeLocation(
                        "/path/to/some/file.c",
                        "",
                        line_start=11,
                        line_end=11,
                        col_start=None,
                        col_end=None,
                    ),
                ),
                linter="iwyu",
                suggested_fixes=tuple(),
                package=package_info.parse("cat/pkg"),
            ),
        ]

        for artifact in artifacts:
            mbl.add_artifact(artifact)

        # pylint: disable=protected-access
        parses = mbl._fetch_iwyu_lints()

        self.assertCountEqual(parses, expected_findings)


class TestEmergeAndUploadLints(cros_test_lib.RunCommandTestCase):
    """Unit tests for emerge_and_upload_lints"""

    def setUp(self):
        self.rc.AddCmdResult(
            [
                "lint_package",
                "--fetch-only",
                "--json",
                "--no-clippy",
                "--no-golint",
            ],
            stdout="linting output",
        )

    @mock.patch.object(toolchain.gs.GSContext, "CreateWithContents")
    def testEmergeAndUploadLints(self, copy_mock):
        used_gs_path = toolchain.emerge_and_upload_lints("atlas", 9999)
        target_gs_path = (
            "gs://chromeos-toolchain-artifacts/code-health/9999/atlas.json"
        )

        self.assertEqual(used_gs_path, target_gs_path)
        copy_mock.assert_called_with(target_gs_path, b"linting output")