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

"""Unit tests for Autopkgtest work request plugin."""

import textwrap
from typing import ClassVar, override
from unittest.mock import patch

from django.http import HttpResponseBase
from django.template.loader import get_template
from django.template.response import SimpleTemplateResponse

from debusine.artifacts import AutopkgtestArtifact
from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianAutopkgtest,
    DebianAutopkgtestSource,
)
from debusine.db.models import Artifact, WorkRequest, Workspace
from debusine.db.playground import scenarios
from debusine.tasks.models import (
    AutopkgtestData,
    AutopkgtestDynamicData,
    AutopkgtestInput,
    BackendType,
    LookupMultiple,
    LookupSingle,
)
from debusine.test.django import TestCase
from debusine.web.views import files
from debusine.web.views.autopkgtest import (
    AutopkgtestLogParser,
    AutopkgtestTab,
    AutopkgtestViewWorkRequestPlugin,
)
from debusine.web.views.files import LogFileWidget, TooBigFileWidget
from debusine.web.views.work_request import Tab


class AutopkgtestViewWorkRequestPluginTests(TestCase):
    """Tests for AutopkgtestViewWorkRequestPlugin."""

    environment_artifact: ClassVar[Artifact]
    source_artifact: ClassVar[Artifact]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.environment_artifact, _ = cls.playground.create_artifact(
            paths=["system.tar.xz"],
            category=ArtifactCategory.SYSTEM_TARBALL,
            data={
                "vendor": "debian",
                "codename": "bookworm",
                "architecture": "amd64",
                "variant": "",
            },
            create_files=True,
        )
        cls.source_artifact, _ = cls.playground.create_artifact(
            paths=["hello.dsc"],
            category=ArtifactCategory.SOURCE_PACKAGE,
            data={"name": "hello", "version": "1.0", "type": "dpkg"},
            create_files=True,
        )

    def get_tab(self, work_request: WorkRequest) -> AutopkgtestTab:
        """Return the tab for a work request or a plugin."""
        plugin = AutopkgtestViewWorkRequestPlugin(work_request.get_task())
        tabs: list[Tab] = []
        plugin.add_tabs(tabs)
        self.assertEqual(len(tabs), 1)
        tab = tabs[0]
        assert isinstance(tab, AutopkgtestTab)
        self.assertEqual(tab.name, "autopkgtest")
        self.assertEqual(tab.label, "Autopkgtest")
        return tab

    def create_work_request_for_artifacts(
        self,
        environment_artifact: Artifact,
        source_artifact: Artifact,
        binary_artifacts: list[Artifact],
        context_artifacts: list[Artifact] | None = None,
        workspace: Workspace | None = None,
    ) -> WorkRequest:
        """Create an autopkgtest work request for some artifacts."""
        environment_lookup: int | str
        if workspace is None:
            workspace = self.playground.get_default_workspace()
        if "vendor" in environment_artifact.data:
            environments_collection = self.playground.create_collection(
                environment_artifact.data["vendor"],
                CollectionCategory.ENVIRONMENTS,
                workspace=workspace,
            )
            environments_collection.manager.add_artifact(
                environment_artifact,
                user=self.playground.get_default_user(),
                variables={"backend": BackendType.UNSHARE},
            )
            environment_lookup = "{vendor}/match:codename={codename}".format(
                **environment_artifact.data
            )
        else:
            environment_lookup = environment_artifact.id
        task_data_input = AutopkgtestInput(
            source_artifact=source_artifact.id,
            binary_artifacts=LookupMultiple.model_validate(
                [binary_artifact.id for binary_artifact in binary_artifacts]
            ),
        )
        if context_artifacts:
            task_data_input.context_artifacts = LookupMultiple.model_validate(
                [context_artifact.id for context_artifact in context_artifacts]
            )

        return self.playground.create_worker_task(
            result=WorkRequest.Results.SUCCESS,
            task_data=AutopkgtestData(
                build_architecture="amd64",
                environment=environment_lookup,
                input=task_data_input,
            ),
            task_name="autopkgtest",
            workspace=workspace,
            use_data=False,
        )

    def create_autopkgtest_artifact(
        self, work_request: WorkRequest
    ) -> Artifact:
        """Create an AutopkgtestArtifact for a work request."""
        artifact, _ = self.playground.create_artifact(
            paths=["log", "summary"],
            work_request=work_request,
            create_files=True,
        )
        artifact.category = AutopkgtestArtifact._category
        artifact.data = {
            "results": {
                "command1": {"status": "PASS"},
                "command2": {"status": "FAIL", "details": "partial"},
            },
            "cmdline": "/usr/bin/hello",
            "source_package": {
                "name": "hello",
                "version": "1.0",
                "url": "https://example.org/hello.deb",
            },
            "architecture": "amd64",
            "distribution": "debian:sid",
        }
        artifact.save()
        return artifact

    def test_is_enabled(self) -> None:
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, self.source_artifact, []
        )
        autopkgtest_view = AutopkgtestViewWorkRequestPlugin(
            work_request.get_task()
        )
        self.assertTrue(autopkgtest_view.is_enabled())

    def test_get_context_data_simple(self) -> None:
        """get_context_data returns the expected data for a simple case."""
        binary_artifact_1, _ = self.playground.create_artifact(
            paths=["hello-1.deb"], create_files=True
        )
        binary_artifact_2, _ = self.playground.create_artifact(
            paths=["hello-2.deb"], create_files=True
        )
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact,
            self.source_artifact,
            [binary_artifact_1, binary_artifact_2],
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=self.environment_artifact.id,
            input_source_artifact_id=self.source_artifact.id,
            input_binary_artifacts_ids=[
                binary_artifact_1.id,
                binary_artifact_2.id,
            ],
        ).model_dump()
        work_request.save()
        result_artifact = self.create_autopkgtest_artifact(work_request)
        context_data = self.get_tab(work_request).get_context_data()
        self.assertEqual(
            context_data,
            {
                "request_data": {
                    "backend": BackendType.AUTO,
                    "binary_artifacts": {
                        "lookup": [binary_artifact_1.id, binary_artifact_2.id],
                        "artifacts": [
                            {
                                "files": ["hello-1.deb"],
                                "id": binary_artifact_1.id,
                                "artifact": binary_artifact_1,
                            },
                            {
                                "files": ["hello-2.deb"],
                                "id": binary_artifact_2.id,
                                "artifact": binary_artifact_2,
                            },
                        ],
                    },
                    "build_architecture": "amd64",
                    "codename": "bookworm",
                    "context_artifacts": {"lookup": [], "artifacts": []},
                    "debug_level": 0,
                    "exclude_tests": [],
                    "extra_repositories": None,
                    "extra_environment": {},
                    "fail_on_scenarios": ["failed"],
                    "include_tests": [],
                    "needs_internet": "run",
                    "source_artifact": {
                        "files": ["hello.dsc"],
                        "id": self.source_artifact.id,
                        "lookup": self.source_artifact.id,
                        "artifact": self.source_artifact,
                    },
                    "source_name": "hello",
                    "source_version": "1.0",
                    "task_description": "hello_1.0_amd64 in debian:bookworm",
                    "timeout": None,
                    "use_packages_from_base_repository": False,
                    "vendor": "debian",
                },
                "result": work_request.result,
                "result_artifact": result_artifact,
                "result_data": [
                    ("command1", "PASS", None),
                    ("command2", "FAIL", "partial"),
                ],
            },
        )

    def test_get_context_data_with_context_artifacts(self) -> None:
        """Any context artifacts show up in the context data."""
        context_artifact_1, _ = self.playground.create_artifact(
            paths=["hello-rdep-1.deb"], create_files=True
        )
        context_artifact_2, _ = self.playground.create_artifact(
            paths=["hello-rdep-2.deb"], create_files=True
        )
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact,
            self.source_artifact,
            [],
            context_artifacts=[context_artifact_1, context_artifact_2],
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=self.environment_artifact.id,
            input_source_artifact_id=self.source_artifact.id,
            input_binary_artifacts_ids=[],
            input_context_artifacts_ids=[
                context_artifact_1.id,
                context_artifact_2.id,
            ],
        ).model_dump()
        work_request.save()
        context_data = self.get_tab(work_request).get_context_data()

        self.assertEqual(
            context_data["request_data"]["context_artifacts"],
            {
                "lookup": [context_artifact_1.id, context_artifact_2.id],
                "artifacts": [
                    {
                        "files": ["hello-rdep-1.deb"],
                        "id": context_artifact_1.id,
                    },
                    {
                        "files": ["hello-rdep-2.deb"],
                        "id": context_artifact_2.id,
                    },
                ],
            },
        )

    def test_context_data_no_artifacts(self) -> None:
        """If there are no output artifacts, context data omits that info."""
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, self.source_artifact, []
        )
        context_data = self.get_tab(work_request).get_context_data()
        self.assertCountEqual(context_data.keys(), {"request_data", "result"})

    def test_context_data_no_dynamic_task_data(self) -> None:
        """If there is no dynamic data, context data omits artifact info."""
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, self.source_artifact, []
        )
        context_data = self.get_tab(work_request).get_context_data()
        self.assertIsNone(context_data["request_data"]["codename"])
        self.assertIsNone(context_data["request_data"]["vendor"])
        self.assertEqual(
            context_data["request_data"]["task_description"],
            "UNKNOWN_UNKNOWN_amd64",
        )
        self.assertEqual(
            context_data["request_data"]["source_artifact"],
            {
                "lookup": self.source_artifact.id,
                "id": None,
                "files": [],
                "artifact": None,
            },
        )

    def test_context_data_missing_environment_artifact(self) -> None:
        """If the environment artifact is missing, context data omits it."""
        environment_artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.SYSTEM_TARBALL
        )
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            environment_artifact, self.source_artifact, []
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=environment_artifact.id,
            input_source_artifact_id=self.source_artifact.id,
            input_binary_artifacts_ids=[],
        ).model_dump()
        work_request.save()
        environment_artifact.delete()
        context_data = self.get_tab(work_request).get_context_data()
        self.assertIsNone(context_data["request_data"]["codename"])
        self.assertIsNone(context_data["request_data"]["vendor"])
        self.assertEqual(
            context_data["request_data"]["task_description"], "hello_1.0_amd64"
        )

    def test_context_data_environment_artifact_wrong_category(self) -> None:
        """We cope with the environment artifact having an odd category."""
        environment_artifact, _ = self.playground.create_artifact(
            paths=["system.tar.xz"],
            category=ArtifactCategory.TEST,
            create_files=True,
        )
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            environment_artifact, self.source_artifact, []
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=environment_artifact.id,
            input_source_artifact_id=self.source_artifact.id,
            input_binary_artifacts_ids=[],
        ).model_dump()
        work_request.save()
        tab = self.get_tab(work_request)
        context_data = tab.get_context_data()
        self.assertIsNone(context_data["request_data"]["codename"])
        self.assertIsNone(context_data["request_data"]["vendor"])
        self.assertEqual(
            context_data["request_data"]["task_description"], "hello_1.0_amd64"
        )

    def test_context_data_missing_source_artifact(self) -> None:
        """If the source artifact is missing, context data omits its files."""
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, self.source_artifact, []
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=self.environment_artifact.id,
            input_source_artifact_id=self.source_artifact.id,
            input_binary_artifacts_ids=[],
        ).model_dump()
        work_request.save()
        source_artifact_id = self.source_artifact.id
        self.source_artifact.files.clear()
        self.source_artifact.delete()
        context_data = self.get_tab(work_request).get_context_data()
        self.assertEqual(
            context_data["request_data"]["source_artifact"],
            {
                "lookup": source_artifact_id,
                "id": source_artifact_id,
                "artifact": None,
                "files": [],
            },
        )

    def test_context_data_source_artifact_wrong_category(self) -> None:
        """We cope with the source artifact having an odd category."""
        self.source_artifact.category = ArtifactCategory.TEST
        self.source_artifact.save()
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, self.source_artifact, []
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=self.environment_artifact.id,
            input_source_artifact_id=self.source_artifact.id,
            input_binary_artifacts_ids=[],
        ).model_dump()
        work_request.save()
        context_data = self.get_tab(work_request).get_context_data()
        self.assertIsNone(context_data["request_data"]["source_name"])
        self.assertIsNone(context_data["request_data"]["source_version"])
        self.assertEqual(
            context_data["request_data"]["task_description"],
            "UNKNOWN_UNKNOWN_amd64 in debian:bookworm",
        )

    def test_context_data_missing_binary_artifact(self) -> None:
        """If a binary artifact is missing, context data omits its files."""
        binary_artifact_1, _ = self.playground.create_artifact(
            paths=["hello-1.deb"], create_files=True
        )
        binary_artifact_2, _ = self.playground.create_artifact(
            paths=["hello-2.deb"], create_files=True
        )
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, self.source_artifact, []
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=self.environment_artifact.id,
            input_source_artifact_id=self.source_artifact.id,
            input_binary_artifacts_ids=[
                binary_artifact_1.id,
                binary_artifact_2.id,
            ],
        ).model_dump()
        work_request.save()
        binary_artifact_1_id = binary_artifact_1.id
        binary_artifact_1.files.clear()
        binary_artifact_1.delete()
        tab = self.get_tab(work_request)
        context_data = tab.get_context_data()
        self.assertEqual(
            context_data["request_data"]["binary_artifacts"],
            {
                "lookup": work_request.task_data["input"]["binary_artifacts"],
                "artifacts": [
                    {
                        "id": binary_artifact_1_id,
                        "files": [],
                        "artifact": None,
                    },
                    {
                        "id": binary_artifact_2.id,
                        "files": ["hello-2.deb"],
                        "artifact": binary_artifact_2,
                    },
                ],
            },
        )

    def test_get_context_data_skips_incomplete_files(self) -> None:
        """get_context_data() skips incomplete files."""
        source_artifact, _ = self.playground.create_artifact(
            paths=["hello.dsc"],
            category=ArtifactCategory.SOURCE_PACKAGE,
            data={"name": "hello", "version": "1.0", "type": "dpkg"},
            create_files=True,
            skip_add_files_in_store=True,
        )
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, source_artifact, []
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=self.environment_artifact.id,
            input_source_artifact_id=source_artifact.id,
            input_binary_artifacts_ids=[],
        ).model_dump()
        work_request.save()

        context_data = self.get_tab(work_request).get_context_data()
        self.assertEqual(
            context_data["request_data"]["source_artifact"]["files"], []
        )

    def assert_contains_artifact_information(
        self,
        response: HttpResponseBase,
        description: str,
        artifact_lookup: LookupSingle,
        artifact_id: int,
    ) -> None:
        """Assert contains information of an artifact: link and files."""
        artifact = Artifact.objects.get(id=artifact_id)
        files = artifact.fileinartifact_set.order_by("path").values_list(
            "path", flat=True
        )
        artifact_path = artifact.get_absolute_url()
        artifact_link = f'<a href="{artifact_path}">#{artifact_id}</a>'
        files_li = "".join([f"<li>{file}</li>" for file in files])
        self.assertContains(
            response,
            f"<li>{description} ({artifact_lookup}: {artifact_link})"
            f"<ul>{files_li}</ul>"
            f"</li>",
            html=True,
        )

    def assert_contains_artifacts_information(
        self,
        response: HttpResponseBase,
        description: str,
        artifact_lookup: LookupMultiple,
        artifact_ids: list[int],
    ) -> None:
        """Assert contains information of a set of artifacts."""
        expected_response = (
            f"<li>{description} ({artifact_lookup.export()})<ul>"
        )
        for artifact_id in artifact_ids:
            artifact = Artifact.objects.get(id=artifact_id)
            files = artifact.fileinartifact_set.order_by("path").values_list(
                "path", flat=True
            )
            artifact_path = artifact.get_absolute_url()
            artifact_link = f'<a href="{artifact_path}">#{artifact_id}</a>'
            files_li = "".join([f"<li>{file}</li>" for file in files])
            expected_response += f"<li>{artifact_link}<ul>{files_li}</ul></li>"
        expected_response += "</ul></li>"
        self.assertContains(response, expected_response, html=True)

    def test_template_output(self) -> None:
        """Generic output of the template."""
        workspace = self.playground.create_workspace(name="Public", public=True)
        environment = self.playground.create_debian_environment(
            workspace=workspace
        )
        assert environment.artifact
        source_artifact = self.playground.create_source_artifact(
            workspace=workspace
        )
        binary_artifact_1 = (
            self.playground.create_minimal_binary_package_artifact(
                package="hello-1"
            )
        )
        binary_artifact_2 = (
            self.playground.create_minimal_binary_package_artifact(
                package="hello-2"
            )
        )
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            environment.artifact,
            source_artifact,
            [binary_artifact_1, binary_artifact_2],
            workspace=workspace,
        )
        work_request.dynamic_task_data = AutopkgtestDynamicData(
            environment_id=environment.artifact.id,
            input_source_artifact_id=source_artifact.id,
            input_binary_artifacts_ids=[
                binary_artifact_1.id,
                binary_artifact_2.id,
            ],
        ).model_dump()
        work_request.save()
        self.create_autopkgtest_artifact(work_request)

        response = self.client.get(work_request.get_absolute_url())
        tree = self.assertResponseHTML(response, dump_on_error=True)
        title = self.assertHasElement(tree, "//head/title")
        self.assertTextContentEqual(
            title, f"debusine - Public - #{work_request.id}: autopkgtest"
        )

        result_output = get_template("web/_work_request-result.html").render(
            {"result": work_request.result}
        )
        self.assertContains(response, f"Result: {result_output}", html=True)

        self.assert_contains_artifact_information(
            response, "Source artifact", source_artifact.id, source_artifact.id
        )

        self.assert_contains_artifacts_information(
            response,
            "Binary artifacts",
            LookupMultiple((binary_artifact_1.id, binary_artifact_2.id)),
            [binary_artifact_1.id, binary_artifact_2.id],
        )

    def test_get_fail_on_scenarios(self) -> None:
        """_get_fail_on_scenarios processes the task's fail_on correctly."""
        sub_tests = [
            ({"failed_test": True}, ["failed"]),
            (
                {
                    "failed_test": False,
                    "flaky_test": True,
                    "skipped_test": False,
                },
                ["flaky"],
            ),
            (
                {
                    "failed_test": True,
                    "flaky_test": True,
                    "skipped_test": True,
                },
                ["failed", "flaky", "skipped"],
            ),
        ]

        for fail_on, expected_scenarios in sub_tests:
            with self.subTest(fail_on=fail_on):
                self.assertEqual(
                    AutopkgtestTab._get_fail_on_scenarios(fail_on),
                    expected_scenarios,
                )

    def test_show_configured_task_data(self) -> None:
        """The view shows configured task data if present."""
        work_request: WorkRequest = self.create_work_request_for_artifacts(
            self.environment_artifact, self.source_artifact, []
        )

        # Try unconfigured
        context_data = self.get_tab(work_request).get_context_data()
        self.assertEqual(
            context_data["request_data"]["build_architecture"], "amd64"
        )

        # Try configured
        work_request.configured_task_data = work_request.task_data.copy()
        work_request.configured_task_data["build_architecture"] = "arm64"
        work_request.save()
        context_data = self.get_tab(work_request).get_context_data()
        self.assertEqual(
            context_data["request_data"]["build_architecture"], "arm64"
        )


