# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.


import datetime
import logging
import os
import re
import sys
import warnings

import attr
from taskcluster.utils import fromNow
from taskgraph.util.keyed_by import evaluate_keyed_by
from taskgraph.util.treeherder import join_symbol
from taskgraph.util.verify import VerificationSequence

from gecko_taskgraph import GECKO
from gecko_taskgraph.util.attributes import (
    ALL_PROJECTS,
    RELEASE_PROJECTS,
    RUN_ON_PROJECT_ALIASES,
)
from gecko_taskgraph.util.constants import TEST_KINDS
from gecko_taskgraph.util.sparse_profiles import (
    is_path_covered_by_taskgraph_sparse_profile,
)

logger = logging.getLogger(__name__)
doc_base_path = os.path.join(GECKO, "taskcluster", "docs")


verifications = VerificationSequence()


@attr.s(frozen=True)
class DocPaths:
    _paths = attr.ib(factory=list)

    def get_files(self, filename):
        rv = []
        for p in self._paths:
            doc_path = os.path.join(p, filename)
            if os.path.exists(doc_path):
                rv.append(doc_path)
        return rv

    def add(self, path):
        """
        Projects that make use of Firefox's taskgraph can extend it with
        their own task kinds by registering additional paths for documentation.
        documentation_paths.add() needs to be called by the project's Taskgraph
        registration function. See taskgraph.config.
        """
        self._paths.append(path)


documentation_paths = DocPaths()
documentation_paths.add(doc_base_path)


def verify_docs(filename, identifiers, appearing_as):
    """
    Look for identifiers of the type appearing_as in the files
    returned by documentation_paths.get_files(). Firefox will have
    a single file in a list, but projects such as Thunderbird can have
    documentation in another location and may return multiple files.
    """
    # We ignore identifiers starting with '_' for the sake of tests.
    # Strings starting with "_" are ignored for doc verification
    # hence they can be used for faking test values
    doc_files = documentation_paths.get_files(filename)
    doctext = "".join([open(d).read() for d in doc_files])

    if appearing_as == "inline-literal":
        expression_list = [
            "``" + identifier + "``"
            for identifier in identifiers
            if not identifier.startswith("_")
        ]
    elif appearing_as == "heading":
        expression_list = [
            "\n" + identifier + "\n(?:(?:(?:-+\n)+)|(?:(?:.+\n)+))"
            for identifier in identifiers
            if not identifier.startswith("_")
        ]
    else:
        raise Exception(f"appearing_as = `{appearing_as}` not defined")

    for expression, identifier in zip(expression_list, identifiers):
        match_group = re.search(expression, doctext)
        if not match_group:
            raise Exception(
                f"{appearing_as}: `{identifier}` missing from doc file: `{filename}`"
            )


@verifications.add("initial")
def verify_run_using():
    from gecko_taskgraph.transforms.job import registry

    verify_docs(
        filename="transforms/job.rst",
        identifiers=registry.keys(),
        appearing_as="inline-literal",
    )


@verifications.add("parameters")
def verify_parameters_docs(parameters):
    if not parameters.strict:
        return

    parameters_dict = dict(**parameters)
    verify_docs(
        filename="parameters.rst",
        identifiers=list(parameters_dict),
        appearing_as="inline-literal",
    )


@verifications.add("kinds")
def verify_kinds_docs(kinds):
    verify_docs(filename="kinds.rst", identifiers=kinds.keys(), appearing_as="heading")


@verifications.add("full_task_set")
def verify_attributes(task, taskgraph, scratch_pad, graph_config, parameters):
    if task is None:
        verify_docs(
            filename="attributes.rst",
            identifiers=list(scratch_pad["attribute_set"]),
            appearing_as="heading",
        )
        return
    scratch_pad.setdefault("attribute_set", set()).update(task.attributes.keys())


