# Copyright 2023 The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Task to use autopkgtest in debusine."""
import logging
import re
import shlex
from pathlib import Path
from typing import Any, Optional

try:
    import pydantic.v1 as pydantic
except ImportError:
    import pydantic as pydantic  # type: ignore

from debusine import utils
from debusine.artifacts.local_artifact import AutopkgtestArtifact
from debusine.artifacts.models import (
    DebianAutopkgtest,
    DebianAutopkgtestResult,
    DebianAutopkgtestResultStatus,
    DebianAutopkgtestSource,
)
from debusine.tasks import (
    BaseTaskWithExecutor,
    RunCommandTask,
)
from debusine.tasks.models import AutopkgtestData


log = logging.getLogger(__name__)

# Use TypeAlias when not supporting Python 3.9 from bullseye
ParsedSummaryFile = dict[str, DebianAutopkgtestResult]


class Autopkgtest(
    RunCommandTask[AutopkgtestData],
    BaseTaskWithExecutor[AutopkgtestData],
):
    """Task to use autopkgtest in debusine."""

    TASK_VERSION = 1

    ARTIFACT_DIR = "artifact-dir"
    SUMMARY_FILE = "artifact-dir/summary"

    def __init__(self, task_data: dict[str, Any]) -> None:
        """Initialize object."""
        super().__init__(task_data)

        self._source_package_name: Optional[str] = None
        self._source_package_version: Optional[str] = None
        self._source_package_url: Optional[str] = None
        self._source_package_path: Optional[str] = None

        self._parsed: Optional[ParsedSummaryFile] = None
        self._autopkgtest_targets: list[Path] = []

    @classmethod
    def analyze_worker(cls):
        """Report metadata for this task on this worker."""
        metadata = super().analyze_worker()

        available_key = cls.prefix_with_task_name("available")
        metadata[available_key] = utils.is_command_available("autopkgtest")

        return metadata

    def can_run_on(self, worker_metadata: dict[str, Any]) -> bool:
        """Check if the specified worker can run the task."""
        if not super().can_run_on(worker_metadata):
            return False

        executor_available_key = f"executor:{self.backend}:available"
        available_key = self.prefix_with_task_name("available")
        return worker_metadata.get(
            executor_available_key, False
        ) and worker_metadata.get(available_key, False)

    def check_directory_for_consistency_errors(
        self, build_directory: Path  # noqa: U100
    ) -> list[str]:
        """Autopkgtest ARTIFACT_DIR/summary file does not exist."""
        summary_path = build_directory / self.SUMMARY_FILE
        if not summary_path.exists():
            return [f"'{self.SUMMARY_FILE}' does not exist"]

        return []

    @staticmethod
    def _parse_summary_file(
        summary_file: Path,
    ) -> dict[str, DebianAutopkgtestResult]:
        """
        Parse autopkgtest summary file (from autopkgtest --summary).

        :param summary_file: file to parse.
        :return: dictionary with the result. Structure:
            .. code-block::
                {"test-name-1": {"status": "PASS"},
                 "test-name-2: {"status": "FAIL", "details": "partial"},
                }

            "status": always in the dictionary (PASS, FAIL, FLAKY, SKIP or
            any other status written by autopkgtest)
            "details": only in the dictionary if details are found
            The test-name may be "*" if it wasn't known.
            If there's a testbed failure, the dictionary will be empty.
        :raises: ValueError if a line cannot be parsed
        """  # noqa: RST301
        parsed = {}
        with summary_file.open() as file:
            for line in file.readlines():
                line = line.rstrip()

                m = re.match(r"(?P<error>[a-z ]+): (?P<details>.*)", line)
                if m is not None:
                    log.info(
                        "Autopkgtest error: %s: %s",
                        m.group("error"),
                        m.group("details"),
                    )
                    continue

                m = re.match(
                    r"(?P<name>\S+)\s+(?P<status>\S+)"
                    r"(?:\s+(?P<details>.*))?",
                    line,
                )

                if m is None:
                    raise ValueError(f"Failed to parse line: {line}")

                name = m.group("name")
                result = m.group("status")

                if result not in ("PASS", "FAIL", "SKIP", "FLAKY"):
                    raise ValueError(f"Line with unexpected result: {line}")

                parsed[name] = DebianAutopkgtestResult(
                    status=DebianAutopkgtestResultStatus(result)
                )

                if details := m.group("details"):
                    parsed[name].details = details

        return parsed

    def fetch_input(self, destination: Path) -> bool:
        """Download the required artifacts."""
        data_input = self.data.input
        artifact = self.fetch_artifact(
            data_input.source_artifact_id, destination
        )

        for file_path, file_data in artifact.files.items():
            if file_path.endswith(".dsc"):
                self._source_package_url = file_data.url
                self._source_package_path = file_path
                break

        for artifact_id in data_input.binary_artifacts_ids:
            self.fetch_artifact(artifact_id, destination)

        for artifact_id in data_input.context_artifacts_ids:
            self.fetch_artifact(artifact_id, destination)

        return True

    def _cmdline(self) -> list[str]:
        """
        Return autopkgtest command line (idempotent).

        Use configuration of self.data.
        """
        if not self.executor:
            raise AssertionError(
                "self.executor not set - self._prepare_for_execution() "
                "must be called before _cmdline()"
            )

        cmd = [
            "autopkgtest",
            "--apt-upgrade",
            f"--output-dir={self.ARTIFACT_DIR}",
            f"--summary={self.SUMMARY_FILE}",
            "--no-built-binaries",
        ]

        for include_test in self.data.include_tests:
            cmd.append(f"--test-name={include_test}")

        for exclude_test in self.data.exclude_tests:
            cmd.append(f"--skip-test={exclude_test}")

        if debug_level := self.data.debug_level:
            cmd.append("-" + "d" * debug_level)

        for extra_apt_source in self.data.extra_apt_sources:
            cmd.append(f"--add-apt-source={extra_apt_source}")

        if self.data.use_packages_from_base_repository:
            release = self.executor.system_image.data["codename"]
            cmd.append(f"--apt-default-release={release}")

        for variable, value in self.data.environment.items():
            cmd.append(f"--env={variable}={value}")

        cmd.append(f"--needs-internet={self.data.needs_internet}")

        if self.data.timeout is not None:
            for key, timeout in self.data.timeout.dict(by_alias=True).items():
                if timeout is not None:
                    cmd.append(f"--timeout-{key}={timeout}")

        cmd.extend(map(str, self._autopkgtest_targets))

        cmd.append("--")
        cmd.append(self.executor.autopkgtest_virt_server())
        cmd.extend(self.executor.autopkgtest_virt_args())

        return cmd

    def configure_for_execution(
        self, download_directory: Path  # noqa: U100
    ) -> bool:
        """Gather information used later on (_cmdline(), upload_artifacts())."""
        # Not yet strictly guaranteed, because the source artifact might be
        # of the wrong type; see
        # https://salsa.debian.org/freexian-team/debusine/-/issues/207.
        assert self._source_package_path is not None

        # Used by upload_artifacts()
        dsc_file = download_directory / self._source_package_path
        dsc = utils.read_dsc(dsc_file)

        if dsc is None:
            self.append_to_log_file(
                "configure_for_execution.log",
                [f"{self._source_package_path} is not a valid .dsc file"],
            )
            return False

        self._source_package_name = dsc["source"]
        self._source_package_version = dsc["version"]

        # Used by _cmdline():
        self._autopkgtest_targets = utils.find_files_suffixes(
            download_directory, [".deb"]
        )
        self._autopkgtest_targets.append(dsc_file)

        self._prepare_executor()

        return True

    def task_succeeded(
        self, returncode: Optional[int], execute_directory: Path  # noqa: U100
    ) -> bool:
        """
        Parse the summary file and return success.

        Use self.data.fail_on.
        """
        self._parsed = self._parse_summary_file(
            execute_directory / self.SUMMARY_FILE
        )

        fail_on = self.data.fail_on

        for result in self._parsed.values():
            if (
                (result.status == "FAIL" and fail_on.failed_test)
                or (result.status == "FLAKY" and fail_on.flaky_test)
                or (result.status == "SKIP" and fail_on.skipped_test)
            ):
                return False

        return True

    def upload_artifacts(
        self, execute_directory: Path, *, execution_success: bool  # noqa: U100
    ):
        """Upload AutopkgtestArtifact with the files, data and relationships."""
        if not self.debusine:
            raise AssertionError("self.debusine not set")

        if not self.executor:
            raise AssertionError(
                "self.executor not set - self._prepare_for_execution() "
                "must be called before upload_artifacts()"
            )

        img_data = self.executor.system_image.data

        assert self._parsed is not None
        assert self._source_package_path is not None
        assert self._source_package_name is not None
        assert self._source_package_version is not None
        assert self._source_package_url is not None

        autopkgtest_artifact = AutopkgtestArtifact.create(
            execute_directory / self.ARTIFACT_DIR,
            DebianAutopkgtest(
                results=self._parsed,
                cmdline=shlex.join(self._cmdline()),
                source_package=DebianAutopkgtestSource(
                    name=self._source_package_name,
                    version=self._source_package_version,
                    url=pydantic.parse_obj_as(
                        pydantic.AnyUrl, self._source_package_url
                    ),
                ),
                architecture=self.data.host_architecture,
                distribution=f"{img_data['vendor']}:{img_data['codename']}",
            ),
        )

        self.debusine.upload_artifact(
            autopkgtest_artifact,
            workspace=self.workspace,
            work_request=self.work_request,
        )

    def run_cmd_succeeded(self, returncode: Optional[int]) -> bool:
        """Return True if autopkgtest have ran successfully."""
        return returncode not in {16, 20}