class AutopkgtestLogParserTests(TestCase):
    """Tests for AutopkgtestLogParser class."""

    scenario = scenarios.PackageBuildLog()

    def test_parse_log(self) -> None:
        Section = AutopkgtestLogParser.Section
        SubSection = AutopkgtestLogParser.SubSection
        Section.model_rebuild()
        SubSection.model_rebuild()

        sections, number_of_lines = AutopkgtestLogParser(
            self.scenario.autopkgtest_log
        ).parse()
        self.assertEqual(
            number_of_lines, self.scenario.autopkgtest_log.count(b"\n")
        )

        self.assertEqual(
            sections,
            [
                Section(
                    text="Preparation",
                    color="primary",
                    subsections=[
                        SubSection(
                            text="start run",
                            color=None,
                            subsections=[],
                            line=1,
                            result=None,
                            expanded=True,
                            include_in_table_of_contents=True,
                            tooltip=None,
                        ),
                        SubSection(
                            text="test bed setup",
                            color=None,
                            subsections=[],
                            line=6,
                            result=None,
                            expanded=False,
                            include_in_table_of_contents=True,
                            tooltip=None,
                        ),
                        SubSection(
                            text="apt-source",
                            color=None,
                            subsections=[],
                            line=10,
                            result=None,
                            expanded=False,
                            include_in_table_of_contents=True,
                            tooltip=None,
                        ),
                    ],
                ),
                Section(
                    text="Tests",
                    color="primary",
                    subsections=[
                        SubSection(
                            text="preparing testbed pybuild-autopkgtest",
                            color=None,
                            subsections=[],
                            line=15,
                            result=None,
                            expanded=False,
                            include_in_table_of_contents=False,
                            tooltip=None,
                        ),
                        SubSection(
                            text="pybuild-autopkgtest",
                            color="success",
                            subsections=[],
                            line=23,
                            result=None,
                            expanded=False,
                            include_in_table_of_contents=True,
                            tooltip="Pass",
                        ),
                        SubSection(
                            text="preparing testbed test-suite",
                            color=None,
                            subsections=[],
                            line=31,
                            result=None,
                            expanded=False,
                            include_in_table_of_contents=False,
                            tooltip=None,
                        ),
                        SubSection(
                            text="test-suite",
                            color="danger",
                            subsections=[],
                            line=43,
                            result=None,
                            expanded=True,
                            include_in_table_of_contents=True,
                            tooltip="Fail",
                        ),
                        SubSection(
                            text="preparing testbed test-suite-skip",
                            color=None,
                            subsections=[],
                            line=50,
                            result=None,
                            expanded=False,
                            include_in_table_of_contents=False,
                            tooltip=None,
                        ),
                        SubSection(
                            text="test-suite-skip",
                            color="secondary",
                            subsections=[],
                            line=51,
                            result=None,
                            expanded=False,
                            include_in_table_of_contents=True,
                            tooltip="Skip",
                        ),
                    ],
                ),
                Section(
                    text="Closing",
                    color="primary",
                    subsections=[
                        SubSection(
                            text="summary",
                            color=None,
                            subsections=[],
                            line=54,
                            result=None,
                            expanded=True,
                            include_in_table_of_contents=True,
                            tooltip=None,
                        )
                    ],
                ),
            ],
        )

    def test_parse_log_without_apt_source(self) -> None:
        Section = AutopkgtestLogParser.Section
        SubSection = AutopkgtestLogParser.SubSection
        Section.model_rebuild()
        SubSection.model_rebuild()

        log_contents = textwrap.dedent(
            """\
        line
        line
        15s autopkgtest [22:36:01]: test m-a-autopkgtest: preparing testbed
        line
        43s autopkgtest [22:36:29]: test m-a-autopkgtest: [-----------------------
        44s I: No Linux header packages are installed.
        line
        line
        """  # noqa: E501
        ).encode()

        sections, number_of_lines = AutopkgtestLogParser(log_contents).parse()
        self.assertEqual(number_of_lines, log_contents.count(b"\n"))
        self.assertEqual(len(sections), 3)
        self.assertEqual(sections[0].subsections, [])

        # 1 subsection: preparing and the test itself
        self.assertEqual(
            sections[1].subsections,
            [
                SubSection(
                    text="preparing testbed m-a-autopkgtest",
                    color=None,
                    subsections=[],
                    line=3,
                    result=None,
                    expanded=False,
                    include_in_table_of_contents=False,
                    tooltip=None,
                ),
                SubSection(
                    text="m-a-autopkgtest",
                    color=None,
                    subsections=[],
                    line=5,
                    result=None,
                    expanded=False,
                    include_in_table_of_contents=True,
                    tooltip=None,
                ),
            ],
        )
        self.assertEqual(sections[2].subsections, [])

    def test_parse_log_with_test_setup_error(self) -> None:
        Section = AutopkgtestLogParser.Section
        SubSection = AutopkgtestLogParser.SubSection
        Section.model_rebuild()
        SubSection.model_rebuild()

        log_contents = textwrap.dedent(
            """\
            line
            line
            2s autopkgtest [00:00:01]: test unit-tests-tasks: preparing testbed
            3s E: erroneous package: Test dependencies are unsatisfiable.
            4s autopkgtest [00:00:01]: test unit-tests-sign: preparing testbed
            5s autopkgtest [20:30:43]: test unit-tests-sign: [-----------------------
            5s unit-tests-sign output
            5s autopkgtest [20:30:44]: test unit-tests-sign: -----------------------]
            5s autopkgtest [20:30:44]: test unit-tests-sign:  - - - - - - - - - - results - - - - - - - - - -
            5s unit-tests-sign             PASS
            5s autopkgtest [00:00:01]: @@@@@@@@@@@@@@@@@@@@ summary
            5s unit-tests-tasks  FAIL badpkg
            6s unit-tests-sign  PASS
            line
            line
            """  # noqa: E501
        ).encode()

        sections, _ = AutopkgtestLogParser(log_contents).parse()
        self.assertEqual(len(sections[1].subsections), 3)

        self.assertEqual(
            sections[1].subsections,
            [
                SubSection(
                    text="preparing testbed unit-tests-tasks",
                    color="danger",
                    subsections=[],
                    line=3,
                    result=None,
                    expanded=True,
                    include_in_table_of_contents=True,
                    tooltip="Fail",
                ),
                SubSection(
                    text="preparing testbed unit-tests-sign",
                    color=None,
                    subsections=[],
                    line=5,
                    result=None,
                    expanded=False,
                    include_in_table_of_contents=False,
                    tooltip=None,
                ),
                SubSection(
                    text="unit-tests-sign",
                    color="success",
                    subsections=[],
                    line=6,
                    result=None,
                    expanded=False,
                    include_in_table_of_contents=True,
                    tooltip="Pass",
                ),
            ],
        )

    def test_parse_log_inconsistent_log(self) -> None:
        """
        Log contains, in summary, a result for a test that has not been seen.

        In that case the summary test gets ignored.
        """
        Section = AutopkgtestLogParser.Section
        SubSection = AutopkgtestLogParser.SubSection
        Section.model_rebuild()
        SubSection.model_rebuild()

        log_contents = textwrap.dedent(
            """\
            line
            line
            4s autopkgtest [00:00:01]: test unit-tests-sign: preparing testbed
            5s autopkgtest [20:30:43]: test unit-tests-sign: [-----------------------
            5s unit-tests-sign output
            5s autopkgtest [20:30:44]: test unit-tests-sign: -----------------------]
            5s autopkgtest [20:30:44]: test unit-tests-sign:  - - - - - - - - - - results - - - - - - - - - -
            5s unit-tests-sign             PASS
            5s autopkgtest [00:00:01]: @@@@@@@@@@@@@@@@@@@@ summary
            5s unit-tests-sign  PASS
            6s some-test-not-seen-before  PASS
            """  # noqa: E501
        ).encode()

        sections, _ = AutopkgtestLogParser(log_contents).parse()
        self.assertEqual(len(sections[1].subsections), 2)

        self.assertEqual(
            sections[1].subsections,
            [
                SubSection(
                    text="preparing testbed unit-tests-sign",
                    color=None,
                    subsections=[],
                    line=3,
                    result=None,
                    expanded=False,
                    include_in_table_of_contents=False,
                    tooltip=None,
                ),
                SubSection(
                    text="unit-tests-sign",
                    color="success",
                    subsections=[],
                    line=4,
                    result=None,
                    expanded=False,
                    include_in_table_of_contents=True,
                    tooltip="Pass",
                ),
            ],
        )

    def test_parse_log_with_error_before_any_test(self) -> None:
        Section = AutopkgtestLogParser.Section
        SubSection = AutopkgtestLogParser.SubSection
        Section.model_rebuild()
        SubSection.model_rebuild()

        log_contents = textwrap.dedent(
            """\
            10s autopkgtest [10:30:23]: @@@@@@@@@@@@@@@@@@@@ apt-source
            15s autopkgtest [10:30:25]: ERROR: pkg fail.
            """
        ).encode()

        sections, _ = AutopkgtestLogParser(log_contents).parse()
        self.assertEqual(len(sections[1].subsections), 0)

    def test_parse_log_with_preparation_error(self) -> None:
        Section = AutopkgtestLogParser.Section
        SubSection = AutopkgtestLogParser.SubSection
        Section.model_rebuild()
        SubSection.model_rebuild()

        log_contents = textwrap.dedent(
            """\
            1s autopkgtest [10:00:00]: starting date and time
            2s autopkgtest [10:00:01]: testbed dpkg architecture: amd64
            3s autopkgtest [10:00:02]: ERROR: "dpkg" failed with stderr
            4s Debug: line 1
            5s Debug: line 2
            6s autopkgtest [22:00:06]: test mytest: preparing testbed
            line
            """
        ).encode()

        sections, number_of_lines = AutopkgtestLogParser(log_contents).parse()
        self.assertEqual(
            sections[0].subsections[-1],
            SubSection(
                text="start run",
                color="danger",
                subsections=[],
                line=1,
                result=None,
                expanded=True,
                include_in_table_of_contents=True,
                tooltip=None,
            ),
        )
        self.assertEqual(number_of_lines, log_contents.count(b"\n"))
        self.assertEqual(sections[1].subsections[0].line, 6)

    def test_parse_log_with_early_preparation_error(self) -> None:
        Section = AutopkgtestLogParser.Section
        SubSection = AutopkgtestLogParser.SubSection
        Section.model_rebuild()
        SubSection.model_rebuild()

        log_contents = textwrap.dedent(
            """\
            1s autopkgtest [21:00:00]: ERROR: "dpkg" failed with stderr
            """
        ).encode()

        sections, _ = AutopkgtestLogParser(log_contents).parse()
        self.assertEqual(len(sections[0].subsections), 0)


