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

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

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

from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianLintian,
    DebianLintianSummary,
    TaskTypes,
)
from debusine.client.models import LookupChildType
from debusine.db.models import (
    Artifact,
    ArtifactRelation,
    Collection,
    CollectionItem,
    WorkRequest,
)
from debusine.db.models.work_requests import SkipWorkRequest
from debusine.server.workflows import LintianWorkflow, WorkflowValidationError
from debusine.server.workflows.models import (
    BaseWorkflowData,
    LintianWorkflowData,
    LintianWorkflowDynamicData,
    RegressionTrackingUpdateQAResults,
    SbuildWorkflowData,
)
from debusine.server.workflows.tests.helpers import (
    SampleWorkflow,
    WorkflowTestCase,
)
from debusine.server.workflows.tests.test_regression_tracking import (
    RegressionTrackingWorkflowTestMixin,
)
from debusine.tasks.lintian import Lintian
from debusine.tasks.models import (
    ActionSkipIfLookupResultChanged,
    ActionUpdateCollectionWithArtifacts,
    BackendType,
    BaseDynamicTaskData,
    InputArtifactMultiple,
    InputArtifactSingle,
    LintianFailOnSeverity,
    LintianOutput,
    LookupMultiple,
    OutputData,
    RegressionAnalysisStatus,
    SbuildData,
    SbuildInput,
)
from debusine.test.django import preserve_db_task_registry


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

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

    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.id
        self.task_data["binary_artifacts"] = LookupMultiple(
            (self.binary_artifact.id,)
        )

    def create_lintian_workflow(
        self,
        *,
        extra_task_data: dict[str, Any] | None = None,
        validate: bool = True,
    ) -> LintianWorkflow:
        """Create a lintian workflow."""
        binary_artifact_1 = (
            self.playground.create_minimal_binary_package_artifact()
        )
        binary_artifact_2 = (
            self.playground.create_minimal_binary_package_artifact()
        )
        task_data = {
            "source_artifact": self.source_artifact.pk,
            "binary_artifacts": [
                "internal@collections/name:build-amd64",
                "internal@collections/name:build-i386",
            ],
            "vendor": "debian",
            "codename": "bookworm",
        }
        if extra_task_data is not None:
            task_data.update(extra_task_data)

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=task_data,
            validate=validate,
            internal_collection_artifacts=[
                (binary_artifact_1, {"name": "build-amd64"}),
                (binary_artifact_2, {"name": "build-i386"}),
            ],
        )

        return self.get_workflow(wr)

    def _expected_task_display_name(
        self, architecture: str, lintian: WorkRequest
    ) -> str:
        """Return the expected display name for a lintian task."""
        display_name = f"Lintian for {architecture}"
        output = lintian.task_data["output"]

        binary_all_analysis = output.get("binary_all_analysis", True)
        source_analysis = output.get("source_analysis", True)

        if architecture != "all" and binary_all_analysis:
            display_name += "+all"

        if source_analysis:
            display_name += "+source"

        return display_name

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

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

            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=list(architectures) or ["all"],
                    ),
                )
                self.playground.advance_work_request(sbuild, mark_running=True)

                for arch in architectures:
                    child = sbuild.create_child_worker(
                        task_name="sbuild",
                        task_data=SbuildData(
                            input=SbuildInput(
                                source_artifact=task_data.source_artifact
                            ),
                            build_architecture=arch,
                            environment="debian/match:codename=sid",
                        ),
                    )

                    self_.provides_artifact(
                        child,
                        ArtifactCategory.BINARY_PACKAGE,
                        f"build-{arch}",
                        data={"architecture": arch},
                    )
                sbuild.unblock_workflow_children()

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

        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 add_lintian_binary(
        self, suite: Collection, version: str, architecture: str = "all"
    ) -> None:
        """Add a lintian binary package to a suite."""
        artifact = self.playground.create_minimal_binary_package_artifact(
            "lintian", version, "lintian", version, architecture
        )
        suite.manager.add_artifact(
            artifact,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

    def add_qa_result(
        self,
        qa_results: Collection,
        package: str,
        version: str,
        architecture: str,
        lintian_version: str,
    ) -> CollectionItem:
        """Add a lintian result to a ``debian:qa-results`` collection."""
        work_request = self.playground.create_worker_task(
            task_name="lintian",
            result=WorkRequest.Results.SUCCESS,
            validate=False,
            use_data=False,
        )
        return qa_results.manager.add_artifact(
            self.playground.create_artifact(
                category=ArtifactCategory.LINTIAN,
                data=DebianLintian(
                    architecture=architecture,
                    summary=DebianLintianSummary(
                        tags_count_by_severity={},
                        package_filename={},
                        tags_found=[],
                        overridden_tags_found=[],
                        lintian_version=lintian_version,
                        distribution="debian:sid",
                    ),
                ),
            )[0],
            user=self.playground.get_default_user(),
            variables={
                "package": package,
                "version": version,
                "architecture": architecture,
                "timestamp": int(timezone.now().timestamp()),
                "work_request_id": work_request.id,
            },
        )

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

        w.validate_input()

    def test_validate_input_bad_qa_suite(self) -> None:
        """validate_input raises errors in looking up a suite."""
        w = self.create_lintian_workflow(
            extra_task_data={"qa_suite": "nonexistent@debian:suite"},
            validate=False,
        )

        with self.assertRaisesRegex(
            WorkflowValidationError,
            "'nonexistent@debian:suite' does not exist or is hidden",
        ):
            w.validate_input()

    def test_validate_input_bad_reference_qa_results(self) -> None:
        """validate_input raises errors in looking up reference QA results."""
        w = self.create_lintian_workflow(
            extra_task_data={
                "reference_qa_results": "nonexistent@debian:qa-results"
            },
            validate=False,
        )

        with self.assertRaisesRegex(
            WorkflowValidationError,
            "'nonexistent@debian:qa-results' does not exist or is hidden",
        ):
            w.validate_input()

    def test_has_current_reference_qa_result_no_match(self) -> None:
        """_has_current_reference_qa_result: no matching result."""
        sid_suite = self.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(sid_suite, "2.116.3")
        self.add_qa_result(sid_qa_results, "other", "1.0-1", "amd64", "2.116.3")

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                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.get_workflow(wr)._has_current_reference_qa_result(
                architecture="amd64", build_architecture="amd64"
            )
        )

    def test_has_current_reference_qa_result_different_version(self) -> None:
        """_has_current_reference_qa_result: result for different version."""
        sid_suite = self.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(sid_suite, "2.116.3")
        self.add_qa_result(sid_qa_results, "hello", "1.0-1", "amd64", "2.116.3")
        source_artifact = self.playground.create_source_artifact(
            name="hello", version="1.0-2"
        )

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                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.get_workflow(wr)._has_current_reference_qa_result(
                architecture="amd64", build_architecture="amd64"
            )
        )

    def test_has_current_reference_qa_result_old_lintian_version(self) -> None:
        """_has_current_reference_qa_result: run with old lintian version."""
        sid_suite = self.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(sid_suite, "2.116.3")
        self.add_qa_result(sid_qa_results, "hello", "1.0-1", "amd64", "2.15.0")

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                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.get_workflow(wr)._has_current_reference_qa_result(
                architecture="amd64", build_architecture="amd64"
            )
        )

    def test_has_current_reference_qa_result_no_lintian_version(self) -> None:
        """_has_current_reference_qa_result: no reference lintian 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", "2.116.3")

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                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.get_workflow(wr)._has_current_reference_qa_result(
                architecture="amd64", build_architecture="amd64"
            )
        )

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

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                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.get_workflow(wr)._has_current_reference_qa_result(
                architecture="amd64", build_architecture="amd64"
            )
        )

    def test_has_current_reference_qa_result_honours_build_architecture(
        self,
    ) -> None:
        """_has_current_reference_qa_result: lintian checked by build arch."""
        sid_suite = self.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        sid_qa_results = self.playground.create_collection(
            "sid", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(sid_suite, "2.116.3")
        self.add_qa_result(sid_qa_results, "hello", "1.0-1", "all", "2.116.3")

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                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.get_workflow(wr)._has_current_reference_qa_result(
                architecture="all", build_architecture="amd64"
            )
        )

        self.add_lintian_binary(sid_suite, "2.117.0", architecture="amd64")

        self.assertFalse(
            self.get_workflow(wr)._has_current_reference_qa_result(
                architecture="all", build_architecture="amd64"
            )
        )

    @preserve_db_task_registry()
    def test_populate(self) -> None:
        """Test populate."""
        architectures = ["amd64", "i386", "all"]

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                architectures=architectures,
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in architectures
                    ]
                ),
                vendor="debian",
                codename="trixie",
            ),
            architectures=architectures,
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        lintians = workflow.children.filter(task_name="lintian")

        # architecture "all" does not have a lintian job
        self.assertEqual(lintians.count(), len(architectures) - 1)

        for architecture in architectures:
            if architecture == "all":
                continue

            lintian = lintians.get(
                workflow_data_json__step=f"lintian-{architecture}"
            )

            self.assertEqual(
                lintian.task_data,
                {
                    "backend": Lintian.DEFAULT_BACKEND,
                    "build_architecture": architecture,
                    "environment": "debian/match:codename=trixie",
                    "exclude_tags": [],
                    "fail_on_severity": LintianFailOnSeverity.ERROR,
                    "include_tags": [],
                    "input": {
                        "binary_artifacts": sorted(
                            [
                                f"internal@collections/name:build-{arch}"
                                for arch in (architecture, "all")
                            ]
                        ),
                        "source_artifact": self.source_artifact.id,
                    },
                    "output": (
                        {"source_analysis": False, "binary_all_analysis": False}
                        if architecture != "amd64"
                        else {}
                    ),
                    "target_distribution": "debian:trixie",
                },
            )

            self.assertEqual(
                lintian.workflow_data_json,
                {
                    "display_name": self._expected_task_display_name(
                        architecture, lintian
                    ),
                    "step": f"lintian-{architecture}",
                },
            )

            self.assertQuerySetEqual(
                lintian.dependencies.all(),
                set(
                    WorkRequest.objects.filter(
                        task_name="sbuild",
                        task_data__build_architecture__in=(architecture, "all"),
                    )
                ),
                ordered=False,
            )

    @preserve_db_task_registry()
    def test_populate_source_only(self) -> None:
        """Test populate with no binary artifacts."""
        root = self.orchestrate(
            task_data=LintianWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple(()),
                vendor="debian",
                codename="trixie",
            ),
            architectures=("all",),
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        lintian = workflow.children.get(task_name="lintian")

        self.assertEqual(
            lintian.task_data,
            {
                "backend": BackendType.UNSHARE,
                "build_architecture": "amd64",
                "environment": "debian/match:codename=trixie",
                "exclude_tags": [],
                "fail_on_severity": LintianFailOnSeverity.ERROR,
                "include_tags": [],
                "input": {
                    "binary_artifacts": [],
                    "source_artifact": self.source_artifact.id,
                },
                "output": {},
                "target_distribution": "debian:trixie",
            },
        )

    @preserve_db_task_registry()
    def test_populate_without_architectures(self) -> None:
        """Test populate with only arch-indep packages."""
        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "hello", "1.0.0", "all"
            )
        )

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple((binary_artifact.id,)),
                vendor="debian",
                codename="trixie",
            ),
            architectures=("all",),
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        lintian = workflow.children.get(task_name="lintian")

        self.assertEqual(
            lintian.task_data,
            {
                "backend": BackendType.UNSHARE,
                "build_architecture": "amd64",
                "environment": "debian/match:codename=trixie",
                "exclude_tags": [],
                "fail_on_severity": LintianFailOnSeverity.ERROR,
                "include_tags": [],
                "input": {
                    "binary_artifacts": [binary_artifact.id],
                    "source_artifact": self.source_artifact.id,
                },
                "output": {},
                "target_distribution": "debian:trixie",
            },
        )

    @preserve_db_task_registry()
    def test_populate_without_architectures_arch_all_build_architecture(
        self,
    ) -> None:
        """Workflow honours `arch_all_build_architecture`."""
        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "hello", "1.0.0", "all"
            )
        )

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple((binary_artifact.id,)),
                vendor="debian",
                codename="trixie",
                arch_all_build_architecture="s390x",
            ),
            architectures=("all",),
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        lintian = workflow.children.get(task_name="lintian")

        self.assertEqual(
            lintian.task_data,
            {
                "backend": BackendType.UNSHARE,
                "build_architecture": "s390x",
                "environment": "debian/match:codename=trixie",
                "exclude_tags": [],
                "fail_on_severity": LintianFailOnSeverity.ERROR,
                "include_tags": [],
                "input": {
                    "binary_artifacts": [binary_artifact.id],
                    "source_artifact": self.source_artifact.id,
                },
                "output": {},
                "target_distribution": "debian:trixie",
            },
        )

    @preserve_db_task_registry()
    def test_populate_mixed_without_arch_all_build_architecture(self) -> None:
        """Mixed all/any binaries but without the usual arch-all arch."""
        binary_artifacts = {
            "all": self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "hello", "1.0.0", "all"
            ),
            "i386": self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "libhello1", "1.0.0", "i386"
            ),
        }

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple(
                    (binary_artifacts["all"].id, binary_artifacts["i386"].id)
                ),
                vendor="debian",
                codename="trixie",
                arch_all_build_architecture="amd64",
            ),
            architectures=("all", "i386"),
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        self.assertQuerySetEqual(
            workflow.children.filter(task_name="lintian").values_list(
                "task_data", flat=True
            ),
            [
                {
                    "backend": BackendType.UNSHARE,
                    "build_architecture": "i386",
                    "environment": "debian/match:codename=trixie",
                    "exclude_tags": [],
                    "fail_on_severity": LintianFailOnSeverity.ERROR,
                    "include_tags": [],
                    "input": {
                        "binary_artifacts": sorted(
                            [
                                binary_artifacts["all"].id,
                                binary_artifacts["i386"].id,
                            ]
                        ),
                        "source_artifact": self.source_artifact.id,
                    },
                    "output": {},
                    "target_distribution": "debian:trixie",
                }
            ],
        )

    @preserve_db_task_registry()
    def test_populate_arch_dep_without_arch_all_build_architecture(
        self,
    ) -> None:
        """``Architecture: any`` binaries without the usual arch-all arch."""
        binary_artifacts = {
            "i386": self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "libhello1", "1.0.0", "i386"
            ),
            "s390x": self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "libhello1", "1.0.0", "s390x"
            ),
        }

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple(
                    (binary_artifacts["i386"].id, binary_artifacts["s390x"].id)
                ),
                vendor="debian",
                codename="trixie",
                arch_all_build_architecture="amd64",
            ),
            architectures=("i386", "s390x"),
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        self.assertQuerySetEqual(
            workflow.children.filter(task_name="lintian")
            .order_by("id")
            .values_list("task_data", flat=True),
            [
                {
                    "backend": BackendType.UNSHARE,
                    "build_architecture": architecture,
                    "environment": "debian/match:codename=trixie",
                    "exclude_tags": [],
                    "fail_on_severity": LintianFailOnSeverity.ERROR,
                    "include_tags": [],
                    "input": {
                        "binary_artifacts": [binary_artifacts[architecture].id],
                        "source_artifact": self.source_artifact.id,
                    },
                    "output": (
                        {"source_analysis": False, "binary_all_analysis": False}
                        if architecture != "i386"
                        else {}
                    ),
                    "target_distribution": "debian:trixie",
                }
                for architecture in ("i386", "s390x")
            ],
        )

    @preserve_db_task_registry()
    def test_populate_with_no_architecture_overlap(self) -> None:
        """If architectures and binary artifacts are disjoint, do nothing."""
        root = self.orchestrate(
            task_data=LintianWorkflowData(
                architectures=["i386"],
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple(
                    ("internal@collections/name:build-amd64",)
                ),
                vendor="debian",
                codename="trixie",
            ),
            architectures=["amd64"],
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        self.assertFalse(workflow.children.filter(task_name="lintian").exists())

    @preserve_db_task_registry()
    def test_populate_no_binary_any_analysis(self) -> None:
        """Disabling binary-any analysis skips no-op work requests."""
        binary_artifacts = {
            "all": self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "hello", "1.0.0", "all"
            ),
            "amd64": self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "libhello1", "1.0.0", "amd64"
            ),
            "i386": self.playground.create_minimal_binary_package_artifact(
                "hello", "1.0.0", "libhello1", "1.0.0", "i386"
            ),
        }

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple(
                    (
                        binary_artifacts["all"].id,
                        binary_artifacts["amd64"].id,
                        binary_artifacts["i386"].id,
                    )
                ),
                vendor="debian",
                codename="trixie",
                arch_all_build_architecture="amd64",
                output=LintianOutput(binary_any_analysis=False),
            ),
            architectures=("all", "amd64", "i386"),
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        self.assertQuerySetEqual(
            workflow.children.filter(task_name="lintian").values_list(
                "task_data", flat=True
            ),
            [
                {
                    "backend": BackendType.UNSHARE,
                    "build_architecture": "amd64",
                    "environment": "debian/match:codename=trixie",
                    "exclude_tags": [],
                    "fail_on_severity": LintianFailOnSeverity.ERROR,
                    "include_tags": [],
                    "input": {
                        "binary_artifacts": sorted(
                            [
                                binary_artifacts["all"].id,
                                binary_artifacts["amd64"].id,
                            ]
                        ),
                        "source_artifact": self.source_artifact.id,
                    },
                    "output": {"binary_any_analysis": False},
                    "target_distribution": "debian:trixie",
                }
            ],
        )

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

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                architectures=architectures,
                source_artifact=upload_artifacts.upload.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in architectures
                    ]
                ),
                vendor="debian",
                codename="trixie",
            ),
            architectures=architectures,
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        lintians = workflow.children.filter(task_name="lintian")

        # architecture "all" does not have a lintian job
        self.assertEqual(lintians.count(), len(architectures) - 1)

        for architecture in architectures:
            if architecture == "all":
                continue

            lintian = lintians.get(
                workflow_data_json__step=f"lintian-{architecture}"
            )

            self.assertEqual(
                lintian.task_data,
                {
                    "backend": Lintian.DEFAULT_BACKEND,
                    "build_architecture": architecture,
                    "environment": "debian/match:codename=trixie",
                    "exclude_tags": [],
                    "fail_on_severity": LintianFailOnSeverity.ERROR,
                    "include_tags": [],
                    "input": {
                        "binary_artifacts": sorted(
                            [
                                f"internal@collections/name:build-{arch}"
                                for arch in (architecture, "all")
                            ]
                        ),
                        "source_artifact": (
                            f"{upload_artifacts.source.id}@artifacts"
                        ),
                    },
                    "output": (
                        {"source_analysis": False, "binary_all_analysis": False}
                        if architecture != "amd64"
                        else {}
                    ),
                    "target_distribution": "debian:trixie",
                },
            )

            self.assertEqual(
                lintian.workflow_data_json,
                {
                    "display_name": self._expected_task_display_name(
                        architecture, lintian
                    ),
                    "step": f"lintian-{architecture}",
                },
            )

            self.assertQuerySetEqual(
                lintian.dependencies.all(),
                set(
                    WorkRequest.objects.filter(
                        task_name="sbuild",
                        task_data__build_architecture__in=(architecture, "all"),
                    )
                ),
                ordered=False,
            )

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

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                source_artifact=artifacts.upload.id,
                binary_artifacts=LookupMultiple((artifacts.upload.id,)),
                architectures=architectures,
                vendor="debian",
                codename="sid",
            ),
        )
        self.get_workflow(wr).populate()
        children = list(wr.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")
                        ),
                    },
                )

    @preserve_db_task_registry()
    def test_populate_has_current_reference_qa_result(self) -> None:
        """The workflow does nothing with a current reference QA result."""
        trixie_suite = self.playground.create_collection(
            "trixie", CollectionCategory.SUITE
        )
        trixie_qa_results = self.playground.create_collection(
            "trixie", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(trixie_suite, "2.116.3")

        architectures = ["amd64", "i386", "all"]
        for architecture in architectures:
            self.add_qa_result(
                trixie_qa_results,
                "hello",
                self.source_artifact.data["version"],
                architecture,
                "2.116.3",
            )

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                prefix="reference-qa-result|",
                architectures=architectures,
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in architectures
                    ]
                ),
                vendor="debian",
                codename="trixie",
                qa_suite=f"trixie@{CollectionCategory.SUITE}",
                reference_qa_results=f"trixie@{CollectionCategory.QA_RESULTS}",
                update_qa_results=RegressionTrackingUpdateQAResults.YES,
            ),
            architectures=architectures,
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        self.assertQuerySetEqual(workflow.children.all(), [])

    @preserve_db_task_registry()
    def test_populate_has_current_reference_qa_result_force(self) -> None:
        """The workflow can be forced despite a current reference QA result."""
        trixie_suite = self.playground.create_collection(
            "trixie", CollectionCategory.SUITE
        )
        trixie_qa_results = self.playground.create_collection(
            "trixie", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(trixie_suite, "2.116.3")

        architectures = ["amd64", "i386", "all"]
        for architecture in architectures:
            self.add_qa_result(
                trixie_qa_results,
                "hello",
                self.source_artifact.data["version"],
                architecture,
                "2.116.3",
            )

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                prefix="reference-qa-result|",
                architectures=architectures,
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in architectures
                    ]
                ),
                vendor="debian",
                codename="trixie",
                qa_suite=f"trixie@{CollectionCategory.SUITE}",
                reference_qa_results=f"trixie@{CollectionCategory.QA_RESULTS}",
                update_qa_results=RegressionTrackingUpdateQAResults.FORCE,
            ),
            architectures=architectures,
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        self.assertTrue(
            workflow.children.filter(
                task_type=TaskTypes.WORKER, task_name="lintian"
            ).exists()
        )

    @preserve_db_task_registry()
    def test_populate_no_previous_reference_qa_result(self) -> None:
        """The workflow produces a reference QA result if there is none."""
        trixie_suite = self.playground.create_collection(
            "trixie", CollectionCategory.SUITE
        )
        self.playground.create_collection(
            "trixie", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(trixie_suite, "2.116.3")

        architectures = ["amd64", "i386", "all"]

        source_item = trixie_suite.manager.add_artifact(
            self.source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )

        root = self.orchestrate(
            task_data=LintianWorkflowData(
                prefix="reference-qa-result|",
                architectures=architectures,
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in architectures
                    ]
                ),
                vendor="debian",
                codename="trixie",
                qa_suite=f"trixie@{CollectionCategory.SUITE}",
                reference_qa_results=f"trixie@{CollectionCategory.QA_RESULTS}",
                update_qa_results=RegressionTrackingUpdateQAResults.YES,
            ),
            architectures=architectures,
        )

        workflow = WorkRequest.objects.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="lintian",
            parent=root,
        )

        lintians = workflow.children.filter(task_name="lintian")

        # architecture "all" does not have a lintian job
        concrete_architectures = [
            architecture
            for architecture in architectures
            if architecture != "all"
        ]
        self.assertEqual(lintians.count(), len(concrete_architectures))

        provide_actions: dict[str, ActionUpdateCollectionWithArtifacts] = {}
        skip_actions: dict[str, ActionSkipIfLookupResultChanged] = {}
        qa_result_actions: dict[str, ActionUpdateCollectionWithArtifacts] = {}
        results: dict[str, Artifact] = {}
        for architecture in concrete_architectures:
            lintian = lintians.get(
                workflow_data_json__step=f"lintian-{architecture}"
            )

            self.assertEqual(
                lintian.task_data,
                {
                    "backend": Lintian.DEFAULT_BACKEND,
                    "build_architecture": architecture,
                    "environment": "debian/match:codename=trixie",
                    "exclude_tags": [],
                    "fail_on_severity": LintianFailOnSeverity.ERROR,
                    "include_tags": [],
                    "input": {
                        "binary_artifacts": sorted(
                            [
                                f"internal@collections/name:build-{arch}"
                                for arch in (architecture, "all")
                            ]
                        ),
                        "source_artifact": self.source_artifact.id,
                    },
                    "output": (
                        {"source_analysis": False, "binary_all_analysis": False}
                        if architecture != "amd64"
                        else {}
                    ),
                    "target_distribution": "debian:trixie",
                },
            )

            self.assertEqual(
                lintian.workflow_data_json,
                {
                    "allow_failure": True,
                    "display_name": self._expected_task_display_name(
                        architecture, lintian
                    ),
                    "step": f"lintian-{architecture}",
                },
            )

            self.assertQuerySetEqual(
                lintian.dependencies.all(),
                set(
                    WorkRequest.objects.filter(
                        task_name="sbuild",
                        task_data__build_architecture__in=(architecture, "all"),
                    )
                ),
                ordered=False,
            )

            provide_actions[architecture] = ActionUpdateCollectionWithArtifacts(
                collection="internal@collections",
                name_template=f"reference-qa-result|lintian-{architecture}",
                artifact_filters={"category": ArtifactCategory.LINTIAN},
            )
            skip_actions[architecture] = ActionSkipIfLookupResultChanged(
                lookup=(
                    f"trixie@{CollectionCategory.QA_RESULTS}/"
                    f"latest:lintian_hello_{architecture}"
                ),
                collection_item_id=None,
                promise_name=f"reference-qa-result|lintian-{architecture}",
            )
            qa_result_actions[architecture] = (
                ActionUpdateCollectionWithArtifacts(
                    collection=f"trixie@{CollectionCategory.QA_RESULTS}",
                    variables={
                        "package": "hello",
                        "version": self.source_artifact.data["version"],
                        "$architecture": "architecture",
                        "timestamp": int(source_item.created_at.timestamp()),
                        "work_request_id": lintian.id,
                    },
                    artifact_filters={"category": ArtifactCategory.LINTIAN},
                )
            )
            self.assert_work_request_event_reactions(
                lintian,
                on_assignment=[skip_actions[architecture]],
                on_failure=[qa_result_actions[architecture]],
                on_success=[
                    provide_actions[architecture],
                    qa_result_actions[architecture],
                ],
            )

            # Completing the work request stores the QA result.
            artifact_architectures = (
                ["source", "all", architecture]
                if architecture == "amd64"
                else [architecture]
            )
            for artifact_architecture in artifact_architectures:
                results[artifact_architecture] = (
                    self.playground.create_artifact(
                        category=ArtifactCategory.LINTIAN,
                        data={"architecture": artifact_architecture},
                        work_request=lintian,
                    )[0]
                )
            self.playground.advance_work_request(
                lintian, result=WorkRequest.Results.SUCCESS, use_data=False
            )

        for architecture, result in results.items():
            self.assertEqual(
                lintian.lookup_single(
                    f"trixie@{CollectionCategory.QA_RESULTS}/"
                    f"latest:lintian_hello_{architecture}",
                    expect_type=LookupChildType.ARTIFACT,
                ).artifact,
                result,
            )

        # Population is idempotent, even if qa_suite changes.
        with (
            mock.patch.object(
                LintianWorkflow,
                "_has_current_reference_qa_result",
                return_value=False,
            ),
            mock.patch.object(
                LintianWorkflow,
                "qa_suite_changed",
                mock.PropertyMock(
                    return_value=timezone.now() + dt.timedelta(hours=1)
                ),
            ),
        ):
            self.get_workflow(workflow).populate()
        lintians = workflow.children.filter(task_name="lintian")
        self.assertEqual(lintians.count(), len(concrete_architectures))
        for architecture in concrete_architectures:
            lintian = lintians.get(
                workflow_data_json__step=f"lintian-{architecture}"
            )
            self.assert_work_request_event_reactions(
                lintian,
                on_assignment=[skip_actions[architecture]],
                on_failure=[qa_result_actions[architecture]],
                on_success=[
                    provide_actions[architecture],
                    qa_result_actions[architecture],
                ],
            )

    def test_populate_reference_qa_result_backs_off(self) -> None:
        """Reference tasks are skipped if another workflow got there first."""
        trixie_suite = self.playground.create_collection(
            "trixie", CollectionCategory.SUITE
        )
        trixie_qa_results = self.playground.create_collection(
            "trixie", CollectionCategory.QA_RESULTS
        )
        self.add_lintian_binary(trixie_suite, "2.116.3")
        self.playground.create_debian_environment(
            codename="trixie", variant="lintian"
        )

        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello",
                srcpkg_version="1.0-1",
                architecture="amd64",
            )
        )
        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                prefix="reference-qa-result|",
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple((binary_artifact.id,)),
                vendor="debian",
                codename="trixie",
                qa_suite=f"trixie@{CollectionCategory.SUITE}",
                reference_qa_results=f"trixie@{CollectionCategory.QA_RESULTS}",
                update_qa_results=RegressionTrackingUpdateQAResults.YES,
            ),
        )

        self.get_workflow(wr).populate()

        racing_qa_result = self.add_qa_result(
            trixie_qa_results, "hello", "1.0-1", "amd64", "2.116.3"
        )
        [child] = wr.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"'trixie@{CollectionCategory.QA_RESULTS}/"
                    f"latest:lintian_hello_amd64' changed"
                )
            ),
        )
        self.assertEqual(
            child.lookup_single(
                f"trixie@{CollectionCategory.QA_RESULTS}/"
                f"latest:lintian_hello_amd64",
                expect_type=LookupChildType.ARTIFACT,
            ).collection_item,
            racing_qa_result,
        )

    @preserve_db_task_registry()
    def test_orchestrate_idempotent(self) -> None:
        """Calling orchestrate twice does not create new work requests."""
        binary = self.playground.create_minimal_binary_package_artifact(
            "hello", "1.0.0", "hello", "1.0.0", "all"
        )

        wr = self.playground.create_workflow(
            task_name="lintian",
            task_data=LintianWorkflowData(
                source_artifact=self.source_artifact.id,
                binary_artifacts=LookupMultiple((binary.id,)),
                vendor="debian",
                codename="trixie",
            ),
        )

        self.get_workflow(wr).populate()

        children = set(wr.children.all())

        self.get_workflow(wr).populate()

        self.assertQuerySetEqual(wr.children.all(), children, ordered=False)

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

    def test_compute_dynamic_data(self) -> None:
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            LintianWorkflowDynamicData(
                subject="hello",
                parameter_summary="hello_1.0-1",
                source_artifact_id=self.source_artifact.id,
                binary_artifacts_ids=[self.binary_artifact.id],
            ),
        )

    def test_get_input_artifacts(self) -> None:
        workflow = self.task
        self.assertIsNone(workflow.dynamic_data)

        self.assertEqual(
            workflow.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=workflow.data.source_artifact,
                    label="source_artifact",
                    artifact_id=None,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=workflow.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=None,
                ),
            ],
        )

        workflow.dynamic_data = LintianWorkflowDynamicData(
            source_artifact_id=self.source_artifact.id,
            binary_artifacts_ids=[self.binary_artifact.id],
        )

        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=[self.binary_artifact.id],
                ),
            ],
        )

    @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", "i386", "ppc64el", "s390x"]
        reference_source_artifact = self.playground.create_source_artifact(
            name="hello", version="0.9-1"
        )

        root = self.playground.create_workflow(mark_running=True)
        reference = self.orchestrate(
            task_data=LintianWorkflowData(
                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 ("all", *architectures)
                        if arch != "ppc64el"
                    ]
                ),
                vendor="debian",
                codename="sid",
            ),
            architectures=["all", *architectures],
            parent=root,
            pipeline_task_name="referencepipeline",
        )
        new = self.orchestrate(
            task_data=LintianWorkflowData(
                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=self.source_artifact.id,
                binary_artifacts=LookupMultiple.model_validate(
                    [
                        f"internal@collections/name:build-{arch}"
                        for arch in ("all", *architectures)
                    ]
                ),
                vendor="debian",
                codename="sid",
            ),
            architectures=["all", *architectures],
            parent=root,
        )

        # Unblock the lintian 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
            )

        reference_lintian_workflow = reference.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="lintian"
        )
        reference_tasks = {
            wr.task_data["build_architecture"]: wr
            for wr in reference_lintian_workflow.children.filter(
                task_type=TaskTypes.WORKER, task_name="lintian"
            )
        }
        reference_results = {
            architecture: self.playground.create_artifact(
                category=ArtifactCategory.LINTIAN,
                data={
                    "architecture": architecture,
                    "summary": {
                        "tags_count_by_severity": tags_count_by_severity,
                        "tags_found": tags_found,
                    },
                },
                work_request=reference_tasks[
                    "amd64" if architecture == "all" else architecture
                ],
            )[0]
            for architecture, tags_count_by_severity, tags_found in (
                ("all", {"warning": 1, "error": 0}, ["warning-1"]),
                (
                    "amd64",
                    {"warning": 0, "error": 0},
                    [],
                ),
                (
                    "i386",
                    {"warning": 2, "error": 1},
                    ["warning-1", "warning-2", "error-1"],
                ),
                ("s390x", {"warning": 1, "error": 1}, ["warning-1", "error-1"]),
            )
        }
        for artifact in reference_results.values():
            self.playground.create_artifact_relation(
                artifact,
                reference_source_artifact,
                ArtifactRelation.Relations.RELATES_TO,
            )
        new_lintian_workflow = new.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="lintian"
        )
        new_tasks = {
            wr.task_data["build_architecture"]: wr
            for wr in new_lintian_workflow.children.filter(
                task_type=TaskTypes.WORKER, task_name="lintian"
            )
        }
        new_results = {
            architecture: self.playground.create_artifact(
                category=ArtifactCategory.LINTIAN,
                data={
                    "architecture": architecture,
                    "summary": {
                        "tags_count_by_severity": tags_count_by_severity,
                        "tags_found": tags_found,
                    },
                },
                work_request=new_tasks[
                    "amd64" if architecture == "all" else architecture
                ],
            )[0]
            for architecture, tags_count_by_severity, tags_found in (
                ("all", {"warning": 1, "error": 0}, ["warning-1"]),
                (
                    "amd64",
                    {"warning": 0, "error": 0},
                    [],
                ),
                (
                    "i386",
                    {"warning": 1, "error": 2},
                    ["warning-1", "error-1", "error-2"],
                ),
                ("ppc64el", {"warning": 1, "error": 0}, ["warning-1"]),
                ("s390x", {"warning": 1, "error": 0}, ["warning-1"]),
            )
        }
        for artifact in new_results.values():
            self.playground.create_artifact_relation(
                artifact,
                self.source_artifact,
                ArtifactRelation.Relations.RELATES_TO,
            )

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

        self.schedule_and_run_workflow_callback(new_lintian_workflow)

        expected_analysis: dict[
            str, tuple[RegressionAnalysisStatus, dict[str, list[str]]]
        ] = {
            "all": (
                RegressionAnalysisStatus.STABLE,
                {"added_tags": [], "removed_tags": []},
            ),
            "amd64": (
                RegressionAnalysisStatus.STABLE,
                {"added_tags": [], "removed_tags": []},
            ),
        }
        new_lintian_workflow.refresh_from_db()
        self.assert_regression_analysis(
            new_lintian_workflow,
            "0.9-1",
            reference_results,
            "1.0-1",
            new_results,
            expected_analysis,
        )

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

        for architecture, reference_result, new_result, status, details in (
            (
                "i386",
                WorkRequest.Results.FAILURE,
                WorkRequest.Results.FAILURE,
                RegressionAnalysisStatus.REGRESSION,
                {"added_tags": ["error-2"], "removed_tags": ["warning-2"]},
            ),
            (
                "ppc64el",
                None,
                WorkRequest.Results.SUCCESS,
                RegressionAnalysisStatus.NO_RESULT,
                {"added_tags": ["warning-1"], "removed_tags": []},
            ),
            (
                "s390x",
                WorkRequest.Results.FAILURE,
                WorkRequest.Results.SUCCESS,
                RegressionAnalysisStatus.IMPROVEMENT,
                {"added_tags": [], "removed_tags": ["error-1"]},
            ),
        ):
            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_lintian_workflow, WorkRequest.Statuses.PENDING
        )

        self.schedule_and_run_workflow_callback(new_lintian_workflow)

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