@verifications.add("full_task_graph")
def verify_task_graph_symbol(task, taskgraph, scratch_pad, graph_config, parameters):
    """
    This function verifies that tuple
    (collection.keys(), machine.platform, groupSymbol, symbol) is unique
    for a target task graph.
    """
    if task is None:
        return
    task_dict = task.task
    if "extra" in task_dict:
        extra = task_dict["extra"]
        if "treeherder" in extra:
            treeherder = extra["treeherder"]

            collection_keys = tuple(sorted(treeherder.get("collection", {}).keys()))
            if len(collection_keys) != 1:
                raise Exception(
                    f"Task {task.label} can't be in multiple treeherder collections "
                    f"(the part of the platform after `/`): {collection_keys}"
                )
            platform = treeherder.get("machine", {}).get("platform")
            group_symbol = treeherder.get("groupSymbol")
            symbol = treeherder.get("symbol")

            key = (platform, collection_keys[0], group_symbol, symbol)
            if key in scratch_pad:
                raise Exception(
                    "Duplicate treeherder platform and symbol in tasks "
                    "`{}`and `{}`: {} {}".format(
                        task.label,
                        scratch_pad[key],
                        f"{platform}/{collection_keys[0]}",
                        join_symbol(group_symbol, symbol),
                    )
                )
            else:
                scratch_pad[key] = task.label


@verifications.add("full_task_graph")
def verify_trust_domain_v2_routes(
    task, taskgraph, scratch_pad, graph_config, parameters
):
    """
    This function ensures that any two tasks have distinct ``index.{trust-domain}.v2`` routes.
    """
    if task is None:
        return
    route_prefix = "index.{}.v2".format(graph_config["trust-domain"])
    task_dict = task.task
    routes = task_dict.get("routes", [])

    for route in routes:
        if route.startswith(route_prefix):
            if route in scratch_pad:
                raise Exception(
                    f"conflict between {task.label}:{scratch_pad[route]} for route: {route}"
                )
            else:
                scratch_pad[route] = task.label


@verifications.add("full_task_graph")
def verify_routes_notification_filters(
    task, taskgraph, scratch_pad, graph_config, parameters
):
    """
    This function ensures that only understood filters for notifications are
    specified.

    See: https://firefox-ci-tc.services.mozilla.com/docs/manual/using/task-notifications
    """
    if task is None:
        return
    route_prefix = "notify."
    valid_filters = (
        "on-any",
        "on-completed",
        "on-defined",
        "on-failed",
        "on-exception",
        "on-pending",
        "on-resolved",
        "on-running",
        "on-transition",
    )
    task_dict = task.task
    routes = task_dict.get("routes", [])

    for route in routes:
        if route.startswith(route_prefix):
            # Get the filter of the route
            route_filter = route.split(".")[-1]
            if route_filter not in valid_filters:
                raise Exception(
                    f"{task.label} has invalid notification filter ({route_filter})"
                )
            if route_filter == "on-any":
                warnings.warn(
                    DeprecationWarning(
                        f"notification filter '{route_filter}' is deprecated. Use "
                        "'on-transition' or 'on-resolved'."
                    )
                )


@verifications.add("full_task_graph")
def verify_dependency_tiers(task, taskgraph, scratch_pad, graph_config, parameters):
    tiers = scratch_pad
    if task is not None:
        tiers[task.label] = (
            task.task.get("extra", {}).get("treeherder", {}).get("tier", sys.maxsize)
        )
    else:

        def printable_tier(tier):
            if tier == sys.maxsize:
                return "unknown"
            return tier

        for current_task in taskgraph.tasks.values():
            tier = tiers[current_task.label]
            for d in current_task.dependencies.values():
                if taskgraph[d].task.get("workerType") == "always-optimized":
                    continue
                if "dummy" in taskgraph[d].kind:
                    continue
                if tier < tiers[d]:
                    raise Exception(
                        f"{current_task.label} (tier {printable_tier(tier)}) cannot depend on {d} (tier {printable_tier(tiers[d])})"
                    )


