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

from collections.abc import Iterable, Mapping, Sequence
from typing import Any, ClassVar, override

from debusine.artifacts.local_artifact import Upload
from debusine.artifacts.models import (
    ArtifactCategory,
    BareDataCategory,
    CollectionCategory,
    DebianBinaryPackage,
    DebianSourcePackage,
    TaskTypes,
)
from debusine.client.models import LookupChildType
from debusine.db.models import Artifact, Collection, WorkRequest
from debusine.server.workflows import (
    ReverseDependenciesAutopkgtestWorkflow,
    WorkflowValidationError,
)
from debusine.server.workflows.base import orchestrate_workflow
from debusine.server.workflows.models import (
    BaseWorkflowData,
    RegressionTrackingUpdateQAResults,
    ReverseDependenciesAutopkgtestWorkflowData,
    ReverseDependenciesAutopkgtestWorkflowDynamicData,
    SbuildWorkflowData,
)
from debusine.server.workflows.reverse_dependencies_autopkgtest import (
    NoBinaryNames,
    _BinaryPackage,
    _SourcePackage,
)
from debusine.server.workflows.tests.helpers import (
    SampleWorkflow,
    WorkflowTestCase,
)
from debusine.tasks import TaskConfigError
from debusine.tasks.models import (
    ActionSkipIfLookupResultChanged,
    ActionUpdateCollectionWithArtifacts,
    AutopkgtestNeedsInternet,
    BackendType,
    BaseDynamicTaskData,
    ExtraExternalRepository,
    InputArtifactMultiple,
    InputArtifactSingle,
    LookupMultiple,
    LookupSingle,
    SbuildData,
    SbuildInput,
)
from debusine.test.django import preserve_db_task_registry