class AutopkgtestArtifactPluginTests(TestCase):
    """Tests for AutopkgtestArtifactPlugin class."""

    scenario = scenarios.PackageBuildLog()

    def test_get_context_data(self) -> None:
        artifact = self.scenario.autopkgtest_log_artifact
        log_file = artifact.fileinartifact_set.get(path="log")
        log_file.content_type = "text/plain; charset=utf-8"
        log_file.save()

        response = self.client.get(artifact.get_absolute_url())

        tree = self.assertResponseHTML(response)

        assert isinstance(response, SimpleTemplateResponse)
        assert response.context_data is not None
        self.assertEqual(
            response.context_data["specialized_tab"],
            {
                "label": "Autopkgtest log",
                "slug": "autopkgtest-log",
                "template": "web/_autopkgtest-artifact-detail.html",
            },
        )

        (sections, _) = AutopkgtestLogParser(
            self.scenario.autopkgtest_log
        ).parse()

        self.assertEqual(response.context_data["sections"], sections)
        FileSection = LogFileWidget.Section

        self.assertEqual(
            response.context_data["log_widget"]._sections,
            [
                FileSection(
                    title="start run", start_line=1, end_line=5, expanded=True
                ),
                FileSection(
                    title="test bed setup",
                    start_line=6,
                    end_line=9,
                    expanded=False,
                ),
                FileSection(
                    title="apt-source",
                    start_line=10,
                    end_line=14,
                    expanded=False,
                ),
                FileSection(
                    title="preparing testbed pybuild-autopkgtest",
                    start_line=15,
                    end_line=22,
                    expanded=False,
                ),
                FileSection(
                    title="pybuild-autopkgtest",
                    start_line=23,
                    end_line=30,
                    expanded=False,
                ),
                FileSection(
                    title="preparing testbed test-suite",
                    start_line=31,
                    end_line=42,
                    expanded=False,
                ),
                FileSection(
                    title="test-suite",
                    start_line=43,
                    end_line=49,
                    expanded=True,
                ),
                FileSection(
                    title="preparing testbed test-suite-skip",
                    start_line=50,
                    end_line=50,
                    expanded=False,
                ),
                FileSection(
                    title="test-suite-skip",
                    start_line=51,
                    end_line=53,
                    expanded=False,
                ),
                FileSection(
                    title="summary", start_line=54, end_line=58, expanded=True
                ),
            ],
        )

        # Assert line number rendering is correct (has the ID and content)
        line_number_1 = self.assertHasElement(tree, "//a[@id='L1']")
        self.assertTextContentEqual(line_number_1, "1")

    def test_get_context_data_empty_log(self) -> None:
        data = DebianAutopkgtest(
            results={},
            cmdline="autopkgtest",
            source_package=DebianAutopkgtestSource(
                name="hello", version="1.0", url="https://example.com"
            ),
            architecture="amd64",
            distribution="debian:trixie",
        )

        artifact, _ = self.playground.create_artifact(
            paths={"log": b""},
            category=ArtifactCategory.AUTOPKGTEST,
            create_files=True,
            data=data,
        )
        log_file = artifact.fileinartifact_set.get(path="log")
        log_file.content_type = "text/plain; charset=utf-8"
        log_file.save()

        response = self.client.get(artifact.get_absolute_url())

        self.assertResponseHTML(response)

    def test_autopkgtest_without_log(self) -> None:
        data = DebianAutopkgtest(
            results={},
            cmdline="autopkgtest",
            source_package=DebianAutopkgtestSource(
                name="hello", version="1.0", url="https://example.com"
            ),
            architecture="amd64",
            distribution="debian:trixie",
        )

        artifact, _ = self.playground.create_artifact(
            paths={"log": b""},
            category=ArtifactCategory.AUTOPKGTEST,
            create_files=False,
            data=data,
        )

        response = self.client.get(artifact.get_absolute_url())
        tree = self.assertResponseHTML(response)

        error_paragraph = self.assertHasElement(
            tree, "//div[@id='error-artifact']"
        )

        self.assertTextContentEqual(
            error_paragraph,
            "Cannot display autopkgtest: Artifact does not contain log",
        )

    def test_autopkgtest_with_file_in_artifact_error(self) -> None:
        data = DebianAutopkgtest(
            results={},
            cmdline="autopkgtest",
            source_package=DebianAutopkgtestSource(
                name="hello", version="1.0", url="https://example.com"
            ),
            architecture="amd64",
            distribution="debian:trixie",
        )

        artifact, _ = self.playground.create_artifact(
            paths={"log": b""},
            category=ArtifactCategory.AUTOPKGTEST,
            create_files=True,
            skip_add_files_in_store=True,
            data=data,
        )

        response = self.client.get(artifact.get_absolute_url())
        tree = self.assertResponseHTML(response)

        error = self.assertHasElement(
            tree,
            "//div[@id='error-artifact']",
        )

        self.assertTextContentEqual(
            error,
            'Cannot display autopkgtest: '
            'File "log" has not completed uploading.',
        )

    @patch("debusine.web.views.files.MAX_FILE_SIZE", 10)
    def test_autopkgtest_with_file_too_big_to_render(self) -> None:
        data = DebianAutopkgtest(
            results={},
            cmdline="autopkgtest",
            source_package=DebianAutopkgtestSource(
                name="hello", version="1.0", url="https://example.com"
            ),
            architecture="amd64",
            distribution="debian:trixie",
        )

        artifact, _ = self.playground.create_artifact(
            paths={"log": b"x" * (files.MAX_FILE_SIZE + 1)},
            category=ArtifactCategory.AUTOPKGTEST,
            create_files=True,
            data=data,
        )

        response = self.client.get(artifact.get_absolute_url())
        self.assertResponseHTML(response)

        assert response.context_data is not None
        self.assertIsInstance(
            response.context_data["log_widget"], TooBigFileWidget
        )