@verifications.add("full_task_graph")
def verify_toolchain_resources_in_sparse_profile(
    task, taskgraph, scratch_pad, graph_config, parameters
):
    """
    Verify that all toolchain resources are covered by the taskgraph sparse profile.
    If not, the decision task's sparse checkout won't have these files,
    causing incorrect hashes and breaking 'mach bootstrap' for developers.
    """
    if task is not None:
        if task.kind != "toolchain":
            return
        resources = task.attributes.get("toolchain-resources", [])
        uncovered = [
            f for f in resources if not is_path_covered_by_taskgraph_sparse_profile(f)
        ]
        if uncovered:
            uncovered_list = "\n".join(f"  path:{path}" for path in uncovered)
            scratch_pad.setdefault("errors", []).append(
                f"Toolchain '{task.label}' has resources not covered "
                f"by the taskgraph sparse profile.\n"
                f"Uncovered resources:\n{uncovered_list}"
            )
    else:
        errors = scratch_pad.get("errors", [])
        if errors:
            raise Exception(
                "Found toolchain resource(s) not covered by taskgraph sparse profile.\n"
                "This will cause incorrect hashes in the decision task.\n\n"
                + "\n\n".join(errors)
                + "\n\nTo fix, add the above path(s) to 'build/sparse-profiles/taskgraph'."
            )


@verifications.add("full_task_graph")
def verify_aliases(task, taskgraph, scratch_pad, graph_config, parameters):
    """
    This function verifies that aliases are not reused.
    """
    if task is None:
        return
    if task.kind not in ("toolchain", "fetch"):
        return
    for_kind = scratch_pad.setdefault(task.kind, {})
    aliases = for_kind.setdefault("aliases", {})
    alias_attribute = f"{task.kind}-alias"
    if task.label in aliases:
        raise Exception(
            f"Task `{aliases[task.label]}` has a {alias_attribute} of `{task.label[len(task.kind) + 1 :]}`, masking a task of that name."
        )
    labels = for_kind.setdefault("labels", set())
    labels.add(task.label)
    attributes = task.attributes
    if alias_attribute in attributes:
        keys = attributes[alias_attribute]
        if not keys:
            keys = []
        elif isinstance(keys, str):
            keys = [keys]
        for key in keys:
            full_key = f"{task.kind}-{key}"
            if full_key in labels:
                raise Exception(
                    f"Task `{task.label}` has a {alias_attribute} of `{key}`,"
                    " masking a task of that name."
                )
            if full_key in aliases:
                raise Exception(
                    f"Duplicate {alias_attribute} in tasks `{task.label}`and `{aliases[full_key]}`: {key}"
                )
            else:
                aliases[full_key] = task.label


@verifications.add("optimized_task_graph")
def verify_always_optimized(task, taskgraph, scratch_pad, graph_config, parameters):
    """
    This function ensures that always-optimized tasks have been optimized.
    """
    if task is None:
        return
    if task.task.get("workerType") == "always-optimized":
        raise Exception(f"Could not optimize the task {task.label!r}")


@verifications.add("full_task_graph", run_on_projects=RELEASE_PROJECTS)
def verify_shippable_no_sccache(task, taskgraph, scratch_pad, graph_config, parameters):
    if task and task.attributes.get("shippable"):
        if task.task.get("payload", {}).get("env", {}).get("USE_SCCACHE"):
            raise Exception(f"Shippable job {task.label} cannot use sccache")


