# Copyright © 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.

"""View for Autopkgtest work request."""

import itertools
import re
from enum import StrEnum
from typing import Any, Optional, TypeAlias, assert_never, override

import pydantic
from more_itertools import peekable

from debusine.artifacts import AutopkgtestArtifact
from debusine.artifacts.models import ArtifactCategory, TaskTypes
from debusine.db.models import Artifact, FileInArtifact
from debusine.db.tasks import DBTask
from debusine.tasks.models import AutopkgtestData, AutopkgtestDynamicData
from debusine.web.views import view_utils
from debusine.web.views.artifacts import ArtifactPlugin
from debusine.web.views.files import FileWidget, LogFileWidget
from debusine.web.views.work_request import Tab, WorkRequestPlugin

DBAutopkgtest: TypeAlias = DBTask[AutopkgtestData, AutopkgtestDynamicData]


class AutopkgtestTab(Tab):
    """Autopkgtest specialized tab."""

    def __init__(self, task: DBAutopkgtest) -> None:
        """Define this tab."""
        super().__init__(name="autopkgtest", label="Autopkgtest")
        self.work_request = task.work_request
        self.task = task

    @staticmethod
    def _get_fail_on_scenarios(fail_on: dict[str, bool]) -> list[str]:
        """Process fail_on task data into a list of strings for viewing."""
        return [
            key.replace("_test", "")
            for key, value in fail_on.items()
            if key.endswith("_test") and value
        ]

    @staticmethod
    def _list_files(artifact: Artifact | None) -> list[str]:
        if artifact is not None:
            return [
                file.path
                for file in artifact.fileinartifact_set.filter(
                    complete=True
                ).order_by("path")
            ]
        else:
            return []

    @property
    def _source_artifact(self) -> dict[str, Any]:
        task_data = self.task.data
        source_artifact: dict[str, Any] = {
            "lookup": task_data.input.source_artifact,
            "id": None,
            "artifact": None,
            "files": [],
        }

        if (dynamic_data := self.task.dynamic_data) is not None:
            pk = dynamic_data.input_source_artifact_id
            source_artifact["id"] = pk
            try:
                source_artifact["artifact"] = artifact = Artifact.objects.get(
                    id=pk
                )
            except Artifact.DoesNotExist:
                pass
            else:
                source_artifact["files"] = self._list_files(artifact)
        return source_artifact

    def _get_request_data(self) -> dict[str, Any]:
        """Return data about the request."""
        task = self.task
        task_data = task.data
        dynamic_data = task.dynamic_data

        # TODO: It would be useful to show the request data after default
        # values have been set, but unfortunately these are only set by the
        # task after it loads data from the database and aren't saved
        # anywhere else.
        request_data: dict[str, Any] = {
            "source_artifact": self._source_artifact,
            "source_name": None,
            "source_version": None,
            "binary_artifacts": {
                "lookup": task_data.input.binary_artifacts.export(),
                "artifacts": [],
            },
            "context_artifacts": {
                "lookup": task_data.input.context_artifacts.export(),
                "artifacts": [],
            },
            "build_architecture": task_data.build_architecture,
            "vendor": None,
            "codename": None,
            "backend": task_data.backend,
            "include_tests": task_data.include_tests,
            "exclude_tests": task_data.exclude_tests,
            "debug_level": task_data.debug_level,
            "extra_repositories": task_data.extra_repositories,
            "use_packages_from_base_repository": (
                task_data.use_packages_from_base_repository
            ),
            "extra_environment": task_data.extra_environment,
            "needs_internet": task_data.needs_internet,
            "fail_on_scenarios": self._get_fail_on_scenarios(
                task_data.fail_on.model_dump()
            ),
            "timeout": task_data.timeout,
        }

        # This information is in the result artifact's data if it exists,
        # but it won't necessarily exist - for example, a testbed failure
        # will leave us without a result artifact.
        if dynamic_data is not None:
            try:
                environment_artifact = Artifact.objects.get(
                    id=dynamic_data.environment_id
                )
            except Artifact.DoesNotExist:
                pass
            else:
                if environment_artifact.category in {
                    ArtifactCategory.SYSTEM_TARBALL,
                    ArtifactCategory.SYSTEM_IMAGE,
                }:
                    request_data["vendor"] = environment_artifact.data.get(
                        "vendor"
                    )
                    request_data["codename"] = environment_artifact.data.get(
                        "codename"
                    )

            if source_artifact := request_data["source_artifact"]["artifact"]:
                if source_artifact.category == ArtifactCategory.SOURCE_PACKAGE:
                    request_data.update(
                        {
                            "source_name": source_artifact.data.get("name"),
                            "source_version": source_artifact.data.get(
                                "version"
                            ),
                        }
                    )

            binary_artifacts = {
                artifact.id: artifact
                for artifact in Artifact.objects.filter(
                    id__in=dynamic_data.input_binary_artifacts_ids
                )
            }
            for binary_artifact_id in dynamic_data.input_binary_artifacts_ids:
                binary_artifact = binary_artifacts.get(binary_artifact_id)
                request_data["binary_artifacts"]["artifacts"].append(
                    {
                        "id": binary_artifact_id,
                        "files": self._list_files(binary_artifact),
                        "artifact": binary_artifact,
                    }
                )

            context_artifacts = {
                artifact.id: artifact
                for artifact in Artifact.objects.filter(
                    id__in=dynamic_data.input_context_artifacts_ids
                )
            }
            for context_artifact_id in dynamic_data.input_context_artifacts_ids:
                context_artifact = context_artifacts.get(context_artifact_id)
                request_data["context_artifacts"]["artifacts"].append(
                    {
                        "id": context_artifact_id,
                        "files": self._list_files(context_artifact),
                    }
                )

        request_data["task_description"] = (
            f"{request_data['source_name'] or 'UNKNOWN'}_"
            f"{request_data['source_version'] or 'UNKNOWN'}_"
            f"{request_data['build_architecture']}"
        )
        if request_data["vendor"] and request_data["codename"]:
            request_data["task_description"] += (
                f" in {request_data['vendor']}:{request_data['codename']}"
            )

        return request_data

    @staticmethod
    def _get_result_data(
        artifact: Artifact,
    ) -> list[tuple[str, str, str | None]]:
        """Return a list of tuples describing individual test results."""
        result_data = []
        for name, result in sorted(artifact.data.get("results", {}).items()):
            result_data.append((name, result["status"], result.get("details")))
        return result_data

    @override
    def get_context_data(self) -> dict[str, Any]:
        result_artifact = (
            self.work_request.artifact_set.filter(
                category=AutopkgtestArtifact._category
            )
            .order_by("id")
            .first()
        )

        context_data: dict[str, Any] = {
            "request_data": self._get_request_data(),
            "result": self.work_request.result,
        }
        if result_artifact:
            context_data["result_artifact"] = result_artifact
            context_data["result_data"] = self._get_result_data(result_artifact)

        return context_data


