# 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 the autopkgtest workflow."""

import datetime as dt
from collections.abc import Iterable, Sequence
from typing import ClassVar, TypeAlias, override
from unittest import mock

from django.test import override_settings
from django.utils import timezone

from debusine.artifacts.models import (
    ArtifactCategory,
    BareDataCategory,
    CollectionCategory,
    DebianAutopkgtest,
    DebianAutopkgtestResult,
    DebianAutopkgtestResultStatus,
    DebianAutopkgtestSource,
    DebusinePromise,
    TaskTypes,
)
from debusine.client.models import LookupChildType
from debusine.db.models import Artifact, Collection, CollectionItem, WorkRequest
from debusine.db.models.work_requests import SkipWorkRequest
from debusine.server.workflows import (
    AutopkgtestWorkflow,
    SbuildWorkflow,
    WorkflowValidationError,
    workflow_utils,
)
from debusine.server.workflows.models import (
    AutopkgtestWorkflowData,
    AutopkgtestWorkflowDynamicData,
    BaseWorkflowData,
    RegressionTrackingUpdateQAResults,
    SbuildWorkflowData,
)
from debusine.server.workflows.tests.helpers import (
    SampleWorkflow,
    WorkflowTestCase,
)
from debusine.server.workflows.tests.test_regression_tracking import (
    RegressionTrackingWorkflowTestMixin,
)
from debusine.tasks.models import (
    ActionSkipIfLookupResultChanged,
    ActionUpdateCollectionWithArtifacts,
    AutopkgtestNeedsInternet,
    BackendType,
    BaseDynamicTaskData,
    ExtraExternalRepository,
    InputArtifactMultiple,
    InputArtifactSingle,
    LookupMultiple,
    OutputData,
    RegressionAnalysisStatus,
    SbuildData,
    SbuildInput,
)
from debusine.test.django import preserve_db_task_registry


