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

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

from django.utils import timezone

from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianAutopkgtest,
    DebianAutopkgtestResult,
    DebianAutopkgtestResultStatus,
    DebianAutopkgtestSource,
    DebianBinaryPackage,
    DebianLintian,
    DebianLintianSeverity,
    DebianLintianSummary,
    DebianSourcePackage,
    TaskTypes,
    get_source_package_name,
    get_source_package_version,
)
from debusine.db.models import Artifact, WorkRequest
from debusine.server.collections import DebianSuiteManager
from debusine.server.workflows import (
    QAWorkflow,
    WorkflowValidationError,
    workflow_utils,
)
from debusine.server.workflows.models import (
    BaseWorkflowData,
    QAWorkflowData,
    QAWorkflowDynamicData,
    RegressionTrackingUpdateQAResults,
    SbuildWorkflowData,
    WorkRequestWorkflowData,
)
from debusine.server.workflows.tests.helpers import (
    SampleWorkflow,
    WorkflowTestCase,
)
from debusine.server.workflows.tests.test_regression_tracking import (
    RegressionTrackingWorkflowTestMixin,
)
from debusine.tasks import TaskConfigError
from debusine.tasks.models import (
    BackendType,
    BaseDynamicTaskData,
    InputArtifactMultiple,
    InputArtifactSingle,
    LintianFailOnSeverity,
    LookupMultiple,
    RegressionAnalysisStatus,
    SbuildData,
    SbuildInput,
)
from debusine.test.django import preserve_db_task_registry


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

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

    source_artifact: ClassVar[Artifact]
    binary_artifact: ClassVar[Artifact]
    package_build_log: 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(
                architecture="amd64"
            )
        )
        cls.package_build_log = cls.playground.create_build_log_artifact()
        cls.playground.create_artifact_relation(
            artifact=cls.package_build_log, target=cls.binary_artifact
        )

        # Create artifacts needed to satisfy task data lookups
        cls.playground.create_debian_environment(
            codename="sid", architecture="amd64", variant="sbuild"
        )
        cls.playground.create_debian_environment(
            codename="bookworm", architecture="amd64", variant="autopkgtest"
        )
        cls.playground.create_debian_environment(
            codename="bookworm", architecture="amd64", variant="lintian"
        )
        cls.playground.create_debian_environment(
            codename="bookworm", architecture="amd64", variant="blhc"
        )

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

    def orchestrate(
        self,
        extra_data: dict[str, Any],
        *,
        architectures: Sequence[str] | None = None,
        sbuild_workflow_source_artifact: Artifact,
        internal_collection_artifacts: (
            list[tuple[Artifact, dict[str, Any]]] | None
        ) = None,
    ) -> WorkRequest:
        """Create a QAWorkflow and call orchestrate_workflow."""

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

            @override
            def populate(self_) -> None:
                source_artifact_id = (
                    sbuild_workflow_source_artifact.id
                    if sbuild_workflow_source_artifact is not None
                    else self.source_artifact.id
                )
                sbuild = self_.work_request.create_child_workflow(
                    task_name="sbuild",
                    task_data=SbuildWorkflowData(
                        input=SbuildInput(source_artifact=source_artifact_id),
                        target_distribution="debian:sid",
                        architectures=list(architectures or ["all"]),
                    ),
                )
                self.playground.advance_work_request(sbuild, mark_running=True)

                if architectures is not None:
                    for arch in architectures:
                        child = sbuild.create_child_worker(
                            task_name="sbuild",
                            task_data=SbuildData(
                                input=SbuildInput(
                                    source_artifact=source_artifact_id
                                ),
                                build_architecture=arch,
                                environment="debian/match:codename=sid",
                            ),
                            workflow_data=WorkRequestWorkflowData(
                                display_name=f"Build {arch}",
                                step=f"build-{arch}",
                            ),
                        )

                        self_.provides_artifact(
                            child,
                            ArtifactCategory.UPLOAD,
                            f"build-{arch}",
                            data={
                                "binary_names": ["hello"],
                                "architecture": arch,
                                "source_package_name": "hello",
                            },
                        )
                        self_.provides_artifact(
                            child,
                            ArtifactCategory.PACKAGE_BUILD_LOG,
                            f"buildlog-{arch}",
                            data={
                                "architecture": arch,
                                "source_package_name": "hello",
                                "source_package_version": "1.0-1",
                            },
                        )

                sbuild.unblock_workflow_children()

                data = {
                    "source_artifact": source_artifact_id,
                    "binary_artifacts": [self.binary_artifact.id],
                    "enable_blhc": False,
                    "vendor": "debian",
                    "codename": "bookworm",
                }

                qa, _ = self_.work_request_ensure_child_workflow(
                    task_name="qa",
                    task_data=QAWorkflowData(**{**data, **extra_data}),
                    workflow_data=WorkRequestWorkflowData(
                        display_name="DebDiff workflow", step="debdiff-workflow"
                    ),
                )
                self_.orchestrate_child(qa)

        root = self.playground.create_workflow(
            task_name="examplepipeline",
            internal_collection_artifacts=internal_collection_artifacts,
        )
        self.assertTrue(
            self.playground.orchestrate_workflow(root, use_data=False)
        )

        return root.children.get(task_name="qa")

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

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

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

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

    def test_compute_dynamic_data_reference_source_artifact(self) -> None:
        reference_source_artifact = self.source_artifact
        new_source_artifact = self.playground.create_source_artifact(
            name="hello", version="1.0-2"
        )

        self.task_data["source_artifact"] = new_source_artifact.id
        self.task_data["reference_source_artifact"] = (
            reference_source_artifact.id
        )
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            QAWorkflowDynamicData(
                subject="hello",
                parameter_summary="hello_1.0-2",
                source_artifact_id=new_source_artifact.id,
                reference_source_artifact_id=reference_source_artifact.id,
                binary_artifacts_ids=[self.binary_artifact.id],
            ),
        )

    def test_compute_dynamic_data_source_artifact_wrong_category(self) -> None:
        """`source_artifact` must be a source-package or upload."""
        artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.TEST
        )
        self.task_data["source_artifact"] = artifact.id
        with self.assertRaisesRegex(
            TaskConfigError,
            r"^source_artifact: unexpected artifact category: 'debusine:test'. "
            r"Valid categories: \['debian:source-package', 'debian:upload'\]$",
        ):
            self.task

    def test_get_input_artifacts_mandatory_data(self) -> None:
        upload_artifacts = self.playground.create_upload_artifacts()
        self.task_data["source_artifact"] = upload_artifacts.upload.id
        self.task_data["enable_blhc"] = False
        self.assertIsNone(self.task.dynamic_data)
        self.assertEqual(
            self.task.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=self.task.data.source_artifact,
                    label="source_artifact",
                    artifact_id=None,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=self.task.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=None,
                ),
            ],
        )

        self.task.dynamic_data = self.compute_dynamic_data(self.task)
        self.assertIsNotNone(self.task.dynamic_data)

        self.assertEqual(
            self.task.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=self.task.data.source_artifact,
                    label="source_artifact",
                    artifact_id=upload_artifacts.source.id,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=self.task.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=[self.binary_artifact.id],
                ),
            ],
        )

    def test_get_input_artifacts_with_optional_data(self) -> None:
        upload_artifacts = self.playground.create_upload_artifacts()
        reference_source_artifact = self.source_artifact

        self.task_data["source_artifact"] = upload_artifacts.upload.id
        self.task_data["reference_source_artifact"] = (
            reference_source_artifact.id
        )
        self.task_data["enable_blhc"] = False
        self.task.dynamic_data = QAWorkflowDynamicData(
            source_artifact_id=upload_artifacts.source.id,
            binary_artifacts_ids=[self.binary_artifact.id],
            reference_source_artifact_id=reference_source_artifact.id,
        )

        assert self.task.data.reference_source_artifact is not None
        self.assertEqual(
            self.task.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=self.task.data.source_artifact,
                    label="source_artifact",
                    artifact_id=upload_artifacts.source.id,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=self.task.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=self.task.dynamic_data.binary_artifacts_ids,
                ),
                InputArtifactSingle(
                    lookup=self.task.data.reference_source_artifact,
                    label="reference_source_artifact",
                    artifact_id=reference_source_artifact.id,
                ),
            ],
        )

    def test_source_package_upload(self) -> None:
        """`source_artifact` can be an upload."""
        upload_artifacts = self.playground.create_upload_artifacts()
        self.task_data["source_artifact"] = upload_artifacts.upload.id
        self.task_data["enable_blhc"] = False
        self.assertEqual(
            workflow_utils.source_package(self.task), upload_artifacts.source
        )

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

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "architectures": ["amd64"],
                "extra_repositories": [
                    {
                        "url": "http://example.com/",
                        "suite": "bookworm",
                        "components": ["main"],
                    }
                ],
            },
            architectures=architectures,
            sbuild_workflow_source_artifact=source_artifact,
        )

        autopkgtest = workflow.children.get(
            task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
        )

        assert workflow.parent is not None
        self.assertEqual(autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            autopkgtest.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
                "extra_repositories": [
                    {
                        "url": "http://example.com/",
                        "suite": "bookworm",
                        "components": ["main"],
                    }
                ],
            },
        )
        self.assertEqual(
            autopkgtest.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "autopkgtest",
                "step": "autopkgtest",
            },
        )
        self.assertQuerySetEqual(autopkgtest.dependencies.all(), [])

        # AutopkgtestWorkflow.populate() was called and created its tasks
        self.assertQuerySetEqual(
            autopkgtest.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "autopkgtest",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                )
            ],
        )

        self.assertFalse(
            workflow.children.filter(
                task_name="reverse_dependencies_autopkgtest",
                task_type=TaskTypes.WORKFLOW,
            ).exists()
        )

        lintian = workflow.children.get(
            task_name="lintian", task_type=TaskTypes.WORKFLOW
        )

        self.assertEqual(lintian.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            lintian.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "fail_on_severity": LintianFailOnSeverity.ERROR,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
            },
        )
        self.assertEqual(
            lintian.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "lintian",
                "step": "lintian",
            },
        )
        self.assertQuerySetEqual(lintian.dependencies.all(), [])
        # LintianWorkflow.populate() was called and created its tasks
        self.assertQuerySetEqual(
            lintian.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "lintian",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                )
            ],
        )

        piuparts = workflow.children.get(
            task_name="piuparts", task_type=TaskTypes.WORKFLOW
        )

        self.assertEqual(piuparts.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            piuparts.task_data,
            {
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
                "extra_repositories": [
                    {
                        "url": "http://example.com/",
                        "suite": "bookworm",
                        "components": ["main"],
                    }
                ],
            },
        )

        self.assertEqual(
            piuparts.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "piuparts",
                "step": "piuparts",
            },
        )
        self.assertQuerySetEqual(piuparts.dependencies.all(), [])

        # PiupartsWorkflow.populate() was called and created its tasks
        self.assertQuerySetEqual(
            piuparts.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "piuparts",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                )
            ],
        )

    @preserve_db_task_registry()
    def test_populate_use_available_architectures(self) -> None:
        """
        Test populate uses available architectures.

        The user didn't specify "architectures", so QAWorkflow checks
        available architectures and "all" and uses them.
        """
        for arch in ["amd64", "i386"]:
            self.playground.create_debian_environment(architecture=arch)
        source_artifact = self.source_artifact
        binary_artifact = self.binary_artifact

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": [binary_artifact.id],
            },
            sbuild_workflow_source_artifact=source_artifact,
        )

        autopkgtest = workflow.children.get(
            task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(
            autopkgtest.task_data["architectures"], ["all", "amd64", "i386"]
        )

    @preserve_db_task_registry()
    def test_populate_disable_autopkgtest_lintian_piuparts(self) -> None:
        """Populate does not create autopkgtest, lintian nor piuparts."""
        source_artifact = self.source_artifact
        binary_artifact = self.binary_artifact
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": [binary_artifact.id],
                "architectures": ["amd64"],
                "enable_autopkgtest": False,
                "enable_lintian": False,
                "enable_piuparts": False,
            },
            sbuild_workflow_source_artifact=source_artifact,
        )

        self.assertFalse(
            workflow.children.filter(
                task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
            ).exists()
        )
        self.assertFalse(
            workflow.children.filter(
                task_name="reverse_dependencies_autopkgtest",
                task_type=TaskTypes.WORKFLOW,
            ).exists()
        )
        self.assertFalse(
            workflow.children.filter(
                task_name="lintian", task_type=TaskTypes.WORKFLOW
            ).exists()
        )
        self.assertFalse(
            workflow.children.filter(
                task_name="piuparts", task_type=TaskTypes.WORKFLOW
            ).exists()
        )

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

        wr = self.playground.create_workflow(
            task_name="qa",
            task_data=QAWorkflowData(
                source_artifact=f"{artifacts.upload.id}@artifacts",
                binary_artifacts=LookupMultiple((artifacts.upload.id,)),
                architectures=architectures,
                vendor="debian",
                codename="sid",
                qa_suite=f"sid@{CollectionCategory.SUITE}",
                enable_autopkgtest=True,
                enable_reverse_dependencies_autopkgtest=True,
                enable_debdiff=True,
                enable_lintian=True,
                enable_piuparts=True,
                enable_blhc=True,
            ),
        )
        self.get_workflow(wr).populate()
        children = list(wr.children.order_by("id").all())
        expected_src_artifact = f"{artifacts.upload.id}@artifacts"
        expected_bin_artifacts = [artifacts.upload.id]
        expected: dict[str, dict[str, Any]] = {
            "autopkgtest": {
                "source_artifact": expected_src_artifact,
                "binary_artifacts": expected_bin_artifacts,
            },
            "blhc": {
                "source_artifact": expected_src_artifact,
                "binary_artifacts": expected_bin_artifacts,
            },
            "debdiff": {
                "source_artifact": expected_src_artifact,
                "binary_artifacts": expected_bin_artifacts,
            },
            "lintian": {
                "source_artifact": expected_src_artifact,
                "binary_artifacts": expected_bin_artifacts,
            },
            "piuparts": {
                "source_artifact": expected_src_artifact,
                "binary_artifacts": expected_bin_artifacts,
            },
            "reverse_dependencies_autopkgtest": {
                "source_artifact": expected_src_artifact,
                "binary_artifacts": expected_bin_artifacts,
            },
        }
        self.assertEqual(len(children), len(expected))
        for child in children:
            with self.subTest(task_name=child.task_name):
                expected_data = expected[child.task_name]
                for k, v in expected_data.items():
                    self.assertEqual(
                        child.task_data.get(k), v, f"task_data[{k}]"
                    )

    @preserve_db_task_registry()
    def test_populate_enable_reverse_dependencies_autopkgtest(self) -> None:
        """Populate can enable reverse_dependencies_autopkgtest."""
        architectures = ["amd64"]
        source_artifact = self.playground.create_source_artifact(name="hello")
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        dep_source_artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.SOURCE_PACKAGE,
            data=DebianSourcePackage(
                name="depends",
                version="1",
                type="dpkg",
                dsc_fields={
                    "Package": "depends",
                    "Version": "1",
                    "Testsuite": "autopkgtest",
                },
            ),
        )
        sid.manager.add_artifact(
            dep_source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )
        dep_binary_artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.BINARY_PACKAGE,
            data=DebianBinaryPackage(
                srcpkg_name="depends",
                srcpkg_version="1",
                deb_fields={
                    "Package": "depends",
                    "Version": "1",
                    "Architecture": "all",
                    "Depends": "hello",
                },
                deb_control_files=[],
            ),
        )
        sid.manager.add_artifact(
            dep_binary_artifact,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "architectures": ["amd64"],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "enable_reverse_dependencies_autopkgtest": True,
            },
            architectures=architectures,
            sbuild_workflow_source_artifact=source_artifact,
        )

        rdep_autopkgtest = workflow.children.get(
            task_name="reverse_dependencies_autopkgtest",
            task_type=TaskTypes.WORKFLOW,
        )

        assert workflow.parent is not None
        self.assertEqual(rdep_autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            rdep_autopkgtest.task_data,
            {
                "prefix": "",
                "reference_prefix": "",
                "source_artifact": source_artifact.id,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": None,
                "enable_regression_tracking": False,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
                "codename": "bookworm",
                "backend": BackendType.AUTO,
                "architectures": ["amd64"],
                "arch_all_build_architecture": "amd64",
                "extra_repositories": None,
            },
        )
        self.assertEqual(
            rdep_autopkgtest.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "autopkgtests of reverse-dependencies",
                "step": "reverse-dependencies-autopkgtest",
            },
        )
        self.assertQuerySetEqual(rdep_autopkgtest.dependencies.all(), [])

        # ReverseDependenciesAutopkgtestWorkflow.populate() was called and
        # created its sub-workflow.
        autopkgtest = rdep_autopkgtest.children.get()
        self.assertEqual(autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            autopkgtest.task_data["source_artifact"],
            "sid@debian:suite/source-version:depends_1",
        )
        self.assertQuerySetEqual(
            autopkgtest.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "autopkgtest",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                )
            ],
        )

    @preserve_db_task_registry()
    def test_populate_update_qa_results(self) -> None:
        """Populate can update QA results."""
        architectures = ["amd64"]
        source_artifact = self.source_artifact
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        self.playground.create_collection(
            name="sid", category=CollectionCategory.QA_RESULTS
        )
        dep_source_artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.SOURCE_PACKAGE,
            data=DebianSourcePackage(
                name="depends",
                version="1",
                type="dpkg",
                dsc_fields={
                    "Package": "depends",
                    "Version": "1",
                    "Testsuite": "autopkgtest",
                },
            ),
        )
        sid.manager.add_artifact(
            dep_source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )
        dep_binary_artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.BINARY_PACKAGE,
            data=DebianBinaryPackage(
                srcpkg_name="depends",
                srcpkg_version="1",
                deb_fields={
                    "Package": "depends",
                    "Version": "1",
                    "Architecture": "all",
                    "Depends": "hello",
                },
                deb_control_files=[],
            ),
        )
        sid.manager.add_artifact(
            dep_binary_artifact,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

        workflow = self.orchestrate(
            extra_data={
                "prefix": "reference-qa-result|",
                "reference_prefix": "",
                "source_artifact": source_artifact.id,
                # This would normally be the corresponding binary artifacts
                # in qa_suite, but that doesn't really matter for testing
                # purposes.
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "architectures": ["amd64"],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "enable_regression_tracking": False,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "enable_reverse_dependencies_autopkgtest": True,
                "enable_blhc": True,
            },
            architectures=architectures,
            sbuild_workflow_source_artifact=source_artifact,
        )

        autopkgtest = workflow.children.get(
            task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
        )

        assert workflow.parent is not None
        self.assertEqual(autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            autopkgtest.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "prefix": "reference-qa-result|",
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_prefix": "",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "vendor": "debian",
                "extra_repositories": None,
            },
        )
        self.assertEqual(
            autopkgtest.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "autopkgtest",
                "step": "autopkgtest",
            },
        )
        self.assertQuerySetEqual(autopkgtest.dependencies.all(), [])

        # AutopkgtestWorkflow.populate() was called and created its tasks.
        self.assertQuerySetEqual(
            autopkgtest.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "autopkgtest",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                )
            ],
        )

        rdep_autopkgtest = workflow.children.get(
            task_name="reverse_dependencies_autopkgtest",
            task_type=TaskTypes.WORKFLOW,
        )

        assert workflow.parent is not None
        self.assertEqual(rdep_autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            rdep_autopkgtest.task_data,
            {
                "prefix": "reference-qa-result|",
                "reference_prefix": "",
                "source_artifact": source_artifact.id,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "enable_regression_tracking": False,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "vendor": "debian",
                "codename": "bookworm",
                "backend": BackendType.AUTO,
                "architectures": ["amd64"],
                "arch_all_build_architecture": "amd64",
                "extra_repositories": None,
            },
        )
        self.assertEqual(
            rdep_autopkgtest.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "autopkgtests of reverse-dependencies",
                "step": "reverse-dependencies-autopkgtest",
            },
        )
        self.assertQuerySetEqual(rdep_autopkgtest.dependencies.all(), [])

        # ReverseDependenciesAutopkgtestWorkflow.populate() was called and
        # created its sub-workflow.
        sub_autopkgtest = rdep_autopkgtest.children.get()
        self.assertEqual(sub_autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            sub_autopkgtest.task_data["source_artifact"],
            "sid@debian:suite/source-version:depends_1",
        )
        self.assertQuerySetEqual(
            sub_autopkgtest.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "autopkgtest",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                )
            ],
        )

        lintian = workflow.children.get(
            task_name="lintian", task_type=TaskTypes.WORKFLOW
        )

        self.assertEqual(lintian.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            lintian.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "fail_on_severity": LintianFailOnSeverity.ERROR,
                "prefix": "reference-qa-result|",
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_prefix": "",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "vendor": "debian",
            },
        )
        self.assertEqual(
            lintian.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "lintian",
                "step": "lintian",
            },
        )
        self.assertQuerySetEqual(lintian.dependencies.all(), [])

        # LintianWorkflow.populate() was called and created its tasks.
        self.assertQuerySetEqual(
            lintian.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "lintian",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                )
            ],
        )

        piuparts = workflow.children.get(
            task_name="piuparts", task_type=TaskTypes.WORKFLOW
        )

        self.assertEqual(piuparts.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            piuparts.task_data,
            {
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "prefix": "reference-qa-result|",
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_prefix": "",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "vendor": "debian",
                "extra_repositories": None,
            },
        )

        self.assertEqual(
            piuparts.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "piuparts",
                "step": "piuparts",
            },
        )
        self.assertQuerySetEqual(piuparts.dependencies.all(), [])

        blhc = workflow.children.get(
            task_name="blhc", task_type=TaskTypes.WORKFLOW
        )

        assert workflow.parent is not None
        self.assertEqual(blhc.status, WorkRequest.Statuses.BLOCKED)
        self.assertEqual(
            blhc.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "prefix": "reference-qa-result|",
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_prefix": "",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
                "vendor": "debian",
            },
        )
        self.assertEqual(
            blhc.workflow_data_json,
            {
                "allow_failure": False,
                "display_name": "build log hardening check",
                "step": "blhc",
            },
        )
        self.assertQuerySetEqual(
            blhc.dependencies.values_list("task_type", "task_name"),
            [(TaskTypes.WORKER, "sbuild")],
        )
        # No children yet; see test_populate_blhc.
        self.assertFalse(blhc.children.exists())

    @preserve_db_task_registry()
    def test_populate_enable_regression_tracking(self) -> None:
        """Populate can run in a mode that tracks regressions."""
        architectures = ["amd64"]
        source_artifact = self.source_artifact
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        self.playground.create_collection(
            name="sid", category=CollectionCategory.QA_RESULTS
        )
        dep_source_artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.SOURCE_PACKAGE,
            data=DebianSourcePackage(
                name="depends",
                version="1",
                type="dpkg",
                dsc_fields={
                    "Package": "depends",
                    "Version": "1",
                    "Testsuite": "autopkgtest",
                },
            ),
        )
        sid.manager.add_artifact(
            dep_source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )
        dep_binary_artifact, _ = self.playground.create_artifact(
            category=ArtifactCategory.BINARY_PACKAGE,
            data=DebianBinaryPackage(
                srcpkg_name="depends",
                srcpkg_version="1",
                deb_fields={
                    "Package": "depends",
                    "Version": "1",
                    "Architecture": "all",
                    "Depends": "hello",
                },
                deb_control_files=[],
            ),
        )
        sid.manager.add_artifact(
            dep_binary_artifact,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

        workflow = self.orchestrate(
            extra_data={
                "reference_prefix": "reference-qa-result|",
                "source_artifact": source_artifact.id,
                # This would normally be the corresponding binary artifacts
                # in qa_suite, but that doesn't really matter for testing
                # purposes.
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "architectures": ["amd64"],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "enable_regression_tracking": True,
                "enable_reverse_dependencies_autopkgtest": True,
                "enable_blhc": True,
            },
            architectures=architectures,
            sbuild_workflow_source_artifact=source_artifact,
        )

        autopkgtest = workflow.children.get(
            task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertDictContainsAll(
            autopkgtest.task_data,
            {
                "enable_regression_tracking": True,
                "reference_prefix": "reference-qa-result|",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
            },
        )
        self.assertEqual(
            autopkgtest.workflow_data_json,
            {
                "allow_failure": True,
                "display_name": "autopkgtest",
                "step": "autopkgtest",
            },
        )
        self.assertQuerySetEqual(autopkgtest.dependencies.all(), [])
        self.assertTrue(autopkgtest.children.exists())

        rdep_autopkgtest = workflow.children.get(
            task_name="reverse_dependencies_autopkgtest",
            task_type=TaskTypes.WORKFLOW,
        )
        self.assertEqual(rdep_autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertDictContainsAll(
            rdep_autopkgtest.task_data,
            {
                "enable_regression_tracking": True,
                "reference_prefix": "reference-qa-result|",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
            },
        )
        self.assertEqual(
            rdep_autopkgtest.workflow_data_json,
            {
                "allow_failure": True,
                "display_name": "autopkgtests of reverse-dependencies",
                "step": "reverse-dependencies-autopkgtest",
            },
        )
        self.assertQuerySetEqual(rdep_autopkgtest.dependencies.all(), [])

        # ReverseDependenciesAutopkgtestWorkflow.populate() was called and
        # created its sub-workflow.
        sub_autopkgtest = rdep_autopkgtest.children.get()
        self.assertEqual(sub_autopkgtest.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            sub_autopkgtest.task_data["source_artifact"],
            "sid@debian:suite/source-version:depends_1",
        )
        self.assertTrue(sub_autopkgtest.children.exists())

        lintian = workflow.children.get(
            task_name="lintian", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(lintian.status, WorkRequest.Statuses.RUNNING)
        self.assertDictContainsAll(
            lintian.task_data,
            {
                "enable_regression_tracking": True,
                "reference_prefix": "reference-qa-result|",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
            },
        )
        self.assertEqual(
            lintian.workflow_data_json,
            {
                "allow_failure": True,
                "display_name": "lintian",
                "step": "lintian",
            },
        )
        self.assertQuerySetEqual(lintian.dependencies.all(), [])
        self.assertTrue(lintian.children.exists())

        piuparts = workflow.children.get(
            task_name="piuparts", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(piuparts.status, WorkRequest.Statuses.RUNNING)
        self.assertDictContainsAll(
            piuparts.task_data,
            {
                "enable_regression_tracking": True,
                "reference_prefix": "reference-qa-result|",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
            },
        )
        self.assertEqual(
            piuparts.workflow_data_json,
            {
                "allow_failure": True,
                "display_name": "piuparts",
                "step": "piuparts",
            },
        )
        self.assertQuerySetEqual(piuparts.dependencies.all(), [])
        self.assertTrue(piuparts.children.exists())

        blhc = workflow.children.get(
            task_name="blhc", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(blhc.status, WorkRequest.Statuses.BLOCKED)
        self.assertDictContainsAll(
            blhc.task_data,
            {
                "enable_regression_tracking": True,
                "reference_prefix": "reference-qa-result|",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
            },
        )
        self.assertEqual(
            blhc.workflow_data_json,
            {
                "allow_failure": True,
                "display_name": "build log hardening check",
                "step": "blhc",
            },
        )
        self.assertQuerySetEqual(
            blhc.dependencies.values_list("task_type", "task_name"),
            [(TaskTypes.WORKER, "sbuild")],
        )
        # No children yet.
        self.assertFalse(blhc.children.exists())

        # Complete sbuild to unblock the blhc workflow.
        assert workflow.parent is not None
        self.simulate_sbuild_workflow_completion(
            workflow.parent.children.get(
                task_type=TaskTypes.WORKFLOW, task_name="sbuild"
            )
        )
        blhc.refresh_from_db()
        self.assertEqual(blhc.status, WorkRequest.Statuses.PENDING)
        self.schedule_and_run_workflow(workflow.parent)
        blhc.refresh_from_db()
        self.assertEqual(blhc.status, WorkRequest.Statuses.RUNNING)
        self.assertTrue(blhc.children.exists())

        ra = workflow.children.workflow_callbacks().get(
            workflow_data_json__step="regression-analysis"
        )
        self.assertEqual(ra.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(
            ra.workflow_data_json,
            {
                "allow_failure": False,
                "allow_dependency_failures": True,
                "display_name": "Regression analysis",
                "step": "regression-analysis",
                "visible": True,
            },
        )
        ra_callbacks = [
            wr.children.get(
                workflow_data_json__step__startswith="regression-analysis"
            )
            for wr in (autopkgtest, sub_autopkgtest, lintian, piuparts, blhc)
        ]
        self.assertQuerySetEqual(
            ra.dependencies.all(), ra_callbacks, ordered=False
        )

    @preserve_db_task_registry()
    def test_populate_architectures_allowed(self) -> None:
        """Populate uses architectures and architectures_allowed."""
        source_artifact = self.source_artifact
        binary_artifact_amd = (
            self.playground.create_minimal_binary_package_artifact(
                architecture="amd64"
            )
        )
        binary_artifact_i386 = (
            self.playground.create_minimal_binary_package_artifact(
                architecture="i386"
            )
        )
        binary_artifact_arm64 = (
            self.playground.create_minimal_binary_package_artifact(
                architecture="arm64"
            )
        )

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": [
                    binary_artifact_amd.id,
                    binary_artifact_i386.id,
                    binary_artifact_arm64.id,
                ],
                "architectures": ["amd64", "i386"],
                "architectures_allowlist": ["i386"],
            },
            sbuild_workflow_source_artifact=source_artifact,
        )

        autopkgtest = workflow.children.get(
            task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
        )

        self.assertEqual(
            autopkgtest.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["i386"],
                "backend": BackendType.AUTO,
                "binary_artifacts": [binary_artifact_i386.id],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
                "extra_repositories": None,
            },
        )

    @preserve_db_task_registry()
    def test_architectures_deny(self) -> None:
        """Populate uses architectures and architectures_denylist."""
        source_artifact = self.source_artifact
        binary_artifact_amd = (
            self.playground.create_minimal_binary_package_artifact(
                architecture="amd64"
            )
        )
        binary_artifact_i386 = (
            self.playground.create_minimal_binary_package_artifact(
                architecture="i386"
            )
        )
        binary_artifact_arm64 = (
            self.playground.create_minimal_binary_package_artifact(
                architecture="arm64"
            )
        )

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": [
                    binary_artifact_amd.id,
                    binary_artifact_i386.id,
                    binary_artifact_arm64.id,
                ],
                "architectures": ["amd64", "i386"],
                "architectures_denylist": ["i386"],
            },
            architectures=["amd64", "i386"],
            sbuild_workflow_source_artifact=source_artifact,
        )

        autopkgtest = workflow.children.get(
            task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
        )

        self.assertEqual(
            autopkgtest.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "binary_artifacts": [binary_artifact_amd.id],
                "codename": "bookworm",
                "enable_regression_tracking": False,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "source_artifact": source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
                "extra_repositories": None,
            },
        )

    @preserve_db_task_registry()
    def test_populate_subworkflow_configuration(self) -> None:
        """Populate configures sub-workflows according to specified data."""
        self.playground.create_debian_environment()
        source_artifact = self.source_artifact
        binary_artifact = self.binary_artifact

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": [binary_artifact.id],
                "autopkgtest_backend": BackendType.INCUS_VM,
                "lintian_backend": BackendType.UNSHARE,
                "lintian_fail_on_severity": LintianFailOnSeverity.PEDANTIC,
                "piuparts_backend": BackendType.INCUS_LXC,
                "piuparts_environment": "debian/match:codename=trixie",
            },
            sbuild_workflow_source_artifact=source_artifact,
        )

        autopkgtest = workflow.children.get(
            task_name="autopkgtest", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(autopkgtest.task_data["backend"], BackendType.INCUS_VM)

        lintian = workflow.children.get(
            task_name="lintian", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(lintian.task_data["backend"], BackendType.UNSHARE)
        self.assertEqual(
            lintian.task_data["fail_on_severity"],
            LintianFailOnSeverity.PEDANTIC,
        )

        piuparts = workflow.children.get(
            task_name="piuparts", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(piuparts.task_data["backend"], BackendType.INCUS_LXC)
        self.assertEqual(
            piuparts.task_data["environment"], "debian/match:codename=trixie"
        )

    @preserve_db_task_registry()
    def test_populate_debdiff(self) -> None:
        """Populate creates a debdiff workflow."""
        self.playground.create_collection(
            name="bookworm",
            category=CollectionCategory.SUITE,
        )
        trixie = self.playground.create_collection(
            name="trixie",
            category=CollectionCategory.SUITE,
        )
        origin_source_artifact = self.playground.create_source_artifact(
            name="hello", version="1.0.0-1"
        )
        trixie.manager.add_artifact(
            origin_source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )
        origin_binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello",
                srcpkg_version="1.0.0-1",
                version="1.0.0-1",
                architecture="amd64",
            )
        )
        trixie.manager.add_artifact(
            origin_binary_artifact,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )

        source_artifact = self.playground.create_source_artifact(name="hello")
        workflow = self.orchestrate(
            architectures=["amd64"],
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": LookupMultiple(
                    ("internal@collections/name:build-amd64",)
                ),
                "architectures": ["amd64"],
                "enable_autopkgtest": False,
                "enable_lintian": False,
                "enable_piuparts": False,
                "enable_debdiff": True,
                "vendor": "debian",
                "codename": "bookworm",
                "qa_suite": "trixie@debian:suite",
            },
            sbuild_workflow_source_artifact=source_artifact,
        )
        self.playground.advance_work_request(workflow, mark_running=True)

        assert workflow.parent is not None
        sbuild_workflow = workflow.parent.children.get(
            task_name="sbuild", task_type=TaskTypes.WORKFLOW
        )
        self.playground.advance_work_request(sbuild_workflow, mark_running=True)

        sbuild_worker_amd64 = sbuild_workflow.children.get(
            task_name="sbuild", task_type=TaskTypes.WORKER
        )

        debdiff_workflow = workflow.children.get(
            task_name="debdiff", task_type=TaskTypes.WORKFLOW
        )

        self.assertEqual(
            debdiff_workflow.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "codename": "bookworm",
                "original": "trixie@debian:suite",
                "source_artifact": source_artifact.id,
                "vendor": "debian",
            },
        )

        # No children and blocked: it needs the binary artifacts to be able
        # to create the debdiff tasks
        self.assertEqual(debdiff_workflow.children.count(), 0)
        self.assertEqual(debdiff_workflow.status, WorkRequest.Statuses.BLOCKED)
        self.assertQuerySetEqual(
            debdiff_workflow.dependencies.all(), [sbuild_worker_amd64]
        )

        # Create build artifacts and complete sbuild to unblock the debdiff
        # workflow.
        binaries, _ = self.simulate_sbuild_workflow_completion(sbuild_workflow)

        debdiff_workflow.refresh_from_db()
        self.assertEqual(debdiff_workflow.status, WorkRequest.Statuses.PENDING)
        self.schedule_and_run_workflow(workflow.parent)

        debdiff_workflow.refresh_from_db()
        self.assertEqual(debdiff_workflow.status, WorkRequest.Statuses.RUNNING)
        [debdiff_source_work_request, debdiff_binary_work_request] = (
            debdiff_workflow.children.order_by("id")
        )

        self.assertEqual(debdiff_source_work_request.task_name, "debdiff")
        self.assertEqual(
            debdiff_source_work_request.workflow_data_json,
            {
                "display_name": "DebDiff for source package",
                "step": "debdiff-source",
            },
        )
        self.assertEqual(
            debdiff_source_work_request.task_data,
            {
                "build_architecture": "amd64",
                "environment": "debian/match:codename=bookworm",
                "extra_flags": [],
                "input": {
                    "source_artifacts": [
                        "trixie@debian:suite/name:hello_1.0.0-1",
                        source_artifact.id,
                    ]
                },
            },
        )

        self.assertEqual(
            debdiff_binary_work_request.task_data,
            {
                "build_architecture": "amd64",
                "environment": "debian/match:codename=bookworm",
                "extra_flags": [],
                "input": {
                    "binary_artifacts": [
                        ["trixie@debian:suite/name:hello_1.0.0-1_amd64"],
                        [binaries["amd64"][0].id],
                    ]
                },
            },
        )

    @preserve_db_task_registry()
    def test_populate_blhc(self) -> None:
        """Populate enables blhc."""
        architectures = ["amd64"]
        source_artifact = self.source_artifact
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        assert isinstance(sid.manager, DebianSuiteManager)

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": source_artifact.id,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "architectures": ["amd64"],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "enable_blhc": True,
            },
            architectures=architectures,
            sbuild_workflow_source_artifact=source_artifact,
        )

        assert workflow.parent is not None
        sbuild_workflow = workflow.parent.children.get(
            task_name="sbuild", task_type=TaskTypes.WORKFLOW
        )
        self.playground.advance_work_request(sbuild_workflow, mark_running=True)

        blhc_workflow = workflow.children.get(
            task_name="blhc", task_type=TaskTypes.WORKFLOW
        )

        # No children and blocked: it needs the binary artifacts first.
        self.assertFalse(blhc_workflow.children.exists())
        self.assertEqual(blhc_workflow.status, WorkRequest.Statuses.BLOCKED)
        self.assertQuerySetEqual(
            blhc_workflow.dependencies.values_list("task_type", "task_name"),
            [(TaskTypes.WORKER, "sbuild")],
        )

        # Create build artifacts and complete sbuild to unblock the blhc
        # workflow.
        _, build_logs = self.simulate_sbuild_workflow_completion(
            sbuild_workflow
        )

        blhc_workflow.refresh_from_db()
        self.assertEqual(blhc_workflow.status, WorkRequest.Statuses.PENDING)
        self.schedule_and_run_workflow(workflow.parent)

        blhc_workflow.refresh_from_db()
        self.assertEqual(blhc_workflow.children.count(), 1)
        self.assertEqual(blhc_workflow.status, WorkRequest.Statuses.RUNNING)

        [blhc_work_request] = blhc_workflow.children.all()

        self.assertEqual(blhc_work_request.task_name, "blhc")
        self.assertEqual(
            blhc_work_request.workflow_data_json,
            {
                "display_name": "build log hardening check for amd64",
                "step": "blhc-amd64",
            },
        )
        self.assertEqual(
            blhc_work_request.task_data,
            {
                "build_architecture": "amd64",
                "environment": "debian/match:codename=bookworm",
                "extra_flags": [],
                "input": {"artifact": f"{build_logs['amd64'].id}@artifacts"},
            },
        )

    def test_summarize_status(self) -> None:
        for statuses, expected_status in (
            (
                {
                    RegressionAnalysisStatus.REGRESSION,
                    RegressionAnalysisStatus.ERROR,
                    RegressionAnalysisStatus.IMPROVEMENT,
                    RegressionAnalysisStatus.STABLE,
                    RegressionAnalysisStatus.NO_RESULT,
                },
                RegressionAnalysisStatus.REGRESSION,
            ),
            (
                {
                    RegressionAnalysisStatus.ERROR,
                    RegressionAnalysisStatus.IMPROVEMENT,
                    RegressionAnalysisStatus.STABLE,
                    RegressionAnalysisStatus.NO_RESULT,
                },
                RegressionAnalysisStatus.ERROR,
            ),
            (
                {
                    RegressionAnalysisStatus.IMPROVEMENT,
                    RegressionAnalysisStatus.STABLE,
                    RegressionAnalysisStatus.NO_RESULT,
                },
                RegressionAnalysisStatus.IMPROVEMENT,
            ),
            (
                {
                    RegressionAnalysisStatus.STABLE,
                    RegressionAnalysisStatus.NO_RESULT,
                },
                RegressionAnalysisStatus.STABLE,
            ),
            (
                {RegressionAnalysisStatus.NO_RESULT},
                RegressionAnalysisStatus.NO_RESULT,
            ),
            (set(), RegressionAnalysisStatus.NO_RESULT),
        ):
            with self.subTest(statuses=repr(statuses)):
                self.assertEqual(
                    QAWorkflow._summarize_status(statuses), expected_status
                )

    def _create_autopkgtest_artifact(
        self,
        statuses: dict[str, DebianAutopkgtestResultStatus],
        source_artifact: Artifact,
        work_request: WorkRequest,
        *,
        architecture: str = "amd64",
        distribution: str = "sid",
    ) -> Artifact:
        """Create autopkgtest artifact for regression analysis tests."""
        source_data = DebianSourcePackage(**source_artifact.data)
        source_url = f"http://example.com/{source_artifact.get_absolute_url()}"
        return self.playground.create_artifact(
            category=ArtifactCategory.AUTOPKGTEST,
            data=DebianAutopkgtest(
                results={
                    name: DebianAutopkgtestResult(status=status)
                    for name, status in statuses.items()
                },
                cmdline="unused",
                source_package=DebianAutopkgtestSource(
                    name=get_source_package_name(source_data),
                    version=get_source_package_version(source_data),
                    url=source_url,
                ),
                architecture=architecture,
                distribution=distribution,
            ),
            work_request=work_request,
        )[0]

    def _create_lintian_artifact(
        self,
        tags_count_by_severity: dict[DebianLintianSeverity, int],
        tags_found: list[str],
        work_request: WorkRequest,
        *,
        architecture: str = "amd64",
        distribution: str = "sid",
    ) -> Artifact:
        """Create lintian artifact for regression analysis tests."""
        return self.playground.create_artifact(
            category=ArtifactCategory.LINTIAN,
            data=DebianLintian(
                architecture=architecture,
                summary=DebianLintianSummary(
                    tags_count_by_severity=tags_count_by_severity,
                    package_filename={},
                    tags_found=tags_found,
                    overridden_tags_found=[],
                    lintian_version="2.122.0",
                    distribution=distribution,
                ),
            ),
            work_request=work_request,
        )[0]

    def _prepare_test_callback_regression_analysis(
        self,
    ) -> tuple[WorkRequest, dict[str, Artifact], Artifact]:
        reference_source_artifact = self.source_artifact
        new_source_artifact = self.playground.create_source_artifact(
            name="hello", version="1.0-2"
        )
        [new_binary_upload_artifact] = self.playground.create_upload_artifacts(
            source=False,
            src_name="hello",
            version="1.0-1",
            binary=True,
            binaries=[("hello", "amd64")],
        ).binaries
        self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        sid_qa_results = self.playground.create_collection(
            name="sid", category=CollectionCategory.QA_RESULTS
        )

        reference_autopkgtest_task = self.playground.create_worker_task(
            task_name="autopkgtest",
            result=WorkRequest.Results.SUCCESS,
            validate=False,
            use_data=False,
        )
        reference_autopkgtest_artifact = self._create_autopkgtest_artifact(
            {"upstream": DebianAutopkgtestResultStatus.PASS},
            reference_source_artifact,
            reference_autopkgtest_task,
        )
        sid_qa_results.manager.add_artifact(
            reference_autopkgtest_artifact,
            user=self.playground.get_default_user(),
            variables={
                "package": "hello",
                "version": "1.0-1",
                "architecture": "amd64",
                "timestamp": int(timezone.now().timestamp()),
                "work_request_id": reference_autopkgtest_task.id,
            },
        )

        reference_lintian_task = self.playground.create_worker_task(
            task_name="lintian",
            result=WorkRequest.Results.SUCCESS,
            validate=False,
            use_data=False,
        )
        reference_lintian_artifact = self._create_lintian_artifact(
            {}, [], reference_lintian_task
        )
        self.playground.create_artifact_relation(
            reference_lintian_artifact, reference_source_artifact
        )
        sid_qa_results.manager.add_artifact(
            reference_lintian_artifact,
            user=self.playground.get_default_user(),
            variables={
                "package": "hello",
                "version": "1.0-1",
                "architecture": "amd64",
                "timestamp": int(timezone.now().timestamp()),
                "work_request_id": reference_lintian_task.id,
            },
        )

        workflow = self.orchestrate(
            extra_data={
                "reference_prefix": "reference-qa-result|",
                "source_artifact": new_source_artifact.id,
                "binary_artifacts": ["internal@collections/name:build-amd64"],
                "reference_source_artifact": reference_source_artifact.id,
                "architectures": ["amd64"],
                "qa_suite": f"sid@{CollectionCategory.SUITE}",
                "reference_qa_results": f"sid@{CollectionCategory.QA_RESULTS}",
                "enable_regression_tracking": True,
                # Simplify this test by only enabling autopkgtest and lintian.
                "enable_autopkgtest": True,
                "enable_lintian": True,
                "enable_piuparts": False,
                "enable_blhc": False,
            },
            architectures=["amd64"],
            sbuild_workflow_source_artifact=new_source_artifact,
            internal_collection_artifacts=[
                (new_binary_upload_artifact, {"name": "build-amd64"})
            ],
        )
        return (
            workflow,
            {
                "autopkgtest:hello:amd64": reference_autopkgtest_artifact,
                "lintian:hello:amd64": reference_lintian_artifact,
            },
            new_source_artifact,
        )

    @preserve_db_task_registry()
    def test_callback_regression_analysis_success(self) -> None:
        """The collation callback succeeds if there were no regressions."""
        (workflow, reference_results, new_source_artifact) = (
            self._prepare_test_callback_regression_analysis()
        )
        assert workflow.parent is not None
        new_results: dict[str, Artifact] = {}
        analysis: dict[
            str, tuple[RegressionAnalysisStatus, dict[str, Any] | None]
        ] = {
            "": (RegressionAnalysisStatus.NO_RESULT, None),
            "autopkgtest:hello:amd64": (
                RegressionAnalysisStatus.NO_RESULT,
                {"upstream": "stable"},
            ),
            "lintian:hello:amd64": (
                RegressionAnalysisStatus.NO_RESULT,
                {"added_tags": [], "removed_tags": []},
            ),
        }

        # Add an extra sub-workflow with no regression-analysis callbacks,
        # to make sure that doesn't cause a problem.
        noop = workflow.create_child_workflow(task_name="noop")

        sbuild = workflow.parent.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        )
        autopkgtest = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="autopkgtest"
        )
        autopkgtest_ra = autopkgtest.children.workflow_callbacks().get(
            workflow_data_json__step="regression-analysis"
        )
        lintian = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="lintian"
        )
        lintian_ra = lintian.children.workflow_callbacks().get(
            workflow_data_json__step="regression-analysis"
        )
        ra = workflow.children.workflow_callbacks().get(
            workflow_data_json__step="regression-analysis"
        )

        with self.schedule_and_run() as work_requests:
            self.assertCountEqual(
                work_requests, [autopkgtest_ra, lintian_ra, ra]
            )

        autopkgtest_ra.refresh_from_db()
        self.assertEqual(autopkgtest_ra.status, WorkRequest.Statuses.BLOCKED)
        ra.refresh_from_db()
        self.assertEqual(ra.status, WorkRequest.Statuses.BLOCKED)
        workflow.refresh_from_db()
        self.assert_regression_analysis(
            workflow, "1.0-1", reference_results, "1.0-2", new_results, analysis
        )

        # Complete sbuild and autopkgtest.  The top-level analysis is updated.
        self.playground.advance_work_request(
            sbuild.children.get(task_type=TaskTypes.WORKER, task_name="sbuild"),
            result=WorkRequest.Results.SUCCESS,
        )
        autopkgtest_task = autopkgtest.children.get(
            task_type=TaskTypes.WORKER, task_name="autopkgtest"
        )
        new_results["autopkgtest:hello:amd64"] = (
            self._create_autopkgtest_artifact(
                {"upstream": DebianAutopkgtestResultStatus.PASS},
                new_source_artifact,
                autopkgtest_task,
            )
        )
        self.playground.advance_work_request(
            autopkgtest_task, result=WorkRequest.Results.SUCCESS, use_data=False
        )
        analysis[""] = (RegressionAnalysisStatus.STABLE, None)
        analysis["autopkgtest:hello:amd64"] = (
            RegressionAnalysisStatus.STABLE,
            {"upstream": "stable"},
        )

        self.schedule_and_run_workflow_callback(autopkgtest)
        self.schedule_and_run_workflow_callback(workflow)

        autopkgtest.refresh_from_db()
        self.assertEqual(autopkgtest.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(autopkgtest.result, WorkRequest.Results.SUCCESS)
        ra.refresh_from_db()
        self.assertEqual(ra.status, WorkRequest.Statuses.BLOCKED)
        workflow.refresh_from_db()
        self.assert_regression_analysis(
            workflow, "1.0-1", reference_results, "1.0-2", new_results, analysis
        )

        # Complete lintian.  The top-level analysis is updated, and we're done.
        lintian_task = lintian.children.get(
            task_type=TaskTypes.WORKER, task_name="lintian"
        )
        new_results["lintian:hello:amd64"] = self._create_lintian_artifact(
            {}, [], lintian_task
        )
        self.playground.create_artifact_relation(
            new_results["lintian:hello:amd64"], new_source_artifact
        )
        self.playground.advance_work_request(
            lintian_task, result=WorkRequest.Results.SUCCESS, use_data=False
        )
        analysis["lintian:hello:amd64"] = (
            RegressionAnalysisStatus.STABLE,
            {"added_tags": [], "removed_tags": []},
        )
        self.playground.advance_work_request(
            noop, result=WorkRequest.Results.SUCCESS
        )

        self.schedule_and_run_workflow_callback(lintian)
        self.schedule_and_run_workflow_callback(workflow)

        lintian.refresh_from_db()
        self.assertEqual(lintian.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(lintian.result, WorkRequest.Results.SUCCESS)
        ra.refresh_from_db()
        self.assertEqual(ra.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(ra.result, WorkRequest.Results.SUCCESS)
        workflow.refresh_from_db()
        self.assertEqual(workflow.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(workflow.result, WorkRequest.Results.SUCCESS)
        self.assert_regression_analysis(
            workflow, "1.0-1", reference_results, "1.0-2", new_results, analysis
        )

    @preserve_db_task_registry()
    def test_callback_regression_analysis_failure(self) -> None:
        """The collation callback fails if there were regressions."""
        (workflow, reference_results, new_source_artifact) = (
            self._prepare_test_callback_regression_analysis()
        )
        assert workflow.parent is not None
        new_results: dict[str, Artifact] = {}
        analysis: dict[
            str, tuple[RegressionAnalysisStatus, dict[str, Any] | None]
        ] = {
            "": (RegressionAnalysisStatus.NO_RESULT, None),
            "autopkgtest:hello:amd64": (
                RegressionAnalysisStatus.NO_RESULT,
                {"upstream": "stable"},
            ),
            "lintian:hello:amd64": (
                RegressionAnalysisStatus.NO_RESULT,
                {"added_tags": [], "removed_tags": []},
            ),
        }

        sbuild = workflow.parent.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        )
        autopkgtest = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="autopkgtest"
        )
        autopkgtest_ra = autopkgtest.children.workflow_callbacks().get(
            workflow_data_json__step="regression-analysis"
        )
        lintian = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="lintian"
        )
        lintian_ra = lintian.children.workflow_callbacks().get(
            workflow_data_json__step="regression-analysis"
        )
        ra = workflow.children.workflow_callbacks().get(
            workflow_data_json__step="regression-analysis"
        )

        with self.schedule_and_run() as work_requests:
            self.assertCountEqual(
                work_requests, [autopkgtest_ra, lintian_ra, ra]
            )

        autopkgtest_ra.refresh_from_db()
        self.assertEqual(autopkgtest_ra.status, WorkRequest.Statuses.BLOCKED)
        ra.refresh_from_db()
        self.assertEqual(ra.status, WorkRequest.Statuses.BLOCKED)
        workflow.refresh_from_db()
        self.assert_regression_analysis(
            workflow, "1.0-1", reference_results, "1.0-2", new_results, analysis
        )

        # Complete sbuild and autopkgtest.  The top-level analysis is updated.
        self.playground.advance_work_request(
            sbuild.children.get(task_type=TaskTypes.WORKER, task_name="sbuild"),
            result=WorkRequest.Results.SUCCESS,
        )
        autopkgtest_task = autopkgtest.children.get(
            task_type=TaskTypes.WORKER, task_name="autopkgtest"
        )
        new_results["autopkgtest:hello:amd64"] = (
            self._create_autopkgtest_artifact(
                {"upstream": DebianAutopkgtestResultStatus.FAIL},
                new_source_artifact,
                autopkgtest_task,
            )
        )
        self.playground.advance_work_request(
            autopkgtest_task, result=WorkRequest.Results.SUCCESS, use_data=False
        )
        analysis[""] = (RegressionAnalysisStatus.REGRESSION, None)
        analysis["autopkgtest:hello:amd64"] = (
            RegressionAnalysisStatus.REGRESSION,
            {"upstream": "regression"},
        )

        self.schedule_and_run_workflow_callback(autopkgtest)
        self.schedule_and_run_workflow_callback(workflow)

        ra.refresh_from_db()
        self.assertEqual(ra.status, WorkRequest.Statuses.BLOCKED)
        workflow.refresh_from_db()
        self.assertEqual(workflow.status, WorkRequest.Statuses.RUNNING)
        self.assert_regression_analysis(
            workflow, "1.0-1", reference_results, "1.0-2", new_results, analysis
        )

        # Complete lintian.  The top-level analysis is updated, and we're done.
        lintian_task = lintian.children.get(
            task_type=TaskTypes.WORKER, task_name="lintian"
        )
        new_results["lintian:hello:amd64"] = self._create_lintian_artifact(
            {DebianLintianSeverity.ERROR: 1}, ["test"], lintian_task
        )
        self.playground.create_artifact_relation(
            new_results["lintian:hello:amd64"], new_source_artifact
        )
        self.playground.advance_work_request(
            lintian_task, result=WorkRequest.Results.SUCCESS, use_data=False
        )
        analysis["lintian:hello:amd64"] = (
            RegressionAnalysisStatus.REGRESSION,
            {"added_tags": ["test"], "removed_tags": []},
        )

        self.schedule_and_run_workflow_callback(lintian)
        self.schedule_and_run_workflow_callback(workflow)

        ra.refresh_from_db()
        self.assertEqual(ra.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(ra.result, WorkRequest.Results.FAILURE)
        workflow.refresh_from_db()
        self.assertEqual(workflow.status, WorkRequest.Statuses.COMPLETED)
        self.assertEqual(workflow.result, WorkRequest.Results.FAILURE)
        self.assert_regression_analysis(
            workflow, "1.0-1", reference_results, "1.0-2", new_results, analysis
        )