class AutopkgtestViewWorkRequestPlugin(WorkRequestPlugin):
    """View for Autopkgtest work request."""

    task_type = TaskTypes.WORKER
    task_name = "autopkgtest"
    task: DBAutopkgtest

    @override
    def add_tabs(self, tabs: list[Tab]) -> None:
        tabs.insert(0, AutopkgtestTab(self.task))

    @override
    def get_context_data(self) -> dict[str, Any]:
        return {}


class AutopkgtestLogParser:
    """Parse Autopkgtest logs."""

    class Result(StrEnum):
        PASS = "PASS"
        FAIL = "FAIL"
        SKIP = "SKIP"

    class Section(pydantic.BaseModel):
        """Represents a logical section."""

        text: str
        color: str | None = (
            None  # bootstrap color: "primary", "secondary", "success"...
        )
        subsections: list["AutopkgtestLogParser.SubSection"] = []

    class SubSection(Section):
        """Represents a logical subsection with the test result."""

        line: int  # line number, 1-indexed
        result: Optional["AutopkgtestLogParser.Result"] = None
        expanded: bool = False  # subsection's contents is expanded or not
        include_in_table_of_contents: bool = True
        tooltip: str | None = None

    type TestNameToSubSections = dict[str, "AutopkgtestLogParser.SubSection"]

    _PREFIX = r"\s*\d+s\s+autopkgtest\s+\[\d\d:\d\d:\d\d\]:"
    _RE_STARTING = re.compile(rf"{_PREFIX}\s+starting date and time")
    _RE_APT_SOURCE = re.compile(rf"{_PREFIX}\s+@{{20}}\s+apt-source.*")
    _RE_TESTBED_SETUP = re.compile(rf"{_PREFIX}\s+@{{20}} test bed setup")
    _RE_TEST_PREPARING_TESTBED = re.compile(
        rf"{_PREFIX}\s+test\s+(.*):\s+preparing testbed"
    )
    _RE_ERROR = re.compile(rf"{_PREFIX}\s+ERROR:")
    _RE_TEST_STARTS = re.compile(rf"{_PREFIX}\s+test\s+(.*): ")
    _RE_TEST_RUN = re.compile(
        rf"{_PREFIX}\s+test\s+(.*): \[-----------------------"
    )
    _RE_SUMMARY = re.compile(rf"{_PREFIX}\s+@{{20}}\s+(summary)")
    _RE_RESULT_LINE = re.compile(r"^\s*\d+s\s+(\S+)\s*(PASS|FAIL|SKIP)")

    def __init__(self, log: bytes) -> None:
        """Initialize parser."""
        # splitlines() cannot be used: it's needed to separate by
        # "\n" only and splitlines() also separates by "\r" (e.g. the
        # autopkgtest logs contain a line with "\r" in the output of apt-get
        # The rstrip(b"\n") is to avoid an extra line at the end of the log
        # (splitlines() have the "keepends" parameter for this, but not split())
        self._text = log.decode(errors="replace").rstrip("\n")

    def _parse_preparation(self, lines: "peekable[tuple[int, str]]") -> Section:
        preparation = self.Section(text="Preparation", color="primary")

        for line_number, line in lines:
            # One of the two ways that the "prepare section" finishes is that
            # next line starts with a test
            (_, next_line) = lines.peek((None, ""))

            if self._RE_STARTING.match(line):
                preparation.subsections.append(
                    self.SubSection(
                        text="start run", line=line_number, expanded=True
                    )
                )
            elif self._RE_TESTBED_SETUP.match(line):
                preparation.subsections.append(
                    self.SubSection(text="test bed setup", line=line_number)
                )
            elif self._RE_APT_SOURCE.match(line):
                preparation.subsections.append(
                    self.SubSection(text="apt-source", line=line_number)
                )
                # there will not be more subsections for preparation
                break
            elif self._RE_ERROR.match(line):
                if preparation.subsections:
                    preparation.subsections[-1].color = "danger"
                    preparation.subsections[-1].expanded = True

            elif self._RE_TEST_STARTS.match(next_line):
                # preparation has finished
                break

        return preparation

    def _parse_tests(
        self, lines: "peekable[tuple[int, str]]"
    ) -> tuple[Section, TestNameToSubSections]:
        sections = self.Section(text="Tests", color=None)
        test_name_to_subsections = {}

        current_test_name = None

        for line_number, line in lines:
            if m := self._RE_TEST_PREPARING_TESTBED.match(line):
                sections.color = "primary"
                current_test_name = m.group(1)

                # include_in_table_of_contents: initially False (only for
                # the log section). But if the preparation fails,
                # it will be changed later on
                sub_section = self.SubSection(
                    text=f"preparing testbed {current_test_name}",
                    line=line_number,
                    include_in_table_of_contents=False,
                )
                sections.subsections.append(sub_section)

                test_name_to_subsections[current_test_name] = sub_section

            elif (
                self._RE_TEST_RUN.match(line) and current_test_name is not None
            ):
                sub_section = self.SubSection(
                    text=current_test_name,
                    line=line_number,
                    include_in_table_of_contents=True,
                )

                sections.subsections.append(sub_section)

                # The "preparing testbed {current_test_name}" not needed
                # to update it anymore
                test_name_to_subsections[current_test_name] = sub_section

            # The only way to know that we finished with the tests
            # is to check if next line is the "Summary"
            (_, line) = lines.peek((None, ""))
            if self._RE_SUMMARY.match(line):
                break

        return sections, test_name_to_subsections

    def _parse_closing(
        self, lines: "peekable[tuple[int, str]]"
    ) -> tuple[Section, dict[str, Result]]:
        closing = self.Section(text="Closing", color=None)
        test_name_to_result: dict[str, AutopkgtestLogParser.Result] = {}

        for line_number, line in lines:
            if self._RE_SUMMARY.match(line):
                closing.color = "primary"
                closing.subsections.append(
                    self.SubSection(
                        text="summary", line=line_number, expanded=True
                    )
                )

            if (m := self._RE_RESULT_LINE.match(line)) is not None:
                test_name = m.group(1)
                assert isinstance(test_name, str)
                test_name_to_result[test_name] = self.Result(m.group(2))

        return closing, test_name_to_result

    def parse(self) -> tuple[list[Section], int]:
        """Parse log file. Return (list[Section], total_lines)."""
        lines = self._text.split("\n")
        total_lines = len(lines)
        lines_peekable = peekable(enumerate(lines, start=1))

        preparation = self._parse_preparation(lines_peekable)
        sections, test_name_to_subsections = self._parse_tests(lines_peekable)
        closing, results = self._parse_closing(lines_peekable)

        Result = self.Result
        for test_name, test_result in results.items():
            subsection = test_name_to_subsections.get(test_name)

            if subsection is None:
                # In case a log file is inconsistent
                continue

            match test_result:
                case Result.PASS:
                    subsection.color = "success"
                    subsection.tooltip = "Pass"
                case Result.SKIP:
                    subsection.color = "secondary"
                    subsection.tooltip = "Skip"
                case Result.FAIL:
                    subsection.color = "danger"
                    subsection.tooltip = "Fail"

                    # The relevant subsection might be the
                    # "preparing testbed" which was not in the
                    # table of contents. Adds it now
                    subsection.expanded = True
                    subsection.include_in_table_of_contents = True
                case _ as unreachable:
                    assert_never(unreachable)

        return [preparation, sections, closing], total_lines