class AutopkgtestWorkflowTests(
    RegressionTrackingWorkflowTestMixin, WorkflowTestCase[AutopkgtestWorkflow]
):
    """Unit tests for :py:class:`AutopkgtestWorkflow`."""

    SAMPLE_TASK_DATA = {
        "vendor": "debian",
        "codename": "sid",
    }

    source_artifact: ClassVar[Artifact]
    binary_artifact: ClassVar[Artifact]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.source_artifact = cls.playground.create_source_artifact(
            name="hello", version="1.0-1"
        )
        cls.binary_artifact = (
            cls.playground.create_minimal_binary_package_artifact()
        )

    @override
    def setUp(self) -> None:
        super().setUp()
        self.task_data["source_artifact"] = self.source_artifact.pk
        self.task_data["binary_artifacts"] = [self.binary_artifact.pk]

    def create_binary_upload(
        self, architecture: str, filenames: list[str]
    ) -> Artifact:
        """Create a minimal `debian:upload` artifact with binaries."""
        artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.UPLOAD,
            data={
                "type": "dpkg",
                "changes_fields": {
                    "Architecture": architecture,
                    "Files": [{"name": filename} for filename in filenames],
                },
            },
        )
        return artifact

    def add_uploads(
        self, work_request: WorkRequest, architectures: Iterable[str]
    ) -> None:
        """Add multiple uploads to a workflow's internal collection."""
        internal_collection = work_request.internal_collection
        assert internal_collection is not None
        for architecture in architectures:
            upload = self.create_binary_upload(
                architecture, [f"hello_1.0-1_{architecture}.deb"]
            )
            internal_collection.manager.add_artifact(
                upload,
                user=work_request.created_by,
                workflow=work_request,
                name=f"build-{architecture}",
                variables={"architecture": architecture},
            )

    def add_qa_result(
        self,
        qa_results: Collection,
        package: str,
        version: str,
        architecture: str,
        *,
        timestamp: dt.datetime | None = None,
    ) -> CollectionItem:
        """Add an autopkgtest result to a ``debian:qa-results`` collection."""
        work_request = self.playground.create_worker_task(
            task_name="autopkgtest",
            result=WorkRequest.Results.SUCCESS,
            validate=False,
            use_data=False,
        )
        return qa_results.manager.add_artifact(
            self.playground.create_artifact(
                category=ArtifactCategory.AUTOPKGTEST
            )[0],
            user=self.playground.get_default_user(),
            variables={
                "package": package,
                "version": version,
                "architecture": architecture,
                "timestamp": int((timestamp or timezone.now()).timestamp()),
                "work_request_id": work_request.id,
            },
        )

    def orchestrate(
        self,
        task_data: AutopkgtestWorkflowData,
        architectures: Sequence[str],
        parent: WorkRequest | None = None,
        pipeline_task_name: str = "examplepipeline",
    ) -> WorkRequest:
        """Create and orchestrate an AutopkgtestWorkflow."""

        class ExamplePipeline(
            SampleWorkflow[BaseWorkflowData, BaseDynamicTaskData]
        ):
            """Pipeline workflow that runs sbuild and autopkgtest."""

            TASK_NAME = pipeline_task_name

            @override
            def populate(self_) -> None:
                sbuild = self_.work_request.create_child_workflow(
                    task_name="sbuild",
                    task_data=SbuildWorkflowData(
                        input=SbuildInput(
                            source_artifact=task_data.source_artifact
                        ),
                        target_distribution="debian:sid",
                        architectures=["all", *architectures],
                    ),
                )
                sbuild_workflow = sbuild.get_task()
                assert isinstance(sbuild_workflow, SbuildWorkflow)
                source_artifact = (
                    workflow_utils.locate_debian_source_package_lookup(
                        sbuild_workflow,
                        "input.source_artifact",
                        task_data.source_artifact,
                    )
                )
                self.playground.advance_work_request(sbuild, mark_running=True)
                for architecture in ("all", *architectures):
                    child = sbuild.create_child_worker(
                        task_name="sbuild",
                        task_data=SbuildData(
                            input=SbuildInput(source_artifact=source_artifact),
                            build_architecture=architecture,
                            environment="debian/match:codename=sid",
                        ),
                    )
                    self_.provides_artifact(
                        child,
                        ArtifactCategory.UPLOAD,
                        f"build-{architecture}",
                        data={"architecture": architecture},
                    )
                sbuild.unblock_workflow_children()

                autopkgtest = self_.work_request.create_child_workflow(
                    task_name="autopkgtest",
                    task_data=task_data,
                )
                self.playground.advance_work_request(
                    autopkgtest, mark_pending=True
                )
                self_.orchestrate_child(autopkgtest)

        root = self.playground.create_workflow(
            task_name=pipeline_task_name, parent=parent
        )
        self.assertTrue(
            self.playground.orchestrate_workflow(root, use_data=False)
        )

        return root

    def test_create_orchestrator(self) -> None:
        """An AutopkgtestWorkflow can be instantiated."""
        source_artifact = self.source_artifact.pk
        binary_artifacts = ("internal@collections/name:build-arm64",)
        vendor = "debian"
        codename = "trixie"
        self.task_data.update(
            {
                "source_artifact": source_artifact,
                "binary_artifacts": binary_artifacts,
                "vendor": vendor,
                "codename": codename,
            }
        )
        self.validate = False

        w = self.task
        self.assertEqual(w.data.source_artifact, source_artifact)
        self.assertEqual(
            w.data.binary_artifacts, LookupMultiple(binary_artifacts)
        )
        self.assertEqual(w.data.vendor, vendor)
        self.assertEqual(w.data.codename, codename)
        self.assertEqual(w.data.backend, BackendType.UNSHARE)

    def test_create_orchestrator_explicit_backend(self) -> None:
        """An AutopkgtestWorkflow can be instantiated with a backend."""
        self.task_data["backend"] = BackendType.INCUS_LXC
        self.validate = False
        self.assertEqual(self.task.data.backend, BackendType.INCUS_LXC)

    def test_architectures(self) -> None:
        """Workflow uses architectures from binary_artifacts."""
        architectures = ("amd64", "arm64", "i386")
        self.task_data["binary_artifacts"] = [
            f"internal@collections/name:build-{architecture}"
            for architecture in architectures
        ]
        self.validate = False
        self.add_uploads(self.work_request, architectures)
        self.assertCountEqual(self.task.architectures, architectures)

    def test_architectures_arch_indep_and_arch_dep(self) -> None:
        """Workflow handles arch-indep plus arch-dep binary_artifacts."""
        architectures = ("all", "arm64", "armhf")
        self.task_data["binary_artifacts"] = [
            f"internal@collections/name:build-{architecture}"
            for architecture in architectures
        ]
        self.validate = False
        self.add_uploads(self.work_request, architectures)
        self.assertEqual(self.task.architectures, {"arm64", "armhf"})

    def test_architectures_arch_indep_only(self) -> None:
        """Workflow handles only having arch-indep binary_artifacts."""
        self.task_data["binary_artifacts"] = [
            "internal@collections/name:build-all"
        ]
        self.validate = False
        self.add_uploads(self.work_request, ["all"])
        self.assertEqual(self.task.architectures, {"amd64"})

    def test_architectures_arch_indep_only_arch_all_build_architecture(
        self,
    ) -> None:
        """Workflow honours `arch_all_build_architecture`."""
        self.task_data["binary_artifacts"] = [
            "internal@collections/name:build-all"
        ]
        self.task_data["arch_all_build_architecture"] = "s390x"
        self.validate = False
        self.add_uploads(self.work_request, ["all"])
        self.assertEqual(self.task.architectures, {"s390x"})

    def test_architectures_intersect_task_data(self) -> None:
        """Setting architectures in task data constrains the set."""
        architectures = ("amd64", "arm64", "i386")
        self.task_data["binary_artifacts"] = [
            f"internal@collections/name:build-{architecture}"
            for architecture in architectures
        ]
        self.task_data["architectures"] = ["amd64", "i386"]
        self.validate = False
        self.add_uploads(self.work_request, architectures)
        self.assertEqual(self.task.architectures, {"amd64", "i386"})

    def test_validate_input(self) -> None:
        """validate_input passes a valid case."""
        self.validate = False
        self.add_uploads(self.work_request, ("amd64",))
        self.task.validate_input()

    def test_validate_input_bad_qa_suite(self) -> None:
        """validate_input raises errors in looking up a suite."""
        self.task_data["qa_suite"] = "nonexistent@debian:suite"
        self.validate = False
        with self.assertRaisesRegex(
            WorkflowValidationError,
            "'nonexistent@debian:suite' does not exist or is hidden",
        ):
            self.task.validate_input()

    def test_validate_input_bad_reference_qa_results(self) -> None:
        """validate_input raises errors in looking up reference QA results."""
        self.task_data["reference_qa_results"] = "nonexistent@debian:qa-results"
        self.validate = False
        with self.assertRaisesRegex(
            WorkflowValidationError,
            "'nonexistent@debian:qa-results' does not exist or is hidden",
        ):
            self.task.validate_input()

    def test_validate_input_architecture_errors(self) -> None:
        """validate_input raises errors in computing architectures."""
        self.task_data["binary_artifacts"] = [
            "internal@collections/name:build-amd64"
        ]
        self.validate = False
        assert self.work_request.internal_collection is not None
        # Create a promise with no architecture.
        self.playground.create_bare_data_item(
            self.work_request.internal_collection,
            "build-amd64",
            category=BareDataCategory.PROMISE,
            data=DebusinePromise(
                promise_work_request_id=self.work_request.id + 1,
                promise_workflow_id=self.work_request.id,
                promise_category=ArtifactCategory.UPLOAD,
            ),
        )

        with self.assertRaisesRegex(
            WorkflowValidationError,
            "Cannot determine architecture for lookup result",
        ):
            self.task.validate_input()

    def test_has_current_reference_qa_result_no_match(self) -> None:
        """_has_current_reference_qa_result: no matching result."""
        self.playground.create_collection("sid", CollectionCategory.SUITE)
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_qa_result(sid_qa_results, "other", "1.0-1", "amd64")
        self.task_data = {
            "prefix": "reference-qa-result|",
            "source_artifact": self.source_artifact.id,
            "binary_artifacts": LookupMultiple(()),
            "vendor": "debian",
            "codename": "sid",
            "qa_suite": f"sid@{CollectionCategory.SUITE}",
            "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
            "update_qa_results": RegressionTrackingUpdateQAResults.YES,
        }
        self.assertFalse(self.task._has_current_reference_qa_result("amd64"))

    def test_has_current_reference_qa_result_different_version(self) -> None:
        """_has_current_reference_qa_result: result for different version."""
        self.playground.create_collection("sid", CollectionCategory.SUITE)
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_qa_result(sid_qa_results, "hello", "1.0-1", "amd64")
        source_artifact = self.playground.create_source_artifact(
            name="hello", version="1.0-2"
        )
        self.task_data = {
            "prefix": "reference-qa-result|",
            "source_artifact": source_artifact.id,
            "binary_artifacts": LookupMultiple(()),
            "vendor": "debian",
            "codename": "sid",
            "qa_suite": f"sid@{CollectionCategory.SUITE}",
            "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
            "update_qa_results": RegressionTrackingUpdateQAResults.YES,
        }
        self.assertFalse(self.task._has_current_reference_qa_result("amd64"))

    def test_has_current_reference_qa_result_too_old(self) -> None:
        """_has_current_reference_qa_result: result is too old."""
        now = timezone.now()
        sid_suite = self.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_qa_result(
            sid_qa_results,
            "hello",
            "1.0-1",
            "amd64",
            timestamp=now - dt.timedelta(days=31),
        )
        source_item = sid_suite.manager.add_artifact(
            self.source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )
        source_item.created_at = now
        source_item.save()

        self.task_data = {
            "prefix": "reference-qa-result|",
            "source_artifact": self.source_artifact.id,
            "binary_artifacts": LookupMultiple(()),
            "vendor": "debian",
            "codename": "sid",
            "qa_suite": f"sid@{CollectionCategory.SUITE}",
            "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
            "update_qa_results": RegressionTrackingUpdateQAResults.YES,
        }
        self.assertFalse(self.task._has_current_reference_qa_result("amd64"))

    def test_has_current_reference_qa_result_current(self) -> None:
        """_has_current_reference_qa_result: current result."""
        self.playground.create_collection("sid", CollectionCategory.SUITE)
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_qa_result(sid_qa_results, "hello", "1.0-1", "amd64")

        self.task_data = {
            "prefix": "reference-qa-result|",
            "source_artifact": self.source_artifact.id,
            "binary_artifacts": LookupMultiple(()),
            "vendor": "debian",
            "codename": "sid",
            "qa_suite": f"sid@{CollectionCategory.SUITE}",
            "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
            "update_qa_results": RegressionTrackingUpdateQAResults.YES,
        }

        self.assertTrue(self.task._has_current_reference_qa_result("amd64"))

    @preserve_db_task_registry()
    def test_populate(self) -> None:
        """The workflow populates child work requests."""
        architectures = ("amd64", "i386")

        root = self.orchestrate(
            task_data=AutopkgtestWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{architecture}"
                        for architecture in ("all", *architectures)
                    ]
                ),
                vendor="debian",
                codename="sid",
                extra_repositories=[
                    ExtraExternalRepository(
                        url="http://example.com/",
                        suite="bookworm",
                        components=["main"],
                    )
                ],
            ),
            architectures=architectures,
        )

        autopkgtest = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW, task_name="autopkgtest", parent=root
        )
        children = list(
            WorkRequest.objects.filter(parent=autopkgtest).order_by("id")
        )
        self.assertEqual(len(children), len(architectures))
        for child, architecture in zip(children, architectures):
            self.assertEqual(child.status, WorkRequest.Statuses.BLOCKED)
            self.assertEqual(child.task_type, TaskTypes.WORKER)
            self.assertEqual(child.task_name, "autopkgtest")
            self.assertEqual(
                child.task_data,
                {
                    "input": {
                        "source_artifact": autopkgtest.task_data[
                            "source_artifact"
                        ],
                        "binary_artifacts": sorted(
                            [
                                f"internal@collections/"
                                f"name:build-{architecture}",
                                "internal@collections/name:build-all",
                            ]
                        ),
                        "context_artifacts": [],
                    },
                    "build_architecture": architecture,
                    "environment": "debian/match:codename=sid",
                    "extra_repositories": [
                        {
                            "url": "http://example.com/",
                            "suite": "bookworm",
                            "components": ["main"],
                        }
                    ],
                    "backend": BackendType.UNSHARE,
                    "include_tests": [],
                    "exclude_tests": [],
                    "debug_level": 0,
                    "pin_packages": {},
                    "extra_environment": {},
                    "needs_internet": AutopkgtestNeedsInternet.RUN,
                    "fail_on": {},
                    "timeout": None,
                },
            )
            self.assert_work_request_event_reactions(
                child,
                on_success=[
                    ActionUpdateCollectionWithArtifacts(
                        collection="internal@collections",
                        name_template=f"autopkgtest-{architecture}",
                        artifact_filters={
                            "category": ArtifactCategory.AUTOPKGTEST
                        },
                    )
                ],
            )
            self.assertQuerySetEqual(
                child.dependencies.all(),
                list(
                    WorkRequest.objects.filter(
                        task_type=TaskTypes.WORKER,
                        task_name="sbuild",
                        task_data__build_architecture__in={architecture, "all"},
                    )
                ),
            )
            self.assertEqual(
                child.workflow_data_json,
                {
                    "display_name": f"autopkgtest {architecture}",
                    "step": f"autopkgtest-{architecture}",
                },
            )

        # Population is idempotent.
        self.get_workflow(autopkgtest).populate()
        children = list(WorkRequest.objects.filter(parent=autopkgtest))
        self.assertEqual(len(children), len(architectures))

    @preserve_db_task_registry()
    def test_populate_upload(self) -> None:
        """The workflow accepts debian:upload source artifacts."""
        architectures = ("amd64", "i386")
        upload_artifacts = self.playground.create_upload_artifacts()

        root = self.orchestrate(
            task_data=AutopkgtestWorkflowData(
                source_artifact=upload_artifacts.upload.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{architecture}"
                        for architecture in ("all", *architectures)
                    ]
                ),
                vendor="debian",
                codename="sid",
                extra_repositories=[
                    ExtraExternalRepository(
                        url="http://example.com/",
                        suite="bookworm",
                        components=["main"],
                    )
                ],
            ),
            architectures=architectures,
        )

        autopkgtest = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="autopkgtest",
            parent=root,
        )
        children = list(
            WorkRequest.objects.filter(parent=autopkgtest).order_by("id")
        )
        self.assertEqual(len(children), len(architectures))
        for child, architecture in zip(children, architectures):
            self.assertEqual(child.status, WorkRequest.Statuses.BLOCKED)
            self.assertEqual(child.task_type, TaskTypes.WORKER)
            self.assertEqual(child.task_name, "autopkgtest")
            self.assertEqual(
                child.task_data,
                {
                    "input": {
                        "source_artifact": (
                            f"{upload_artifacts.source.id}@artifacts"
                        ),
                        "binary_artifacts": sorted(
                            [
                                f"internal@collections/"
                                f"name:build-{architecture}",
                                "internal@collections/name:build-all",
                            ]
                        ),
                        "context_artifacts": [],
                    },
                    "build_architecture": architecture,
                    "environment": "debian/match:codename=sid",
                    "extra_repositories": [
                        {
                            "url": "http://example.com/",
                            "suite": "bookworm",
                            "components": ["main"],
                        }
                    ],
                    "backend": BackendType.UNSHARE,
                    "include_tests": [],
                    "exclude_tests": [],
                    "debug_level": 0,
                    "pin_packages": {},
                    "extra_environment": {},
                    "needs_internet": AutopkgtestNeedsInternet.RUN,
                    "fail_on": {},
                    "timeout": None,
                },
            )
            self.assert_work_request_event_reactions(
                child,
                on_success=[
                    ActionUpdateCollectionWithArtifacts(
                        collection="internal@collections",
                        name_template=f"autopkgtest-{architecture}",
                        artifact_filters={
                            "category": ArtifactCategory.AUTOPKGTEST
                        },
                    )
                ],
            )
            self.assertQuerySetEqual(
                child.dependencies.all(),
                list(
                    WorkRequest.objects.filter(
                        task_type=TaskTypes.WORKER,
                        task_name="sbuild",
                        task_data__build_architecture__in={architecture, "all"},
                    )
                ),
            )
            self.assertEqual(
                child.workflow_data_json,
                {
                    "display_name": f"autopkgtest {architecture}",
                    "step": f"autopkgtest-{architecture}",
                },
            )

        # Population is idempotent.
        self.get_workflow(autopkgtest).populate()
        children = list(WorkRequest.objects.filter(parent=autopkgtest))
        self.assertEqual(len(children), len(architectures))

    def test_populate_mixed_uploads(self) -> None:
        """The workflow accepts mixed debian:upload artifacts in all inputs."""
        artifacts = self.playground.create_upload_artifacts(
            binaries=[
                ("hello", "amd64"),
                ("hello", "i386"),
                ("hello-doc", "all"),
            ]
        )
        context = self.playground.create_upload_artifacts(
            binaries=[
                ("context", "amd64"),
                ("context", "i386"),
                ("context-common", "all"),
            ]
        )
        architectures = ["amd64", "i386"]

        self.task_data.update(
            {
                "source_artifact": artifacts.upload.id,
                "binary_artifacts": [artifacts.upload.id],
                "context_artifacts": [context.upload.id],
                "architectures": architectures,
            }
        )
        self.task.populate()
        children = list(self.work_request.children.order_by("id").all())
        self.assertEqual(len(children), len(architectures))
        for architecture, child in zip(architectures, children):
            with self.subTest(architecture=architecture):
                self.assertEqual(
                    child.task_data["build_architecture"], architecture
                )
                self.assertEqual(
                    child.task_data["input"],
                    {
                        "source_artifact": (f"{artifacts.source.id}@artifacts"),
                        "binary_artifacts": sorted(
                            f"{artifact.id}@artifacts"
                            for artifact in artifacts.binaries
                            if artifact.data["deb_fields"]["Architecture"]
                            in (architecture, "all")
                        ),
                        "context_artifacts": sorted(
                            f"{artifact.id}@artifacts"
                            for artifact in context.binaries
                            if artifact.data["deb_fields"]["Architecture"]
                            in (architecture, "all")
                        ),
                    },
                )

    def test_populate_experimental(self) -> None:
        """The workflow handles overlay distributions."""
        self.playground.create_debian_environment(
            codename="experimental", variant="autopkgtest"
        )
        self.task_data["codename"] = "experimental"
        self.task.populate()
        children = list(self.work_request.children.all())
        self.assertEqual(len(children), 1)
        child = children[0]
        self.assertEqual(len(child.task_data["extra_repositories"]), 1)
        repo = child.task_data["extra_repositories"][0]
        self.assertEqual(repo["suite"], "experimental")

    def test_populate_pin_packages(self) -> None:
        self.task_data["pin_packages"] = {"unstable": ["src:glibc", "hello"]}
        self.task.populate()
        children = list(self.work_request.children.all())
        self.assertEqual(len(children), 1)
        child = children[0]
        self.assertEqual(
            child.task_data["pin_packages"],
            {"unstable": ["src:glibc", "hello"]},
        )

    def test_populate_has_current_reference_qa_result(self) -> None:
        """The workflow does nothing with a current reference QA result."""
        self.playground.create_collection("sid", CollectionCategory.SUITE)
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_qa_result(sid_qa_results, "hello", "1.0-1", "amd64")

        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello",
                srcpkg_version="1.0-1",
                architecture="amd64",
            )
        )
        self.task_data.update(
            {
                "prefix": "reference-qa-result|",
                "binary_artifacts": [binary_artifact.id],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
            }
        )

        self.task.populate()
        self.assertQuerySetEqual(self.work_request.children.all(), [])

    def test_populate_has_current_reference_qa_result_force(self) -> None:
        """The workflow can be forced despite a current reference QA result."""
        self.playground.create_collection("sid", CollectionCategory.SUITE)
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_qa_result(sid_qa_results, "hello", "1.0-1", "amd64")

        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello",
                srcpkg_version="1.0-1",
                architecture="amd64",
            )
        )
        self.task_data.update(
            {
                "prefix": "reference-qa-result|",
                "binary_artifacts": [binary_artifact.id],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.FORCE,
            }
        )

        self.task.populate()

        self.assertTrue(
            self.work_request.children.filter(
                task_type=TaskTypes.WORKER, task_name="autopkgtest"
            ).exists()
        )

    def test_populate_previous_reference_qa_result_too_old(self) -> None:
        """The workflow produces a reference QA result if it is outdated."""
        sid_suite = self.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        old_qa_result = self.add_qa_result(
            sid_qa_results,
            "hello",
            "1.0-1",
            "amd64",
            timestamp=timezone.now() - dt.timedelta(days=31),
        )

        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello",
                srcpkg_version="1.0-1",
                architecture="amd64",
            )
        )
        source_item = sid_suite.manager.add_artifact(
            self.source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )
        self.task_data.update(
            {
                "prefix": "reference-qa-result|",
                "binary_artifacts": [binary_artifact.id],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "context_artifacts": [],
            }
        )
        self.task.populate()
        [child] = self.work_request.children.all()
        self.assertEqual(child.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(child.task_type, TaskTypes.WORKER)
        self.assertEqual(child.task_name, "autopkgtest")
        self.assertEqual(
            child.task_data,
            {
                "input": {
                    "source_artifact": self.source_artifact.id,
                    "binary_artifacts": [binary_artifact.id],
                    "context_artifacts": [],
                },
                "build_architecture": "amd64",
                "environment": "debian/match:codename=sid",
                "extra_repositories": None,
                "backend": BackendType.UNSHARE,
                "include_tests": [],
                "exclude_tests": [],
                "debug_level": 0,
                "pin_packages": {},
                "extra_environment": {},
                "needs_internet": AutopkgtestNeedsInternet.RUN,
                "fail_on": {},
                "timeout": None,
            },
        )
        provide_action = ActionUpdateCollectionWithArtifacts(
            collection="internal@collections",
            name_template="reference-qa-result|autopkgtest-amd64",
            artifact_filters={"category": ArtifactCategory.AUTOPKGTEST},
        )
        skip_action = ActionSkipIfLookupResultChanged(
            lookup=(
                f"sid@{CollectionCategory.QA_RESULTS}/"
                f"latest:autopkgtest_hello_amd64"
            ),
            collection_item_id=old_qa_result.id,
            promise_name="reference-qa-result|autopkgtest-amd64",
        )
        qa_result_action = ActionUpdateCollectionWithArtifacts(
            collection=f"sid@{CollectionCategory.QA_RESULTS}",
            variables={
                "package": "hello",
                "version": "1.0-1",
                "architecture": "amd64",
                "timestamp": int(source_item.created_at.timestamp()),
                "work_request_id": child.id,
            },
            artifact_filters={"category": ArtifactCategory.AUTOPKGTEST},
        )
        self.assert_work_request_event_reactions(
            child,
            on_assignment=[skip_action],
            on_failure=[qa_result_action],
            on_success=[provide_action, qa_result_action],
        )

        # Completing the work request stores the QA result.
        result, _ = self.playground.create_artifact(
            category=ArtifactCategory.AUTOPKGTEST, work_request=child
        )
        self.playground.advance_work_request(
            child, result=WorkRequest.Results.SUCCESS, use_data=False
        )
        self.assertEqual(
            child.lookup_single(
                f"sid@{CollectionCategory.QA_RESULTS}/"
                f"latest:autopkgtest_hello_amd64",
                expect_type=LookupChildType.ARTIFACT,
            ).artifact,
            result,
        )

        # Population is idempotent, even if qa_suite changes.
        with (
            mock.patch.object(
                AutopkgtestWorkflow,
                "_has_current_reference_qa_result",
                return_value=False,
            ),
            mock.patch.object(
                AutopkgtestWorkflow,
                "qa_suite_changed",
                mock.PropertyMock(
                    return_value=timezone.now() + dt.timedelta(hours=1)
                ),
            ),
        ):
            self.task.populate()
        self.assertEqual(self.work_request.children.count(), 1)
        child.refresh_from_db()
        self.assert_work_request_event_reactions(
            child,
            on_assignment=[skip_action],
            on_failure=[qa_result_action],
            on_success=[provide_action, qa_result_action],
        )

    def test_populate_reference_qa_result_backs_off(self) -> None:
        """Reference tasks are skipped if another workflow got there first."""
        self.playground.create_collection("sid", CollectionCategory.SUITE)
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.playground.create_debian_environment(
            codename="sid", variant="autopkgtest"
        )

        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello",
                srcpkg_version="1.0-1",
                architecture="amd64",
            )
        )
        self.task_data.update(
            {
                "prefix": "reference-qa-result|",
                "binary_artifacts": LookupMultiple((binary_artifact.id,)),
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
            }
        )
        self.task.populate()

        racing_qa_result = self.add_qa_result(
            sid_qa_results, "hello", "1.0-1", "amd64"
        )
        [child] = self.work_request.children.all()
        self.playground.advance_work_request(child, mark_pending=True)

        with self.assertRaises(SkipWorkRequest):
            child.assign_worker(self.playground.create_worker())

        self.assertEqual(child.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(child.result, WorkRequest.Results.SKIPPED)
        self.assertEqual(
            child.output_data,
            OutputData(
                skip_reason=(
                    f"Result of lookup "
                    f"'sid@{CollectionCategory.QA_RESULTS}/"
                    f"latest:autopkgtest_hello_amd64' changed"
                )
            ),
        )
        self.assertEqual(
            child.lookup_single(
                f"sid@{CollectionCategory.QA_RESULTS}/"
                f"latest:autopkgtest_hello_amd64",
                expect_type=LookupChildType.ARTIFACT,
            ).collection_item,
            racing_qa_result,
        )

    @override_settings(DISABLE_AUTOMATIC_SCHEDULING=True)
    @preserve_db_task_registry()
    def test_callback_regression_analysis(self) -> None:
        self.playground.create_collection("sid", CollectionCategory.SUITE)
        self.playground.create_collection("sid", CollectionCategory.QA_RESULTS)
        architectures = [
            "amd64",
            "arm64",
            "i386",
            "ppc64el",
            "riscv64",
            "s390x",
        ]
        reference_source_artifact = self.source_artifact
        new_source_artifact = self.playground.create_source_artifact(
            name="hello", version="1.0-2"
        )

        root = self.playground.create_workflow(mark_running=True)
        reference = self.orchestrate(
            task_data=AutopkgtestWorkflowData(
                prefix="reference-qa-result|",
                qa_suite=f"sid@{CollectionCategory.SUITE}",
                reference_qa_results=f"sid@{CollectionCategory.QA_RESULTS}",
                update_qa_results=RegressionTrackingUpdateQAResults.YES,
                architectures=architectures,
                source_artifact=reference_source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in architectures
                        if arch != "ppc64el"
                    ]
                ),
                vendor="debian",
                codename="sid",
            ),
            architectures=architectures,
            parent=root,
            pipeline_task_name="referencepipeline",
        )
        new = self.orchestrate(
            task_data=AutopkgtestWorkflowData(
                reference_prefix="reference-qa-result|",
                qa_suite=f"sid@{CollectionCategory.SUITE}",
                reference_qa_results=f"sid@{CollectionCategory.QA_RESULTS}",
                enable_regression_tracking=True,
                architectures=architectures,
                source_artifact=new_source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in architectures
                    ]
                ),
                vendor="debian",
                codename="sid",
            ),
            architectures=architectures,
            parent=root,
        )

        # Unblock the autopkgtest tasks.
        for sbuild in reference.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        ).children.filter(task_type=TaskTypes.WORKER, task_name="sbuild"):
            self.playground.advance_work_request(
                sbuild, result=WorkRequest.Results.SUCCESS, use_data=False
            )
        for sbuild in new.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        ).children.filter(task_type=TaskTypes.WORKER, task_name="sbuild"):
            self.playground.advance_work_request(
                sbuild, result=WorkRequest.Results.SUCCESS, use_data=False
            )

        DARS: TypeAlias = DebianAutopkgtestResultStatus

        reference_autopkgtest_workflow = reference.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="autopkgtest"
        )
        reference_source_url = (
            f"http://example.com/{reference_source_artifact.get_absolute_url()}"
        )
        reference_tasks = {
            wr.task_data["build_architecture"]: wr
            for wr in reference_autopkgtest_workflow.children.filter(
                task_type=TaskTypes.WORKER, task_name="autopkgtest"
            )
        }
        reference_results = {
            architecture: self.playground.create_artifact(
                category=ArtifactCategory.AUTOPKGTEST,
                data=DebianAutopkgtest(
                    results={
                        name: DebianAutopkgtestResult(status=status)
                        for name, status in statuses.items()
                    },
                    cmdline="unused",
                    source_package=DebianAutopkgtestSource(
                        name="hello", version="1.0-1", url=reference_source_url
                    ),
                    architecture=architecture,
                    distribution="sid",
                ),
                work_request=reference_tasks[architecture],
            )[0]
            for architecture, statuses in (
                ("amd64", {"upstream": DARS.PASS, "old": DARS.PASS}),
                ("arm64", {"upstream": DARS.PASS}),
                ("i386", {"upstream": DARS.FAIL}),
                ("riscv64", {"upstream": DARS.FAIL, "performance": DARS.FAIL}),
                ("s390x", {"upstream": DARS.FAIL}),
            )
        }
        new_autopkgtest_workflow = new.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="autopkgtest"
        )
        new_source_url = (
            f"http://example.com/{new_source_artifact.get_absolute_url()}"
        )
        new_tasks = {
            wr.task_data["build_architecture"]: wr
            for wr in new_autopkgtest_workflow.children.filter(
                task_type=TaskTypes.WORKER, task_name="autopkgtest"
            )
        }
        new_results = {
            architecture: self.playground.create_artifact(
                category=ArtifactCategory.AUTOPKGTEST,
                data=DebianAutopkgtest(
                    results={
                        name: DebianAutopkgtestResult(status=status)
                        for name, status in statuses.items()
                    },
                    cmdline="unused",
                    source_package=DebianAutopkgtestSource(
                        name="hello", version="1.0-2", url=new_source_url
                    ),
                    architecture=architecture,
                    distribution="sid",
                ),
                work_request=new_tasks[architecture],
            )[0]
            for architecture, statuses in (
                ("amd64", {"upstream": DARS.PASS}),
                ("arm64", {"upstream": DARS.FAIL}),
                ("i386", {"upstream": DARS.PASS}),
                ("ppc64el", {"new": DARS.PASS}),
                ("riscv64", {"upstream": DARS.FAIL, "performance": DARS.PASS}),
                ("s390x", {"upstream": DARS.PASS, "new": DARS.FAIL}),
            )
        }

        self.assertEqual(
            new_autopkgtest_workflow.output_data,
            OutputData(regression_analysis={}),
        )
        self.schedule_and_run_workflow_callback(new_autopkgtest_workflow)
        self.playground.advance_work_request(
            reference_tasks["amd64"],
            result=WorkRequest.Results.SUCCESS,
            use_data=False,
        )
        self.schedule_and_run_workflow_callback(new_autopkgtest_workflow)
        self.playground.advance_work_request(
            new_tasks["amd64"],
            result=WorkRequest.Results.SUCCESS,
            use_data=False,
        )

        self.schedule_and_run_workflow_callback(new_autopkgtest_workflow)

        expected_analysis: dict[
            str, tuple[RegressionAnalysisStatus, dict[str, str]]
        ] = {
            "amd64": (
                RegressionAnalysisStatus.STABLE,
                {"upstream": "stable", "old": "stable"},
            )
        }
        new_autopkgtest_workflow.refresh_from_db()
        self.assert_regression_analysis(
            new_autopkgtest_workflow,
            "1.0-1",
            reference_results,
            "1.0-2",
            new_results,
            expected_analysis,
        )

        self.assert_callback_status(
            new_autopkgtest_workflow, WorkRequest.Statuses.BLOCKED
        )

        for architecture, reference_result, new_result, status, details in (
            (
                "arm64",
                WorkRequest.Results.SUCCESS,
                WorkRequest.Results.FAILURE,
                RegressionAnalysisStatus.REGRESSION,
                {"upstream": "regression"},
            ),
            (
                "i386",
                WorkRequest.Results.FAILURE,
                WorkRequest.Results.SUCCESS,
                RegressionAnalysisStatus.IMPROVEMENT,
                {"upstream": "improvement"},
            ),
            (
                "ppc64el",
                None,
                WorkRequest.Results.SUCCESS,
                RegressionAnalysisStatus.NO_RESULT,
                {"new": "stable"},
            ),
            (
                "riscv64",
                WorkRequest.Results.FAILURE,
                WorkRequest.Results.FAILURE,
                RegressionAnalysisStatus.IMPROVEMENT,
                {"upstream": "stable", "performance": "improvement"},
            ),
            (
                "s390x",
                WorkRequest.Results.FAILURE,
                WorkRequest.Results.FAILURE,
                RegressionAnalysisStatus.REGRESSION,
                {"upstream": "improvement", "new": "regression"},
            ),
        ):
            if reference_result is not None:
                self.playground.advance_work_request(
                    reference_tasks[architecture],
                    result=reference_result,
                    use_data=False,
                )
            self.playground.advance_work_request(
                new_tasks[architecture], result=new_result, use_data=False
            )
            expected_analysis[architecture] = (status, details)

        self.assert_callback_status(
            new_autopkgtest_workflow, WorkRequest.Statuses.PENDING
        )

        self.schedule_and_run_workflow_callback(new_autopkgtest_workflow)

        new_autopkgtest_workflow.refresh_from_db()
        self.assert_regression_analysis(
            new_autopkgtest_workflow,
            "1.0-1",
            reference_results,
            "1.0-2",
            new_results,
            expected_analysis,
        )
        self.assert_callback_status(
            new_autopkgtest_workflow, WorkRequest.Statuses.COMPLETED
        )

    def test_compute_scheduler_tags(self) -> None:
        self.assertComputesSchedulerTags()

    def test_compute_dynamic_data(self) -> None:
        context_artifact = (
            self.playground.create_minimal_binary_package_artifact()
        )
        self.task_data["context_artifacts"] = [context_artifact.id]
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            AutopkgtestWorkflowDynamicData(
                subject="hello",
                parameter_summary="hello_1.0-1",
                source_artifact_id=self.source_artifact.id,
                binary_artifacts_ids=[self.binary_artifact.id],
                context_artifacts_ids=[context_artifact.id],
            ),
        )

    def test_get_input_artifacts(self) -> None:
        context_artifact = (
            self.playground.create_minimal_binary_package_artifact()
        )
        self.task_data["context_artifacts"] = [context_artifact.id]
        self.task.dynamic_data = self.task.compute_dynamic_data()
        self.assertEqual(
            self.task.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=self.task.data.source_artifact,
                    label="source_artifact",
                    artifact_id=self.source_artifact.id,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=self.task.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=[self.binary_artifact.id],
                ),
                InputArtifactMultiple.from_ids(
                    lookup=self.task.data.context_artifacts,
                    label="context_artifacts",
                    artifact_ids=[context_artifact.id],
                ),
            ],
        )