class ReverseDependenciesAutopkgtestWorkflowTests(
    WorkflowTestCase[ReverseDependenciesAutopkgtestWorkflow]
):
    """Unit tests for :py:class:`ReverseDependenciesAutopkgtestWorkflow`."""

    SAMPLE_TASK_DATA = {
        "binary_artifacts": ["internal@collections/name:build-amd64"],
        "qa_suite": "sid@debian:suite",
        "vendor": "debian",
        "codename": "sid",
    }

    sid: ClassVar[Collection]
    source_artifact: ClassVar[Artifact]
    binary_artifact: ClassVar[Artifact]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.sid = cls.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        cls.source_artifact = cls.playground.create_source_artifact(
            name="hello", version="1.0-1"
        )
        cls.binary_artifact = (
            cls.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello", package="hello"
            )
        )

    def setUp(self) -> None:
        super().setUp()
        self.task_data["source_artifact"] = self.source_artifact.pk
        self.internal_collection_artifacts = [
            (self.binary_artifact, {"name": "build-amd64"}),
        ]

    def create_source_package(
        self,
        *,
        name: str,
        version: str,
        dsc_fields: dict[str, Any] | None = None,
        add_to_collection: Collection | None = None,
    ) -> Artifact:
        """Create a `debian:source-package` artifact."""
        artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.SOURCE_PACKAGE,
            data=DebianSourcePackage(
                name=name,
                version=version,
                type="dpkg",
                dsc_fields={
                    "Package": name,
                    "Version": version,
                    **(dsc_fields or {}),
                },
            ),
        )
        if add_to_collection is not None:
            add_to_collection.manager.add_artifact(
                artifact,
                user=self.playground.get_default_user(),
                variables={"component": "main", "section": "devel"},
            )
        return artifact

    def create_binary_package(
        self,
        *,
        source_package: Artifact,
        name: str | None = None,
        version: str | None = None,
        architecture: str = "all",
        deb_fields: dict[str, Any] | None = None,
        add_to_collection: Collection | None = None,
    ) -> Artifact:
        """Create a `debian:binary-package` artifact."""
        source_package_data = source_package.create_data()
        assert isinstance(source_package_data, DebianSourcePackage)
        if name is None:
            name = source_package_data.name
        if version is None:
            version = source_package_data.version
        artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.BINARY_PACKAGE,
            data=DebianBinaryPackage(
                srcpkg_name=source_package_data.name,
                srcpkg_version=source_package_data.version,
                deb_fields={
                    "Package": name,
                    "Version": version,
                    "Architecture": architecture,
                    **(deb_fields or {}),
                },
                deb_control_files=[],
            ),
        )
        if add_to_collection is not None:
            add_to_collection.manager.add_artifact(
                artifact,
                user=self.playground.get_default_user(),
                variables={
                    "component": "main",
                    "section": "devel",
                    "priority": "optional",
                },
            )
        return artifact

    def test_create_orchestrator(self) -> None:
        """A ReverseDependenciesAutopkgtestWorkflow can be instantiated."""
        self.assertEqual(
            self.task.data.source_artifact, self.source_artifact.pk
        )
        self.assertEqual(
            self.task.data.binary_artifacts,
            LookupMultiple(self.SAMPLE_TASK_DATA["binary_artifacts"]),
        )
        self.assertEqual(
            self.task.data.qa_suite, self.SAMPLE_TASK_DATA["qa_suite"]
        )
        self.assertEqual(self.task.data.vendor, "debian")
        self.assertEqual(self.task.data.codename, "sid")
        self.assertEqual(self.task.data.backend, BackendType.UNSHARE)

    def test_create_orchestrator_explicit_backend(self) -> None:
        """A ReverseDependenciesAutopkgtestWorkflow can take a backend."""
        self.task_data["backend"] = BackendType.INCUS_LXC
        self.assertEqual(self.task.data.backend, BackendType.INCUS_LXC)

    def test_validate_input(self) -> None:
        """validate_input passes a valid case."""
        self.task.validate_input()

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

    def test_get_binary_names_promise(self) -> None:
        """Get binary names from a promise."""
        # Delete the automatically-added artifact from the collection,
        # to exercise the promise code
        assert self.work_request.internal_collection is not None
        self.work_request.internal_collection.child_artifacts.clear()

        self.playground.create_bare_data_item(
            self.work_request.internal_collection,
            "build-amd64",
            category=BareDataCategory.PROMISE,
            data={
                "promise_work_request_id": self.work_request.id + 1,
                "promise_workflow_id": self.work_request.id,
                "promise_category": ArtifactCategory.UPLOAD,
                "binary_names": ["hello", "hello-dev"],
            },
        )

        self.assertEqual(self.task.get_binary_names(), {"hello", "hello-dev"})

    def test_get_binary_names_binary_package(self) -> None:
        """Get binary names from a `debian:binary-package` artifact."""
        temp_dir = self.create_temporary_directory()
        binary_package = self.playground.create_artifact_from_local(
            self.playground.create_binary_package(temp_dir, name="single")
        )
        self.task_data["binary_artifacts"] = [binary_package.id]
        self.assertEqual(self.task.get_binary_names(), {"single"})

    def test_get_binary_names_upload(self) -> None:
        """Get binary names from a `debian:upload` artifact."""
        temp_dir = self.create_temporary_directory()
        changes_path = temp_dir / "hello_1.0_amd64.changes"
        self.write_changes_file(
            changes_path, [], binaries=["hello", "hello-dev"]
        )
        upload = self.playground.create_artifact_from_local(
            Upload.create(changes_file=changes_path)
        )
        self.task_data["binary_artifacts"] = [upload.id]
        self.assertEqual(self.task.get_binary_names(), {"hello", "hello-dev"})

    def test_get_binary_names_bad_artifact_category(self) -> None:
        """Cannot get binary names from a non-binary artifact category."""
        temp_dir = self.create_temporary_directory()
        source_package = self.playground.create_artifact_from_local(
            self.playground.create_source_package(temp_dir)
        )
        self.task_data["binary_artifacts"] = [source_package.id]
        with self.assertRaisesRegex(
            NoBinaryNames,
            f"Artifact of category {ArtifactCategory.SOURCE_PACKAGE} has no "
            f"binary packages",
        ):
            self.task.get_binary_names()

    def test_get_reverse_dependencies_bad_source_artifact_category(
        self,
    ) -> None:
        """The source artifact must be a `debian:source-package`."""
        artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.TEST
        )
        self.task_data["source_artifact"] = artifact.id

        with self.assertRaisesRegex(
            TaskConfigError,
            r"^source_artifact: unexpected artifact category: 'debusine:test'. "
            r"Valid categories: \['debian:source-package', 'debian:upload'\]$",
        ):
            self.task

    def assert_reverse_dependencies(
        self,
        workflow: ReverseDependenciesAutopkgtestWorkflow,
        lookups: Mapping[LookupSingle, Iterable[LookupSingle]],
    ) -> None:
        """Assert that reverse-dependencies of a workflow are correct."""
        expected = []
        for source_lookup, binary_lookups in lookups.items():
            source = workflow.work_request.lookup_single(
                source_lookup,
                default_category=CollectionCategory.SUITE,
                expect_type=LookupChildType.ARTIFACT,
            ).artifact
            source_data = source.create_data()
            assert isinstance(source_data, DebianSourcePackage)
            expected_source = _SourcePackage(
                qa_suite=workflow.data.qa_suite,
                name=source_data.name,
                version=source_data.version,
            )
            expected_binaries = []
            for binary_lookup in binary_lookups:
                binary = workflow.work_request.lookup_single(
                    binary_lookup,
                    default_category=CollectionCategory.SUITE,
                    expect_type=LookupChildType.ARTIFACT,
                ).artifact
                binary_data = binary.create_data()
                assert isinstance(binary_data, DebianBinaryPackage)
                expected_binaries.append(
                    _BinaryPackage(
                        qa_suite=workflow.data.qa_suite,
                        name=binary_data.deb_fields["Package"],
                        version=binary_data.deb_fields["Version"],
                        architecture=binary_data.deb_fields["Architecture"],
                    )
                )
            expected.append(
                (
                    expected_source,
                    sorted(
                        expected_binaries,
                        key=lambda binary: (
                            binary.name,
                            binary.version,
                            binary.architecture,
                        ),
                    ),
                )
            )

        rdeps = workflow.get_reverse_dependencies()

        self.assertEqual(
            rdeps,
            sorted(expected, key=lambda rdep: (rdep[0].name, rdep[0].version)),
        )
        # The lookups are all cached.
        for source_package, binary_packages in rdeps:
            with self.assertNumQueries(0):
                workflow.work_request.lookup_single(
                    source_package.to_lookup(),
                    expect_type=LookupChildType.ARTIFACT,
                )
            for binary_package in binary_packages:
                with self.assertNumQueries(0):
                    workflow.work_request.lookup_single(
                        binary_package.to_lookup(),
                        expect_type=LookupChildType.ARTIFACT_OR_PROMISE,
                    )

    def test_get_reverse_dependencies(self) -> None:
        """Get reverse dependencies of a source package and its binaries."""
        trixie = self.playground.create_collection(
            name="trixie", category=CollectionCategory.SUITE
        )

        pre_depends = self.create_source_package(
            name="pre-depends",
            version="1.0",
            dsc_fields={"Testsuite": "autopkgtest"},
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=pre_depends,
            name="python3-pre-depends",
            version="1:1.0",
            deb_fields={"Pre-Depends": "python3-urllib3:any"},
            add_to_collection=self.sid,
        )

        depends = self.create_source_package(
            name="depends",
            version="2.32.3+dfsg-1",
            dsc_fields={
                "Testsuite": "autopkgtest-pkg-python, autopkgtest-pkg-pybuild"
            },
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=depends,
            name="python3-depends",
            deb_fields={"Depends": "python3-urllib3 (>= 1.21.1), python3:any"},
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=depends,
            name="depends-doc",
            deb_fields={"Depends": "libjs-sphinxdoc"},
            add_to_collection=self.sid,
        )

        similar_depends = self.create_source_package(
            name="similar-depends",
            version="1",
            dsc_fields={"Testsuite": "autopkgtest-pkg-pybuild"},
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=similar_depends,
            name="python3-similar-depends",
            deb_fields={"Depends": "python3-urllib3-plugin"},
            add_to_collection=self.sid,
        )

        for version, add_to_collection in (("1", trixie), ("2", self.sid)):
            trigger = self.create_source_package(
                name="trigger",
                version=version,
                dsc_fields={
                    "Testsuite": "autopkgtest",
                    "Testsuite-Triggers": "python3-urllib3",
                },
                add_to_collection=add_to_collection,
            )
            self.create_binary_package(
                source_package=trigger, add_to_collection=add_to_collection
            )

        previous_version = self.create_source_package(
            name="python-urllib3",
            version="2.0.7-1",
            dsc_fields={
                "Testsuite": "autopkgtest",
                "Testsuite-Triggers": "python3-urllib3",
            },
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=previous_version, name="python3-urllib3"
        )

        unrelated = self.create_source_package(
            name="unrelated",
            version="1",
            dsc_fields={"Testsuite": "autopkgtest"},
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=unrelated, add_to_collection=add_to_collection
        )

        untested = self.create_source_package(
            name="untested", version="1", add_to_collection=self.sid
        )
        self.create_binary_package(
            source_package=untested,
            deb_fields={"Depends": "python3-urllib3"},
            add_to_collection=self.sid,
        )

        unsuitable_testsuite = self.create_source_package(
            name="unsuitable-testsuite",
            version="1",
            dsc_fields={
                "Testsuite": "autopkgtest-unsuitable",
                "Testsuite-Triggers": "python3-urllib3",
            },
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=unsuitable_testsuite, add_to_collection=self.sid
        )

        source_package = self.create_source_package(
            name="python-urllib3", version="2.0.7-2"
        )
        binary_package = self.create_binary_package(
            source_package=source_package, name="python3-urllib3"
        )
        workflow = self.task_create(
            source_artifact=source_package.id,
            binary_artifacts=[binary_package.id],
            qa_suite=self.sid.id,
        )

        self.assert_reverse_dependencies(
            workflow,
            {
                "sid/source-version:pre-depends_1.0": (
                    "sid/binary-version:python3-pre-depends_1:1.0_all",
                ),
                "sid/source-version:depends_2.32.3+dfsg-1": (
                    "sid/binary-version:python3-depends_2.32.3+dfsg-1_all",
                    "sid/binary-version:depends-doc_2.32.3+dfsg-1_all",
                ),
                "sid/source-version:trigger_2": (
                    "sid/binary-version:trigger_2_all",
                ),
            },
        )

        workflow.data.packages_denylist = ["depends"]
        self.assert_reverse_dependencies(
            workflow,
            {
                "sid/source-version:pre-depends_1.0": (
                    "sid/binary-version:python3-pre-depends_1:1.0_all",
                ),
                "sid/source-version:trigger_2": (
                    "sid/binary-version:trigger_2_all",
                ),
            },
        )

        workflow.data.packages_denylist = []
        workflow.data.packages_allowlist = ["depends"]
        self.assert_reverse_dependencies(
            workflow,
            {
                "sid/source-version:depends_2.32.3+dfsg-1": (
                    "sid/binary-version:python3-depends_2.32.3+dfsg-1_all",
                    "sid/binary-version:depends-doc_2.32.3+dfsg-1_all",
                ),
            },
        )

        upload_artifacts = self.playground.create_upload_artifacts(
            src_name="python-urllib3",
            version="2.0.7-2",
            binaries=[("python3-urllib3", "all")],
        )
        workflow = self.task_create(
            source_artifact=upload_artifacts.upload.id,
            binary_artifacts=[upload_artifacts.binaries[0].id],
            qa_suite=self.sid.id,
        )
        self.assert_reverse_dependencies(
            workflow,
            {
                "sid/source-version:pre-depends_1.0": (
                    "sid/binary-version:python3-pre-depends_1:1.0_all",
                ),
                "sid/source-version:depends_2.32.3+dfsg-1": (
                    "sid/binary-version:python3-depends_2.32.3+dfsg-1_all",
                    "sid/binary-version:depends-doc_2.32.3+dfsg-1_all",
                ),
                "sid/source-version:trigger_2": (
                    "sid/binary-version:trigger_2_all",
                ),
            },
        )

    def orchestrate(
        self,
        source_artifact: Artifact,
        architectures: Sequence[str],
        *,
        extra_data: dict[str, Any] | None = None,
        priority_base: int = 0,
        parent: WorkRequest | None = None,
        pipeline_task_name: str = "examplepipeline",
    ) -> WorkRequest:
        """Create a workflow and call ``orchestrate_workflow``."""

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

            TASK_NAME = pipeline_task_name

            def populate(self_) -> None:
                """Populate the pipeline."""
                sbuild = self_.work_request.create_child_workflow(
                    task_name="sbuild",
                    task_data=SbuildWorkflowData(
                        input=SbuildInput(source_artifact=source_artifact.id),
                        target_distribution="debian:sid",
                        architectures=["all", *architectures],
                    ),
                )
                for architecture in ("all", *architectures):
                    child = sbuild.create_child_worker(
                        task_name="sbuild",
                        task_data=SbuildData(
                            input=SbuildInput(
                                source_artifact=source_artifact.id
                            ),
                            build_architecture=architecture,
                            environment="debian/match:codename=sid",
                        ),
                    )
                    self_.provides_artifact(
                        child,
                        ArtifactCategory.UPLOAD,
                        f"build-{architecture}",
                        data={
                            "binary_names": ["hello"],
                            "architecture": architecture,
                        },
                    )

                rdep_autopkgtest = self_.work_request.create_child_workflow(
                    task_name="reverse_dependencies_autopkgtest",
                    task_data=ReverseDependenciesAutopkgtestWorkflowData(
                        source_artifact=source_artifact.id,
                        binary_artifacts=LookupMultiple.model_validate(
                            [
                                f"internal@collections/"
                                f"name:build-{architecture}"
                                for architecture in ("all", *architectures)
                            ]
                        ),
                        qa_suite=f"sid@{CollectionCategory.SUITE}",
                        vendor="debian",
                        codename="sid",
                        **(extra_data or {}),
                    ),
                )
                self.playground.advance_work_request(
                    rdep_autopkgtest, mark_pending=True
                )
                self_.orchestrate_child(rdep_autopkgtest)

        root = self.playground.create_workflow(
            task_name=pipeline_task_name, parent=parent
        )
        root.priority_base = priority_base
        root.save()
        self.assertTrue(orchestrate_workflow(root))

        return root.children.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="reverse_dependencies_autopkgtest",
        )

    @preserve_db_task_registry()
    def test_populate(self) -> None:
        """The workflow populates child work requests."""
        depends = self.create_source_package(
            name="depends",
            version="1.0",
            dsc_fields={"Testsuite": "autopkgtest"},
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=depends,
            deb_fields={"Depends": "hello"},
            add_to_collection=self.sid,
        )
        trigger = self.create_source_package(
            name="trigger",
            version="1.0",
            dsc_fields={
                "Testsuite": "autopkgtest",
                "Testsuite-Triggers": "hello",
            },
            add_to_collection=self.sid,
        )
        self.create_binary_package(
            source_package=trigger, add_to_collection=self.sid
        )
        source_artifact = self.source_artifact
        architectures = ("amd64", "i386")

        rdep_autopkgtest = self.orchestrate(
            source_artifact,
            architectures,
            extra_data={
                "extra_repositories": [
                    ExtraExternalRepository(
                        url="http://example.com/",
                        suite="bookworm",
                        components=["main"],
                    )
                ]
            },
        )

        children = list(
            WorkRequest.objects.filter(parent=rdep_autopkgtest).order_by(
                "task_data__prefix"
            )
        )
        for child, source in zip(children, ("depends_1.0", "trigger_1.0")):
            self.assertEqual(child.status, WorkRequest.Statuses.RUNNING)
            self.assertEqual(child.task_type, TaskTypes.WORKFLOW)
            self.assertEqual(child.task_name, "autopkgtest")
            self.assertEqual(
                child.task_data,
                {
                    "prefix": f"{source}|",
                    "reference_prefix": f"{source}|",
                    "source_artifact": (
                        f"sid@{CollectionCategory.SUITE}/"
                        f"source-version:{source}"
                    ),
                    "binary_artifacts": [
                        f"sid@{CollectionCategory.SUITE}/"
                        f"binary-version:{source}_all"
                    ],
                    "context_artifacts": [
                        f"internal@collections/name:build-{architecture}"
                        for architecture in ("all", *architectures)
                    ],
                    "qa_suite": f"sid@{CollectionCategory.SUITE}",
                    "reference_qa_results": None,
                    "enable_regression_tracking": False,
                    "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                    "vendor": "debian",
                    "codename": "sid",
                    "backend": BackendType.UNSHARE,
                    "architectures": [],
                    "arch_all_build_architecture": "amd64",
                    "debug_level": 0,
                    "extra_repositories": [
                        {
                            "url": "http://example.com/",
                            "suite": "bookworm",
                            "components": ["main"],
                        }
                    ],
                },
            )
            self.assertEqual(child.priority_base, -5)

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

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

    @preserve_db_task_registry()
    def test_populate_passes_regression_tracking_parameters(self) -> None:
        """The workflow passes on parameters related to regression tracking."""
        self.playground.create_collection(
            name="sid", category=CollectionCategory.QA_RESULTS
        )
        depends = self.create_source_package(
            name="depends",
            version="1.0",
            dsc_fields={"Testsuite": "autopkgtest"},
            add_to_collection=self.sid,
        )
        binary_artifact = self.create_binary_package(
            source_package=depends,
            deb_fields={"Depends": "hello"},
            add_to_collection=self.sid,
        )
        binary_item = self.sid.child_items.get(artifact=binary_artifact)
        source_artifact = self.source_artifact
        architectures = ("amd64", "i386")

        root = self.playground.create_workflow(mark_running=True)
        reference = self.orchestrate(
            source_artifact,
            architectures,
            extra_data={
                "prefix": "reference-qa-result|",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
            },
            parent=root,
            pipeline_task_name="referencepipeline",
        )
        assert reference.parent is not None
        new = self.orchestrate(
            source_artifact,
            architectures,
            extra_data={
                "reference_prefix": "reference-qa-result|",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "enable_regression_tracking": True,
            },
            parent=root,
        )

        reference_child = reference.children.get()
        self.assertEqual(reference_child.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(reference_child.task_type, TaskTypes.WORKFLOW)
        self.assertEqual(reference_child.task_name, "autopkgtest")
        self.assertEqual(
            reference_child.task_data,
            {
                "prefix": "reference-qa-result|depends_1.0|",
                "reference_prefix": "depends_1.0|",
                "source_artifact": (
                    f"sid@{CollectionCategory.SUITE}/source-version:depends_1.0"
                ),
                "binary_artifacts": [
                    f"sid@{CollectionCategory.SUITE}/"
                    f"binary-version:depends_1.0_all"
                ],
                "context_artifacts": [
                    f"internal@collections/name:build-{architecture}"
                    for architecture in ("all", *architectures)
                ],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "enable_regression_tracking": False,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "vendor": "debian",
                "codename": "sid",
                "backend": BackendType.UNSHARE,
                "architectures": [],
                "arch_all_build_architecture": "amd64",
                "debug_level": 0,
                "extra_repositories": None,
            },
        )

        reference_grandchild = reference_child.children.get(
            task_type=TaskTypes.WORKER, task_name="autopkgtest"
        )
        self.assertEqual(
            reference_grandchild.status, WorkRequest.Statuses.BLOCKED
        )
        self.assertEqual(reference_grandchild.task_type, TaskTypes.WORKER)
        self.assertEqual(reference_grandchild.task_name, "autopkgtest")
        self.assertEqual(
            reference_grandchild.task_data,
            {
                "input": {
                    "source_artifact": (
                        f"sid@{CollectionCategory.SUITE}/"
                        f"source-version:depends_1.0"
                    ),
                    "binary_artifacts": [
                        f"sid@{CollectionCategory.SUITE}/"
                        f"binary-version:depends_1.0_all"
                    ],
                    "context_artifacts": [
                        f"internal@collections/name:build-{architecture}"
                        for architecture in ("all", "amd64")
                    ],
                },
                "build_architecture": "amd64",
                "environment": "debian/match:codename=sid",
                "backend": BackendType.UNSHARE,
                "include_tests": [],
                "exclude_tests": [],
                "debug_level": 0,
                "extra_environment": {},
                "extra_repositories": None,
                "pin_packages": {},
                "needs_internet": AutopkgtestNeedsInternet.RUN,
                "fail_on": {},
                "timeout": None,
            },
        )
        qa_result_action = ActionUpdateCollectionWithArtifacts(
            collection=f"sid@{CollectionCategory.QA_RESULTS}",
            variables={
                "package": "depends",
                "version": "1.0",
                "architecture": "amd64",
                "timestamp": int(binary_item.created_at.timestamp()),
                "work_request_id": reference_grandchild.id,
            },
            artifact_filters={"category": ArtifactCategory.AUTOPKGTEST},
        )
        self.assert_work_request_event_reactions(
            reference_grandchild,
            on_assignment=[
                ActionSkipIfLookupResultChanged(
                    lookup=(
                        f"sid@{CollectionCategory.QA_RESULTS}/"
                        f"latest:autopkgtest_depends_amd64"
                    ),
                    promise_name=(
                        "reference-qa-result|depends_1.0|autopkgtest-amd64"
                    ),
                    collection_item_id=None,
                )
            ],
            on_failure=[qa_result_action],
            on_success=[
                ActionUpdateCollectionWithArtifacts(
                    collection="internal@collections",
                    name_template=(
                        "reference-qa-result|depends_1.0|autopkgtest-amd64"
                    ),
                    artifact_filters={"category": ArtifactCategory.AUTOPKGTEST},
                ),
                qa_result_action,
            ],
        )
        self.assertQuerySetEqual(
            reference_grandchild.dependencies.all(),
            # https://github.com/typeddjango/django-stubs/pull/3087
            reference.parent.children.get(  # type: ignore[arg-type]
                task_type=TaskTypes.WORKFLOW, task_name="sbuild"
            ).children.filter(
                task_data__build_architecture__in={"all", "amd64"}
            ),
        )
        self.assertEqual(
            reference_grandchild.workflow_data_json,
            {
                "allow_failure": True,
                "display_name": "autopkgtest amd64",
                "step": "autopkgtest-amd64",
            },
        )

        new_child = new.children.get()
        self.assertEqual(new_child.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(new_child.task_type, TaskTypes.WORKFLOW)
        self.assertEqual(new_child.task_name, "autopkgtest")
        self.assertEqual(
            new_child.task_data,
            {
                "prefix": "depends_1.0|",
                "reference_prefix": "reference-qa-result|depends_1.0|",
                "source_artifact": (
                    f"sid@{CollectionCategory.SUITE}/source-version:depends_1.0"
                ),
                "binary_artifacts": [
                    f"sid@{CollectionCategory.SUITE}/"
                    f"binary-version:depends_1.0_all"
                ],
                "context_artifacts": [
                    f"internal@collections/name:build-{architecture}"
                    for architecture in ("all", *architectures)
                ],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "enable_regression_tracking": True,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
                "codename": "sid",
                "backend": BackendType.UNSHARE,
                "architectures": [],
                "arch_all_build_architecture": "amd64",
                "debug_level": 0,
                "extra_repositories": None,
            },
        )

        new_grandchild = new_child.children.get(
            task_type=TaskTypes.WORKER, task_name="autopkgtest"
        )
        self.assertEqual(new_grandchild.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(new_grandchild.task_type, TaskTypes.WORKER)
        self.assertEqual(new_grandchild.task_name, "autopkgtest")
        self.assertEqual(
            new_grandchild.task_data,
            {
                "input": {
                    "source_artifact": (
                        f"sid@{CollectionCategory.SUITE}/"
                        f"source-version:depends_1.0"
                    ),
                    "binary_artifacts": [
                        f"sid@{CollectionCategory.SUITE}/"
                        f"binary-version:depends_1.0_all"
                    ],
                    "context_artifacts": [
                        f"internal@collections/name:build-{architecture}"
                        for architecture in ("all", "amd64")
                    ],
                },
                "build_architecture": "amd64",
                "environment": "debian/match:codename=sid",
                "backend": BackendType.UNSHARE,
                "include_tests": [],
                "exclude_tests": [],
                "debug_level": 0,
                "extra_environment": {},
                "extra_repositories": None,
                "pin_packages": {},
                "needs_internet": AutopkgtestNeedsInternet.RUN,
                "fail_on": {},
                "timeout": None,
            },
        )
        self.assert_work_request_event_reactions(
            new_grandchild,
            on_success=[
                ActionUpdateCollectionWithArtifacts(
                    collection="internal@collections",
                    name_template="depends_1.0|autopkgtest-amd64",
                    artifact_filters={"category": ArtifactCategory.AUTOPKGTEST},
                )
            ],
        )
        self.assertQuerySetEqual(
            new_grandchild.dependencies.all(),
            # https://github.com/typeddjango/django-stubs/pull/3087
            reference.parent.children.get(  # type: ignore[arg-type]
                task_type=TaskTypes.WORKFLOW, task_name="sbuild"
            ).children.filter(
                task_data__build_architecture__in={"all", "amd64"}
            ),
        )
        self.assertEqual(
            new_grandchild.workflow_data_json,
            {"display_name": "autopkgtest amd64", "step": "autopkgtest-amd64"},
        )

        # Population is idempotent.
        self.get_workflow(reference).populate()
        reference_children = list(WorkRequest.objects.filter(parent=reference))
        self.assertEqual(len(reference_children), 1)
        self.get_workflow(new).populate()
        new_children = list(WorkRequest.objects.filter(parent=new))
        self.assertEqual(len(new_children), 1)

    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),
            ReverseDependenciesAutopkgtestWorkflowDynamicData(
                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],
                qa_suite_id=self.sid.id,
            ),
        )

    def test_get_input_artifacts(self) -> None:
        binary_artifact_1 = (
            self.playground.create_minimal_binary_package_artifact(
                package="hello"
            )
        )
        binary_artifact_2 = (
            self.playground.create_minimal_binary_package_artifact(
                package="bye"
            )
        )
        context_artifact_1 = (
            self.playground.create_minimal_binary_package_artifact(
                package="python3"
            )
        )
        context_artifact_2 = (
            self.playground.create_minimal_binary_package_artifact(
                package="python3-django"
            )
        )
        self.task_data["binary_artifacts"] = [
            binary_artifact_1.id,
            binary_artifact_2.id,
        ]
        self.task_data["context_artifacts"] = [
            context_artifact_1.id,
            context_artifact_2.id,
        ]
        workflow = self.task
        workflow.dynamic_data = workflow.compute_dynamic_data()
        self.assertEqual(
            workflow.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=workflow.data.source_artifact,
                    label="source_artifact",
                    artifact_id=self.source_artifact.id,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=workflow.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=[binary_artifact_1.id, binary_artifact_2.id],
                ),
                InputArtifactMultiple.from_ids(
                    lookup=workflow.data.context_artifacts,
                    label="context_artifacts",
                    artifact_ids=[context_artifact_1.id, context_artifact_2.id],
                ),
            ],
        )