@verifications.add("full_task_graph")
def verify_test_packaging(task, taskgraph, scratch_pad, graph_config, parameters):
    if task is None:
        # In certain cases there are valid reasons for tests to be missing,
        # don't error out when that happens.
        missing_tests_allowed = any((
            # user specified `--target-kind`
            bool(parameters.get("target-kinds")),
            # manifest scheduling is enabled
            parameters["test_manifest_loader"] != "default",
        ))

        test_env = parameters["try_task_config"].get("env", {})
        if test_env.get("MOZHARNESS_TEST_PATHS", "") or test_env.get(
            "MOZHARNESS_TEST_TAG", ""
        ):
            # This is sort of a hack, as we are filtering, we might filter out all test jobs
            missing_tests_allowed = True

        exceptions = []
        for current_task in taskgraph.tasks.values():
            if current_task.kind == "build" and not current_task.attributes.get(
                "skip-verify-test-packaging"
            ):
                build_env = current_task.task.get("payload", {}).get("env", {})
                package_tests = build_env.get("MOZ_AUTOMATION_PACKAGE_TESTS")
                shippable = current_task.attributes.get("shippable", False)
                build_has_tests = scratch_pad.get(current_task.label)

                if package_tests != "1":
                    # Shippable builds should always package tests.
                    if shippable:
                        exceptions.append(
                            f"Build job {current_task.label} is shippable and does not specify "
                            "MOZ_AUTOMATION_PACKAGE_TESTS=1 in the "
                            "environment."
                        )

                    # Build tasks in the scratch pad have tests dependent on
                    # them, so we need to package tests during build.
                    if build_has_tests:
                        exceptions.append(
                            f"Build job {current_task.label} has tests dependent on it and does not specify "
                            "MOZ_AUTOMATION_PACKAGE_TESTS=1 in the environment"
                        )
                # Build tasks that aren't in the scratch pad have no
                # dependent tests, so we shouldn't package tests.
                # With the caveat that we expect shippable jobs to always
                # produce tests.
                elif not build_has_tests and not shippable:
                    # If we have not generated all task kinds, we can't verify that
                    # there are no dependent tests.
                    if not missing_tests_allowed:
                        exceptions.append(
                            f"Build job {current_task.label} has no tests, but specifies "
                            f"MOZ_AUTOMATION_PACKAGE_TESTS={package_tests} in the environment. "
                            "Unset MOZ_AUTOMATION_PACKAGE_TESTS in the task definition "
                            "to fix."
                        )
        if exceptions:
            raise Exception("\n".join(exceptions))
        return
    if task.kind in TEST_KINDS:
        build_task = taskgraph[task.dependencies["build"]]
        scratch_pad[build_task.label] = 1


@verifications.add("full_task_graph")
def verify_run_known_projects(task, taskgraph, scratch_pad, graph_config, parameters):
    """Validates the inputs in run-on-projects.

    We should never let 'try' (or 'try-comm-central') be in run-on-projects even though it
    is valid because it is not considered for try pushes.  While here we also validate for
    other unknown projects or typos.
    """
    if task and task.attributes.get("run_on_projects"):
        projects = set(task.attributes["run_on_projects"])
        if {"try", "try-comm-central"} & set(projects):
            raise Exception(
                f"In task {task.label}: using try in run-on-projects is invalid; use try "
                "selectors to select this task on try"
            )
        # try isn't valid, but by the time we get here its not an available project anyway.
        valid_projects = ALL_PROJECTS | set(RUN_ON_PROJECT_ALIASES.keys())
        invalid_projects = projects - valid_projects
        if invalid_projects:
            raise Exception(
                f"Task '{task.label}' has an invalid run-on-projects value: "
                f"{invalid_projects}"
            )


@verifications.add("graph_config")
def verify_try_expiration_policies(graph_config):
    """We don't want any configuration leading to anything with an expiry longer
    than 28 days on try."""
    now = datetime.datetime.utcnow()
    cap = "28 days"
    cap_from_now = fromNow(cap, now)
    expiration_policy = evaluate_keyed_by(
        graph_config["expiration-policy"],
        "task expiration",
        {"project": "try", "level": "1"},
    )
    for policy, expires in expiration_policy.items():
        if fromNow(expires, now) > cap_from_now:
            raise Exception(
                f'expiration-policy "{policy}" ({expires}) is larger than {cap} for try'
            )