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

from collections.abc import Iterable
from typing import Any, ClassVar
from unittest.mock import call, patch

from debusine.artifacts.models import (
    ArtifactCategory,
    BareDataCategory,
    CollectionCategory,
    TaskTypes,
)
from debusine.client.models import LookupChildType
from debusine.db.models import Artifact, Collection, WorkRequest, Workspace
from debusine.db.permissioncontext import PermissionContext
from debusine.server.collections.lookup import lookup_single
from debusine.server.workflows import DebDiffWorkflow, workflow_utils
from debusine.server.workflows.debdiff import Architecture
from debusine.server.workflows.models import (
    BaseWorkflowData,
    DebDiffWorkflowData,
    DebDiffWorkflowDynamicData,
    WorkRequestWorkflowData,
)
from debusine.server.workflows.tests.helpers import (
    SampleWorkflow,
    WorkflowTestCase,
)
from debusine.tasks.models import (
    BaseDynamicTaskData,
    DebDiffFlags,
    InputArtifactMultiple,
    InputArtifactSingle,
    LookupMultiple,
    LookupSingle,
    OutputData,
    OutputDataError,
    SbuildData,
    SbuildInput,
)
from debusine.test.django import preserve_db_task_registry


class DebDiffWorkflowTests(WorkflowTestCase[DebDiffWorkflow]):
    """Unit tests for :py:class:`DebDiffWorkflow`."""

    origin_source_artifact: ClassVar[Artifact]
    origin_binary_artifacts: ClassVar[dict[Architecture, Artifact]]

    new_source_artifact: ClassVar[Artifact]
    new_binary_artifacts: ClassVar[dict[Architecture, Artifact]]

    debian_collection: ClassVar[Collection]

    workspace: ClassVar[Workspace]

    @classmethod
    def setUpTestData(cls) -> None:
        """Set up common data."""
        super().setUpTestData()

        cls.workspace = cls.playground.get_default_workspace()

        cls.debian_collection = cls.playground.create_collection(
            name="bookworm",
            category=CollectionCategory.SUITE,
        )

        cls.origin_source_artifact = cls.playground.create_source_artifact(
            name="hello", version="1.0.0"
        )
        cls.new_source_artifact = cls.playground.create_source_artifact(
            name="hello", version="2.0.0", architectures={"all", "amd64"}
        )

        cls.origin_binary_artifacts = {}
        cls.new_binary_artifacts = {}

        for arch in ["amd64", "arm64", "all"]:
            cls.origin_binary_artifacts[Architecture(arch)] = (
                cls.playground.create_minimal_binary_package_artifact(
                    package="hello",
                    srcpkg_name="hello",
                    srcpkg_version="1.0.0",
                    version="1.0.0",
                    architecture=arch,
                )
            )
            cls.new_binary_artifacts[Architecture(arch)] = (
                cls.playground.create_minimal_binary_package_artifact(
                    package="hello",
                    srcpkg_name="hello",
                    srcpkg_version="2.0.0",
                    version="2.0.0",
                    architecture=arch,
                )
            )

    def orchestrate(
        self,
        *,
        task_data: DebDiffWorkflowData,
        sbuild_architectures: list[str],
        source_artifact: Artifact,
        target_distribution: str,
        expect_success: bool = True,
        use_data: bool = False,
    ) -> WorkRequest:
        """Create and orchestrate a DebDiffWorkflow."""

        class ExamplePipeline(
            SampleWorkflow[BaseWorkflowData, BaseDynamicTaskData]
        ):
            """Pipeline workflow."""

            def populate(self) -> None:
                """Populate the pipeline."""
                for arch in sbuild_architectures:
                    sbuild, _ = self.work_request_ensure_child_worker(
                        task_name="sbuild",
                        task_data=SbuildData(
                            input=SbuildInput(
                                source_artifact=source_artifact.id
                            ),
                            environment=target_distribution,
                            build_architecture="amd64",
                        ),
                        workflow_data=WorkRequestWorkflowData(
                            display_name=f"Build {arch}", step=f"build-{arch}"
                        ),
                    )
                    self.provides_artifact(
                        sbuild,
                        ArtifactCategory.BINARY_PACKAGE,
                        name=f"build-{arch}",
                        data={
                            "binary_package_name": source_artifact.data["name"],
                            "architecture": arch,
                        },
                    )

                debdiff, _ = self.work_request_ensure_child_workflow(
                    task_name="debdiff",
                    task_data=task_data,
                    workflow_data=WorkRequestWorkflowData(
                        display_name="debdiff",
                        step="debdiff",
                    ),
                )
                self.orchestrate_child(debdiff)

        root = self.playground.create_workflow(task_name="examplepipeline")
        self.assertEqual(
            self.playground.orchestrate_workflow(root, use_data=use_data),
            expect_success,
        )

        return root

    def setup_debdiff_workflow(
        self,
        *,
        add_origin_source_artifact: bool = True,
        add_origin_binary_artifacts_archs: Iterable[str] | None = None,
        override_source_artifact: LookupSingle | None = None,
        add_new_binary_artifacts_archs: Iterable[str] | None = None,
        override_binary_artifacts: list[LookupSingle] | None = None,
        arch_all_build_architecture: str | None = None,
        original: str | None = None,
        extra_flags: list[DebDiffFlags] | None = None,
        expect_success: bool = True,
        use_data: bool = False,
    ) -> None:
        """
        Create and schedule a DebDiff workflow.

        :param add_origin_source_artifact: whether to add
          self.add_origin_source_artifact in the "original" collection
        :param add_origin_binary_artifacts_archs: add each architecture from
          self.origin_binary_artifacts into the "original" collection.
          If None: adds all of them
        :param override_source_artifact: use this source_artifact. If None:
          use default's self.new_source_artifact
        :param add_new_binary_artifacts_archs: add each architecture from
          self.new_binary_artifacts in the DebDiff's "task_data"
          If None: adds all of them
        :param override_binary_artifacts: if not None: for binary_artifacts
          use this LookupSingle (ignoring add_new_binary_artifacts_archs)
        :param arch_all_build_architecture: if None: does nothing. Else add
           "arch_all_build_architecture" in the "task_data"
        :param original: collection to use as "original". If it's None, use
          self.debian_collection
        :param extra_flags: if not None add "extra_flags" in the task data
        :param expect_success: whether the orchestrator is expected to
          succeed
        """
        if add_origin_source_artifact:
            self.debian_collection.manager.add_artifact(
                self.origin_source_artifact,
                user=self.playground.get_default_user(),
                variables={"component": "main", "section": "devel"},
            )

        if add_origin_binary_artifacts_archs is None:
            add_origin_binary_artifacts_archs = (
                self.origin_binary_artifacts.keys()
            )

        if add_new_binary_artifacts_archs is None:
            add_new_binary_artifacts_archs = self.new_binary_artifacts.keys()

        for arch in add_origin_binary_artifacts_archs:
            self.debian_collection.manager.add_artifact(
                self.origin_binary_artifacts[Architecture(arch)],
                user=self.playground.get_default_user(),
                variables={
                    "component": "main",
                    "section": "devel",
                    "priority": "optional",
                },
            )

        binary_artifacts: list[LookupSingle] = []
        if override_binary_artifacts is None:
            # Use the default binary artifacts for all the architectures
            for arch in add_new_binary_artifacts_archs:
                binary_artifacts.append(
                    self.new_binary_artifacts[Architecture(arch)].id
                )

        else:
            # Use the override_binary_artifacts.
            # Calculate the relevant archs based on the passed artifacts
            binary_artifacts = override_binary_artifacts

        sbuild_archs = self.archs_for_binaries(binary_artifacts)

        task_data: dict[str, Any] = {
            "source_artifact": (
                override_source_artifact
                if override_source_artifact is not None
                else self.new_source_artifact.id
            ),
            "binary_artifacts": binary_artifacts,
            "original": original or str(self.debian_collection),
            "vendor": "debian",
            "codename": "bookworm",
        }

        if arch_all_build_architecture is not None:
            task_data["arch_all_build_architecture"] = (
                arch_all_build_architecture
            )

        if extra_flags is not None:
            task_data["extra_flags"] = extra_flags

        example_pipeline = self.orchestrate(
            task_data=DebDiffWorkflowData(**task_data),
            sbuild_architectures=list(sbuild_archs),
            source_artifact=self.new_source_artifact,
            target_distribution="debian:bookworm",
            expect_success=expect_success,
            use_data=use_data,
        )
        work_request = example_pipeline.children.get(
            task_name="debdiff", task_type=TaskTypes.WORKFLOW
        )
        self.task_set_current(self.get_workflow(work_request))

    def archs_for_binaries(self, binaries: list[LookupSingle]) -> set[str]:
        archs = set()
        for lookup in binaries:
            archs |= workflow_utils.lookup_result_architectures(
                lookup_single(
                    lookup,
                    self.workspace,
                    pc=PermissionContext.for_ui_user(
                        self.playground.get_default_user()
                    ),
                    expect_type=LookupChildType.ARTIFACT,
                )
            )

        return archs

    @classmethod
    def new_binary_artifact_lookup(cls, arch: str) -> int:
        return cls.new_binary_artifacts[Architecture(arch)].id

    @preserve_db_task_registry()
    def test_populate(self) -> None:
        # We'll deliberately add a binary package newer than the one we're
        # interested in, so allow this for test purposes
        self.debian_collection.data["may_rewind_versions"] = True
        self.debian_collection.save()

        # Add a binary package with older version: it will be discarded
        old_package_1 = self.playground.create_minimal_binary_package_artifact(
            package="libhello0",
            srcpkg_name="hello",
            srcpkg_version="0.1.0",
            version="0.1.0",
            architecture="amd64",
        )
        self.debian_collection.manager.add_artifact(
            old_package_1,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )
        old_package_2 = self.playground.create_minimal_binary_package_artifact(
            package="libhello0",
            srcpkg_name="hello",
            srcpkg_version="0.2.0",
            version="0.2.0",
            architecture="amd64",
        )
        self.debian_collection.manager.add_artifact(
            old_package_2,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

        # Add another binary package with same version but new architecture
        new_architecture = (
            self.playground.create_minimal_binary_package_artifact(
                package="hello",
                srcpkg_name="hello",
                srcpkg_version="1.0.0",
                version="1.0.0",
                architecture="riscv64",
            )
        )
        self.debian_collection.manager.add_artifact(
            new_architecture,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

        # Add a binary package that is ignored because its srcpkg_version
        # is too new (when known, it's not using the most recent binary package
        # but one with a matching srcpkg_version)
        too_new = self.playground.create_minimal_binary_package_artifact(
            package="hello",
            srcpkg_name="hello",
            srcpkg_version="3.0.0",
            version="3.0.0",
            architecture="amd64",
        )
        self.debian_collection.manager.add_artifact(
            too_new,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

        # Create workflow
        with patch.object(
            DebDiffWorkflow,
            "requires_artifact",
            wraps=DebDiffWorkflow.requires_artifact,
        ) as requires_artifact:
            self.setup_debdiff_workflow(
                arch_all_build_architecture="arm64",
                extra_flags=[DebDiffFlags.IGNORE_SPACE],
            )

        debdiff_work_requests = self.work_request.children.all()
        bookworm = self.debian_collection
        bookworm_lookup = f"{bookworm.name}@{bookworm.category}"

        # Check work request to compare the source packages
        source_work_request = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-source"
        )
        self.assertEqual(
            source_work_request.workflow_data.display_name,
            "DebDiff for source package",
        )

        self.assertEqual(
            source_work_request.task_data,
            {
                "environment": "debian/match:codename=bookworm",
                "build_architecture": "arm64",
                "extra_flags": [DebDiffFlags.IGNORE_SPACE],
                "input": {
                    "source_artifacts": [
                        f"{bookworm_lookup}/name:hello_1.0.0",
                        self.new_source_artifact.id,
                    ]
                },
            },
        )

        self.assertIn(
            call(
                source_work_request,
                f"{bookworm_lookup}/name:hello_1.0.0",
            ),
            requires_artifact.call_args_list,
        )
        self.assertIn(
            call(source_work_request, self.new_source_artifact.id),
            requires_artifact.call_args_list,
        )

        # Check work requests to compare the binary packages
        self.assertEqual(
            debdiff_work_requests.filter(
                workflow_data_json__step__startswith="debdiff-binaries"
            ).count(),
            2,
        )

        amd64_work_request = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-binaries-amd64"
        )

        self.assertEqual(
            amd64_work_request.task_data,
            {
                "environment": "debian/match:codename=bookworm",
                "build_architecture": "amd64",
                "extra_flags": [DebDiffFlags.IGNORE_SPACE],
                "input": {
                    "binary_artifacts": [
                        [
                            f"{bookworm_lookup}/name:hello_1.0.0_all",
                            f"{bookworm_lookup}/name:hello_1.0.0_amd64",
                        ],
                        sorted(
                            [
                                self.new_binary_artifact_lookup("all"),
                                self.new_binary_artifact_lookup("amd64"),
                            ]
                        ),
                    ]
                },
            },
        )

        self.assertIn(
            call(
                amd64_work_request,
                LookupMultiple(
                    (
                        f"{bookworm_lookup}/name:hello_1.0.0_all",
                        f"{bookworm_lookup}/name:hello_1.0.0_amd64",
                    )
                ),
            ),
            requires_artifact.call_args_list,
        )
        self.assertIn(
            call(
                amd64_work_request,
                LookupMultiple.model_validate(
                    sorted(
                        [
                            self.new_binary_artifact_lookup("amd64"),
                            self.new_binary_artifact_lookup("all"),
                        ]
                    )
                ),
            ),
            requires_artifact.call_args_list,
        )

    @preserve_db_task_registry()
    def test_populate_binary_artifacts_upload(self) -> None:
        upload_artifacts = self.playground.create_upload_artifacts(
            binaries=[("hello", "amd64"), ("bye", "amd64")],
        )

        self.setup_debdiff_workflow(
            override_binary_artifacts=[upload_artifacts.upload.id]
        )

        debdiff_work_requests = self.work_request.children.all()

        work_request_source = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-source"
        )
        self.assertEqual(
            work_request_source.task_data,
            {
                "build_architecture": "amd64",
                "environment": "debian/match:codename=bookworm",
                "extra_flags": [],
                "input": {
                    "source_artifacts": [
                        "bookworm@debian:suite/name:hello_1.0.0",
                        self.new_source_artifact.id,
                    ]
                },
            },
        )

        work_request_binary = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-binaries-amd64"
        )

        self.assertEqual(
            work_request_binary.task_data,
            {
                "build_architecture": "amd64",
                "environment": "debian/match:codename=bookworm",
                "extra_flags": [],
                "input": {
                    "binary_artifacts": [
                        [
                            "bookworm@debian:suite/name:hello_1.0.0_all",
                            "bookworm@debian:suite/name:hello_1.0.0_amd64",
                        ],
                        sorted(
                            [
                                artifact.id
                                for artifact in upload_artifacts.binaries
                            ]
                        ),
                    ]
                },
            },
        )

    @preserve_db_task_registry()
    def test_get_new_packages_by_name_arch_invalid_category(self) -> None:
        # Valid artifact categories are tested as part of other tests, such as
        # test_populate_binary_artifacts_upload(). The invalid category
        # path should never be reached in production
        source_artifact = self.playground.create_source_artifact()
        self.setup_debdiff_workflow()

        # Make workflow.data.binary_artifacts contain an artifact of an
        # unexpected category
        self.task.data.binary_artifacts = LookupMultiple((source_artifact.id,))
        with self.assertRaisesRegex(
            ValueError, "Unexpected category 'debian:source-package'"
        ):
            self.task._get_new_packages_by_name_arch()

    @preserve_db_task_registry()
    def test_populate_new_binary_artifacts_only_all_arch(self) -> None:
        self.setup_debdiff_workflow(
            add_new_binary_artifacts_archs=["all"],
            arch_all_build_architecture="arm64",
        )

        debdiff_work_requests = self.work_request.children.all()

        # Check work requests to compare the binary packages
        self.assertEqual(
            debdiff_work_requests.filter(
                workflow_data_json__step__startswith="debdiff-binaries"
            ).count(),
            1,
        )

        all_work_request = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-binaries-all"
        )

        bookworm = self.debian_collection
        bookworm_lookup = f"{bookworm.name}@{bookworm.category}"

        self.assertEqual(
            all_work_request.task_data,
            {
                "environment": "debian/match:codename=bookworm",
                "build_architecture": "arm64",
                "extra_flags": [],
                "input": {
                    "binary_artifacts": [
                        [
                            f"{bookworm_lookup}/name:hello_1.0.0_all",
                        ],
                        [
                            self.new_binary_artifact_lookup("all"),
                        ],
                    ]
                },
            },
        )

    @preserve_db_task_registry()
    def test_populate_origin_binary_artifacts_no_all_arch(self) -> None:
        self.setup_debdiff_workflow(
            add_origin_binary_artifacts_archs=["amd64", "arm64"]
        )

        debdiff_work_requests = self.work_request.children.all()

        # Check work requests to compare the binary packages
        self.assertEqual(
            debdiff_work_requests.filter(
                workflow_data_json__step__startswith="debdiff-binaries"
            ).count(),
            2,
        )

        amd64_work_request = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-binaries-amd64"
        )

        bookworm = self.debian_collection
        bookworm_lookup = f"{bookworm.name}@{bookworm.category}"

        self.assertEqual(
            amd64_work_request.task_data,
            {
                "environment": "debian/match:codename=bookworm",
                "build_architecture": "amd64",
                "extra_flags": [],
                "input": {
                    "binary_artifacts": [
                        [
                            f"{bookworm_lookup}/name:hello_1.0.0_amd64",
                        ],
                        sorted(
                            [
                                self.new_binary_artifact_lookup("amd64"),
                                self.new_binary_artifact_lookup("all"),
                            ]
                        ),
                    ]
                },
            },
        )

    @preserve_db_task_registry()
    def test_populate_new_package(self) -> None:
        """Run populate with a package not found in original collection."""
        self.setup_debdiff_workflow(
            add_origin_source_artifact=False,
            add_origin_binary_artifacts_archs=[],
        )

        # Source package neither binary packages could be found in the
        # original collection. It's a new package, and there isn't any
        # DebDiff task created.
        self.assertEqual(self.work_request.children.count(), 0)

    @preserve_db_task_registry()
    def test_populate_package_without_source_in_original(self) -> None:
        # Add binary packages that have the same name but are older
        # (to exercise returning only the newest)
        for version in ("0.0.1", "0.1.0"):
            self.debian_collection.manager.add_artifact(
                self.playground.create_minimal_binary_package_artifact(
                    package="hello",
                    srcpkg_name="hello",
                    srcpkg_version=version,
                    version=version,
                    architecture="amd64",
                ),
                user=self.playground.get_default_user(),
                variables={
                    "component": "main",
                    "section": "devel",
                    "priority": "optional",
                },
            )

        self.setup_debdiff_workflow(
            add_origin_source_artifact=False,
        )

        debdiff_work_requests = self.work_request.children.all()

        # Check work requests to compare the binary packages
        self.assertEqual(
            debdiff_work_requests.filter(
                workflow_data_json__step__startswith="debdiff-binaries"
            ).count(),
            2,
        )

        # The version 0.1.0 is not used, only 1.0.0 versions
        for debdiff_work_request in debdiff_work_requests:
            for binary_artifact in debdiff_work_request.task_data["input"][
                "binary_artifacts"
            ][0]:
                self.assertIn("1.0.0", binary_artifact)

    @preserve_db_task_registry()
    def test_populate_source_package_upload_artifact(self) -> None:
        upload_artifact = self.playground.create_upload_artifacts(
            src_name="hello",
            binary=False,
        )

        self.setup_debdiff_workflow(
            override_source_artifact=f"{upload_artifact.upload.id}@artifacts"
        )

        source_work_request = self.work_request.children.get(
            workflow_data_json__step="debdiff-source"
        )

        new_source_artifact = source_work_request.task_data["input"][
            "source_artifacts"
        ][1]

        self.assertEqual(
            new_source_artifact, f"{upload_artifact.source.id}@artifacts"
        )

    @preserve_db_task_registry()
    def test_populate_one_debdiff_per_concrete_arch_includes_all_binaries(
        self,
    ) -> None:
        self.setup_debdiff_workflow()

        debdiff_work_requests = self.work_request.children.all()

        expected_archs = {"amd64", "arm64"}

        binary_wrs = debdiff_work_requests.filter(
            workflow_data_json__step__startswith="debdiff-binaries"
        )

        # One per expected concrete arch (without "all")
        steps = binary_wrs.values_list("workflow_data_json__step", flat=True)
        self.assertCountEqual(
            steps,
            {f"debdiff-binaries-{arch}" for arch in expected_archs},
        )

        bookworm = self.debian_collection
        bookworm_lookup = f"{bookworm.name}@{bookworm.category}"

        for arch in expected_archs:
            wr = binary_wrs.get(
                workflow_data_json__step=f"debdiff-binaries-{arch}"
            )
            originals, news = wr.task_data["input"]["binary_artifacts"]

            # Originals: "all" + specific architecture packages
            self.assertEqual(
                set(originals),
                {
                    f"{bookworm_lookup}/name:hello_1.0.0_all",
                    f"{bookworm_lookup}/name:hello_1.0.0_{arch}",
                },
            )

            # news: "all" + specific architecture packages
            self.assertEqual(
                set(news),
                {
                    self.new_binary_artifact_lookup("all"),
                    self.new_binary_artifact_lookup(arch),
                },
            )

    @preserve_db_task_registry()
    def test_populate_binary_for_all_does_not_include_arch_specific_originals(
        self,
    ) -> None:
        self.setup_debdiff_workflow(
            add_new_binary_artifacts_archs=["all"],
        )

        debdiff_work_requests = self.work_request.children.all()
        all_work_request = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-binaries-all"
        )

        originals, _ = all_work_request.task_data["input"]["binary_artifacts"]
        debian_collection_lookup = (
            f"{self.debian_collection.name}@{self.debian_collection.category}"
        )

        self.assertEqual(
            originals, [f"{debian_collection_lookup}/name:hello_1.0.0_all"]
        )

    @preserve_db_task_registry()
    def test_populate_origin_architecture_from_any_to_all(self) -> None:
        """Binary debdiff tasks created if binaries change from any to all."""
        self.setup_debdiff_workflow(
            add_origin_binary_artifacts_archs=["amd64", "arm64"],
            add_new_binary_artifacts_archs=["all"],
        )

        debdiff_work_requests = self.work_request.children.all()

        binary_wrs = debdiff_work_requests.filter(
            workflow_data_json__step__startswith="debdiff-binaries"
        )

        expected_archs = {"amd64", "arm64"}

        steps = binary_wrs.values_list("workflow_data_json__step", flat=True)

        # Correct archs
        self.assertCountEqual(
            steps,
            {f"debdiff-binaries-{arch}" for arch in expected_archs},
        )

        bookworm = self.debian_collection
        bookworm_lookup = f"{bookworm.name}@{bookworm.category}"

        # Correct task_data per architecture
        amd64_work_request = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-binaries-amd64"
        )

        self.assertEqual(
            amd64_work_request.task_data,
            {
                "environment": "debian/match:codename=bookworm",
                "build_architecture": "amd64",
                "extra_flags": [],
                "input": {
                    "binary_artifacts": [
                        [f"{bookworm_lookup}/name:hello_1.0.0_amd64"],
                        [self.new_binary_artifact_lookup("all")],
                    ]
                },
            },
        )

        arm64_work_request = debdiff_work_requests.get(
            workflow_data_json__step="debdiff-binaries-arm64"
        )

        self.assertEqual(
            arm64_work_request.task_data,
            {
                "environment": "debian/match:codename=bookworm",
                "build_architecture": "arm64",
                "extra_flags": [],
                "input": {
                    "binary_artifacts": [
                        [f"{bookworm_lookup}/name:hello_1.0.0_arm64"],
                        [self.new_binary_artifact_lookup("all")],
                    ]
                },
            },
        )

    @preserve_db_task_registry()
    def test_compute_scheduler_tags(self) -> None:
        self.setup_debdiff_workflow()
        self.assertComputesSchedulerTags()

    @preserve_db_task_registry()
    def test_compute_dynamic_data_from_source(self) -> None:
        self.setup_debdiff_workflow()
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            DebDiffWorkflowDynamicData(
                subject="hello",
                parameter_summary="bookworm: 1.0.0 → 2.0.0",
                original_id=self.origin_source_artifact.id,
                source_artifact_id=self.new_source_artifact.id,
                binary_artifacts_ids=[
                    artifact.id
                    for artifact in self.new_binary_artifacts.values()
                ],
            ),
        )

    @preserve_db_task_registry()
    def test_get_input_artifacts(self) -> None:
        self.setup_debdiff_workflow()
        self.task.dynamic_data = self.compute_dynamic_data(self.task)
        self.assertEqual(
            self.task.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=self.task.data.source_artifact,
                    label="source_artifact",
                    artifact_id=self.new_source_artifact.id,
                ),
                InputArtifactSingle(
                    lookup=self.task.data.original,
                    label="original",
                    artifact_id=self.origin_source_artifact.id,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=self.task.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=[
                        artifact.id
                        for artifact in self.new_binary_artifacts.values()
                    ],
                ),
            ],
        )

    @preserve_db_task_registry()
    def test_compute_dynamic_data_from_binary(self) -> None:
        # Add a binary package with older version: it will be discarded
        old_package = self.playground.create_minimal_binary_package_artifact(
            package="libhello0",
            srcpkg_name="hello",
            srcpkg_version="0.1.0",
            version="0.1.0",
            architecture="amd64",
        )
        self.debian_collection.manager.add_artifact(
            old_package,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )
        self.setup_debdiff_workflow(add_origin_source_artifact=False)

        self.assertEqual(
            self.compute_dynamic_data(self.task),
            DebDiffWorkflowDynamicData(
                subject="hello",
                parameter_summary="bookworm: 1.0.0 → 2.0.0",
                original_id=None,
                source_artifact_id=self.new_source_artifact.id,
                binary_artifacts_ids=[
                    artifact.id
                    for artifact in self.new_binary_artifacts.values()
                ],
            ),
        )

    @preserve_db_task_registry()
    def test_compute_dynamic_data_package_not_in_origin_collection(
        self,
    ) -> None:
        self.setup_debdiff_workflow(
            add_origin_source_artifact=False,
            add_origin_binary_artifacts_archs=[],
        )
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            DebDiffWorkflowDynamicData(
                subject="hello",
                parameter_summary="bookworm: missing → 2.0.0",
                original_id=None,
                source_artifact_id=self.new_source_artifact.id,
                binary_artifacts_ids=[
                    artifact.id
                    for artifact in self.new_binary_artifacts.values()
                ],
            ),
        )

    def test_package_arch_pairs_artifact_debian_binary_package(self) -> None:
        temp_dir = self.create_temporary_directory()

        artifact = self.playground.create_artifact_from_local(
            self.playground.create_binary_package(
                temp_dir, name="hello", architecture="arm64"
            )
        )

        lookup_result = lookup_single(
            f"{artifact.id}@artifacts",
            workspace=self.workspace,
            pc=PermissionContext.for_ui_user(
                self.playground.get_default_user()
            ),
        )

        self.assertEqual(
            DebDiffWorkflow._package_arch_pairs(lookup_result),
            [("hello", "arm64")],
        )

    def test_package_arch_pairs_artifact_raise_value_error(self) -> None:
        self.playground.create_bare_data_item(
            self.debian_collection,
            "build-amd64",
            category=BareDataCategory.PROMISE,
            data={
                "promise_work_request_id": 2,
                "promise_workflow_id": 1,
                "promise_category": ArtifactCategory.TEST,
            },
        )

        lookup_result = lookup_single(
            f"{self.debian_collection}/name:build-amd64",
            self.workspace,
            pc=PermissionContext.for_ui_user(
                self.playground.get_default_user()
            ),
        )

        with self.assertRaisesRegex(
            ValueError,
            r"^DebDiffWorkflow requires binary artifacts \(not promises\)$",
        ):
            DebDiffWorkflow._package_arch_pairs(lookup_result)

    def test_package_arch_pairs_artifact_unsupported_category(self) -> None:
        artifact = self.playground.create_source_artifact()

        lookup_result = lookup_single(
            f"{artifact.id}@artifacts",
            workspace=self.workspace,
            pc=PermissionContext.for_ui_user(
                self.playground.get_default_user()
            ),
        )

        with self.assertRaisesRegex(
            ValueError, rf"Unexpected category {artifact.category}"
        ):
            DebDiffWorkflow._package_arch_pairs(lookup_result)

    def test_package_arch_pairs_artifact_upload(self) -> None:
        # Introspects the relations
        upload_artifacts = self.playground.create_upload_artifacts(
            binaries=[("bye", "all"), ("hello", "all")],
        )

        lookup_result = lookup_single(
            f"{upload_artifacts.upload.id}@artifacts",
            workspace=self.workspace,
            pc=PermissionContext.for_ui_user(
                self.playground.get_default_user()
            ),
        )
        artifact_amd64 = upload_artifacts.binaries[0]
        artifact_amd64.data["deb_fields"]["Architecture"] = "amd64"
        artifact_amd64.save()

        artifact_1 = upload_artifacts.binaries[1]

        self.assertEqual(
            DebDiffWorkflow._package_arch_pairs(lookup_result),
            sorted(
                [
                    (artifact_amd64.data["deb_fields"]["Package"], "amd64"),
                    (artifact_1.data["deb_fields"]["Package"], "all"),
                ]
            ),
        )

    @preserve_db_task_registry()
    def test_validate_input(self) -> None:
        # validate_input() is called and does not raise
        self.setup_debdiff_workflow()

    @preserve_db_task_registry()
    def test_validate_input_invalid_original(self) -> None:
        self.setup_debdiff_workflow(
            original="non-existing", expect_success=False, use_data=True
        )
        # TODO: This should arguably be logged on the debdiff workflow
        # itself, not on its parent.  At the moment,
        # Workflow.orchestrate_child calls mark_pending which fails, and the
        # exception is handled by the parent.
        assert self.work_request.parent is not None
        self.assertEqual(
            self.work_request.parent.output_data,
            OutputData(
                errors=[
                    OutputDataError(
                        message=(
                            "Orchestrator failed: 'non-existing' does not "
                            "specify a category and the context does not "
                            "supply a default"
                        ),
                        code="orchestrator-failed",
                    )
                ]
            ),
        )