class AutopkgtestArtifactPlugin(ArtifactPlugin):
    """Plugin for AutopkgtestArtifact."""

    artifact_category = ArtifactCategory.AUTOPKGTEST
    template_name = "web/artifact-detail.html"
    name = "autopkgtest"

    object_name = "artifact"

    @override
    def get_context_data(self) -> dict[str, Any]:
        slug = "autopkgtest-log"

        specialized_tab = {
            "specialized_tab": {
                "label": "Autopkgtest log",
                "slug": slug,
                "template": "web/_autopkgtest-artifact-detail.html",
            }
        }

        try:
            log_file_in_artifact = self.artifact.fileinartifact_set.get(
                path="log"
            )
        except FileInArtifact.DoesNotExist:
            return {
                **specialized_tab,
                "autopkgtest_artifact_error": "Artifact does not contain log",
            }

        try:
            autopkgtest_contents = view_utils.read_file(
                log_file_in_artifact, max_file_size=50 * 1024 * 1024
            )
        except view_utils.FileInArtifactError as exc:
            return {
                **specialized_tab,
                "autopkgtest_artifact_error": exc.to_html(),
            }

        sections, total_lines = AutopkgtestLogParser(
            autopkgtest_contents
        ).parse()

        file_sections = []

        subsections = list(
            itertools.chain.from_iterable(
                log_section.subsections for log_section in sections
            )
        )

        for subsection, subsection_next in itertools.pairwise(subsections):
            file_sections.append(
                LogFileWidget.Section(
                    title=subsection.text,
                    start_line=subsection.line,
                    end_line=subsection_next.line - 1,
                    expanded=subsection.expanded,
                )
            )

        # Add the latest remaining section
        if subsections:
            last_subsection = subsections[-1]
            file_sections.append(
                LogFileWidget.Section(
                    title=last_subsection.text,
                    start_line=last_subsection.line,
                    end_line=total_lines,
                    expanded=last_subsection.expanded,
                )
            )

        widget = FileWidget.create(
            file_in_artifact=log_file_in_artifact,
            file_contents=autopkgtest_contents,
            sections=file_sections,
        )

        return {
            **specialized_tab,
            "sections": sections,
            "log_widget": widget,
        }
