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

from typing import Any, ClassVar, override

from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianSourcePackage,
    TaskTypes,
)
from debusine.assets import KeyPurpose
from debusine.client.models import LookupChildType
from debusine.db.models import Artifact, WorkRequest
from debusine.server.workflows import (
    DebianPipelineWorkflow,
    WorkflowValidationError,
    workflow_utils,
)
from debusine.server.workflows.models import (
    DebianPipelineWorkflowData,
    DebianPipelineWorkflowDynamicData,
    QAFailOn,
    QAFailurePolicy,
    RegressionTrackingUpdateQAResults,
)
from debusine.server.workflows.tests.helpers import WorkflowTestCase
from debusine.tasks import TaskConfigError
from debusine.tasks.models import (
    BackendType,
    CollectionItemMatcherKind,
    ExtraDebusineRepository,
    ExtraExternalRepository,
    InputArtifactSingle,
    LintianFailOnSeverity,
    OutputData,
    OutputDataError,
)


class DebianPipelineWorkflowTests(WorkflowTestCase[DebianPipelineWorkflow]):
    """Unit tests for :py:class:`DebianPipelineWorkflow`."""

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

    source_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"
        )

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

    def create_debian_pipeline_workflow(
        self,
        *,
        extra_task_data: dict[str, Any] | None = None,
        validate: bool = True,
    ) -> DebianPipelineWorkflow:
        """Create a debian pipeline workflow."""
        task_data = {
            "source_artifact": self.source_artifact.id,
            "vendor": "debian",
            "codename": "bookworm",
        }
        if extra_task_data is not None:
            task_data.update(extra_task_data)
        wr = self.playground.create_workflow(
            task_name="debian_pipeline", task_data=task_data, validate=validate
        )
        return self.get_workflow(wr)

    def orchestrate(
        self, extra_data: dict[str, Any], expect_success: bool = True
    ) -> WorkRequest:
        """Create and orchestrate a DebianPipelineWorkflow."""
        data = {
            "source_artifact": self.source_artifact.id,
            "vendor": "debian",
            "codename": "bookworm",
            "enable_autopkgtest": False,
            "enable_lintian": False,
            "enable_piuparts": False,
            "enable_blhc": False,
            **extra_data,
        }

        wr = self.playground.create_workflow(
            task_name="debian_pipeline",
            task_data=DebianPipelineWorkflowData(**data),
        )
        self.assertEqual(wr.status, WorkRequest.Statuses.PENDING)
        self.assertEqual(
            self.playground.orchestrate_workflow(wr, use_data=False),
            expect_success,
        )

        return wr

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

        workflow.validate_input()

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

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

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

    def test_compute_dynamic_data(self) -> None:
        """Dynamic data includes a parameter summary."""
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            DebianPipelineWorkflowDynamicData(
                subject="hello",
                parameter_summary="hello_1.0-1",
                input_source_artifact_id=self.source_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
        )
        with self.assertRaisesRegex(
            TaskConfigError,
            r"^source_artifact: unexpected artifact category: 'debusine:test'. "
            r"Valid categories: \['debian:source-package', 'debian:upload'\]$",
        ):
            self.playground.create_workflow(
                task_name="debian_pipeline",
                task_data=DebianPipelineWorkflowData(
                    source_artifact=artifact.id, vendor="debian", codename="sid"
                ),
            )

    def test_get_input_artifacts(self) -> None:
        upload_artifacts = self.playground.create_upload_artifacts()
        self.task_data["source_artifact"] = upload_artifacts.upload.id
        workflow = self.task

        # workflow.dynamic_data is None
        self.assertIsNone(workflow.dynamic_data)
        self.assertEqual(
            workflow.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=workflow.data.source_artifact,
                    label="source_artifact",
                    artifact_id=None,
                )
            ],
        )

        # workflow.dynamic_data.source_artifact_id is None
        workflow.dynamic_data = DebianPipelineWorkflowDynamicData()
        self.assertEqual(
            workflow.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=workflow.data.source_artifact,
                    label="source_artifact",
                    artifact_id=None,
                )
            ],
        )

        # workflow.dynamic_data.source_artifact_id is the id: return it
        workflow.dynamic_data.input_source_artifact_id = (
            upload_artifacts.upload.id
        )
        self.assertEqual(
            workflow.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=workflow.data.source_artifact,
                    label="source_artifact",
                    artifact_id=upload_artifacts.upload.id,
                )
            ],
        )

    def test_get_extra_repositories_none(self) -> None:
        self.assertIsNone(self.task.get_extra_repositories())

    def test_get_extra_repositories_explicit(self) -> None:
        external = ExtraExternalRepository(
            url="http://example.com/", suite="sid", components=["main"]
        )
        self.task_data["extra_repositories"] = [external]
        self.assertEqual(self.task.get_extra_repositories(), [external])

    def test_get_extra_repositories_suite_without_active_indexes(self) -> None:
        self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        self.task_data["suite"] = f"sid@{CollectionCategory.SUITE}"
        self.assertIsNone(self.task.get_extra_repositories())

    def test_get_extra_repositories_suite(self) -> None:
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        sid.manager.add_artifact(
            self.playground.create_repository_index("InRelease"),
            user=self.playground.get_default_user(),
            variables={"path": "InRelease"},
        )
        self.task_data["suite"] = f"sid@{CollectionCategory.SUITE}"
        self.assertEqual(
            self.task.get_extra_repositories(),
            [ExtraDebusineRepository(suite=f"sid@{CollectionCategory.SUITE}")],
        )

    def test_get_extra_repositories_suite_and_external(self) -> None:
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        sid.manager.add_artifact(
            self.playground.create_repository_index("InRelease"),
            user=self.playground.get_default_user(),
            variables={"path": "InRelease"},
        )
        external = ExtraExternalRepository(
            url="http://example.com/", suite="sid", components=["main"]
        )
        self.task_data["suite"] = f"sid@{CollectionCategory.SUITE}"
        self.task_data["extra_repositories"] = [external]
        self.assertEqual(
            self.task.get_extra_repositories(),
            [
                ExtraDebusineRepository(
                    suite=f"sid@{CollectionCategory.SUITE}"
                ),
                external,
            ],
        )

    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.assertEqual(
            workflow_utils.source_package(self.task), upload_artifacts.source
        )

    def test_populate_use_available_architectures(self) -> None:
        """
        Test populate use available architectures.

        The user didn't specify "architectures", DebianPipelineWorkflow
        checks available architectures and "all" and use them.
        """
        collection = self.playground.create_collection(
            "debian",
            CollectionCategory.ENVIRONMENTS,
            workspace=self.playground.get_default_workspace(),
        )
        for arch in ["amd64", "i386"]:
            artifact, _ = self.playground.create_artifact(
                category=ArtifactCategory.SYSTEM_TARBALL,
                data={"codename": "bookworm", "architecture": arch},
            )
            collection.manager.add_artifact(
                artifact, user=self.playground.get_default_user()
            )

        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "vendor": "debian",
                "codename": "bookworm",
            }
        )

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

        self.assertEqual(
            sbuild.task_data["architectures"], ["all", "amd64", "i386"]
        )

    def test_populate_sbuild(self) -> None:
        """Test populate create sbuild."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_blhc": False,
            }
        )

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

        self.assertEqual(sbuild.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            sbuild.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "backend": BackendType.AUTO,
                "environment_variant": None,
                "extra_repositories": None,
                "input": {"source_artifact": self.source_artifact.id},
                "target_distribution": "debian:bookworm",
                "signing_template_names": {},
            },
        )
        self.assertEqual(
            sbuild.workflow_data_json,
            {"display_name": "sbuild", "step": "sbuild"},
        )

        # SbuildWorkflow.populate() was called and created its tasks
        self.assertTrue(sbuild.children.exists())
        for child in sbuild.children.all():
            self.assertEqual(child.status, WorkRequest.Statuses.PENDING)

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

    def test_populate_sbuild_with_suite(self) -> None:
        bookworm = self.playground.create_collection(
            name="bookworm", category=CollectionCategory.SUITE
        )
        bookworm.manager.add_artifact(
            self.playground.create_repository_index("InRelease"),
            user=self.playground.get_default_user(),
            variables={"path": "InRelease"},
        )

        workflow = self.orchestrate(
            extra_data={
                "suite": f"bookworm@{CollectionCategory.SUITE}",
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
            }
        )

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

        self.assertEqual(sbuild.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            sbuild.task_data["extra_repositories"],
            [{"suite": f"bookworm@{CollectionCategory.SUITE}"}],
        )

    def assert_qa(
        self, workflow: WorkRequest, task_data: dict[str, Any]
    ) -> None:
        """Assert workflow has a sub-workflow qa with task_data."""
        qa = workflow.children.get(task_name="qa", task_type=TaskTypes.WORKFLOW)
        self.assertEqual(qa.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(qa.task_data, task_data)

        self.assertEqual(
            qa.workflow_data_json, {"display_name": "QA", "step": "qa"}
        )

        # QaWorkflow.populate() was called and created its tasks
        self.assertTrue(qa.children.exists())
        for child in qa.children.all():
            # Children are sub-workflows, so go straight to RUNNING.
            self.assertEqual(child.status, WorkRequest.Statuses.RUNNING)

        self.assertQuerySetEqual(
            qa.children.earliest("id").dependencies.all(), []
        )

    def test_populate_qa_with_sbuild_subset(self) -> None:
        """`sbuild` may only create a subset of the requested architectures."""
        self.assertEqual(
            self.source_artifact.data["dsc_fields"]["Architecture"], "any"
        )

        workflow = self.orchestrate(
            extra_data={
                "architectures": ["all", "amd64"],
                "source_artifact": f"{self.source_artifact.id}@artifacts",
                "enable_lintian": True,
                "extra_repositories": [
                    {
                        "url": "http://example.com/",
                        "suite": "bookworm",
                        "components": ["main"],
                    }
                ],
            }
        )

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

        self.assertEqual(
            sbuild.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["all", "amd64"],
                "backend": BackendType.AUTO,
                "environment_variant": None,
                "input": {
                    "source_artifact": f"{self.source_artifact.id}@artifacts"
                },
                "target_distribution": "debian:bookworm",
                "signing_template_names": {},
                "extra_repositories": [
                    {
                        "url": "http://example.com/",
                        "suite": "bookworm",
                        "components": ["main"],
                    }
                ],
            },
        )
        self.assertEqual(
            sbuild.workflow_data_json,
            {"display_name": "sbuild", "step": "sbuild"},
        )

        self.assertTrue(sbuild.children.exists())
        self.assert_qa(
            workflow,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["all", "amd64"],
                "autopkgtest_backend": BackendType.AUTO,
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "bookworm",
                "extra_repositories": [
                    {
                        "url": "http://example.com/",
                        "suite": "bookworm",
                        "components": ["main"],
                    }
                ],
                "enable_autopkgtest": False,
                "enable_lintian": True,
                "enable_piuparts": False,
                "enable_regression_tracking": False,
                "enable_reverse_dependencies_autopkgtest": False,
                "enable_debdiff": False,
                "enable_blhc": False,
                "fail_on": QAFailOn.FAILURE,
                "lintian_backend": BackendType.AUTO,
                "lintian_fail_on_severity": LintianFailOnSeverity.ERROR,
                "piuparts_backend": BackendType.AUTO,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "reference_source_artifact": None,
                "source_artifact": f"{self.source_artifact.id}@artifacts",
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
            },
        )

    def test_populate_qa_lintian(self) -> None:
        """Test populate create qa: lintian enabled."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_lintian": True,
            }
        )
        self.assert_qa(
            workflow,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "autopkgtest_backend": BackendType.AUTO,
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "bookworm",
                "extra_repositories": None,
                "enable_autopkgtest": False,
                "enable_lintian": True,
                "enable_piuparts": False,
                "enable_regression_tracking": False,
                "enable_reverse_dependencies_autopkgtest": False,
                "enable_debdiff": False,
                "enable_blhc": False,
                "fail_on": QAFailOn.FAILURE,
                "lintian_backend": BackendType.AUTO,
                "lintian_fail_on_severity": LintianFailOnSeverity.ERROR,
                "piuparts_backend": BackendType.AUTO,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "reference_source_artifact": None,
                "source_artifact": self.source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
            },
        )

    def test_populate_qa_autopkgtest(self) -> None:
        """Test populate create qa: autopkgtest enabled."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_autopkgtest": True,
            }
        )

        self.assert_qa(
            workflow,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "autopkgtest_backend": BackendType.AUTO,
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "bookworm",
                "extra_repositories": None,
                "enable_autopkgtest": True,
                "enable_lintian": False,
                "enable_piuparts": False,
                "enable_regression_tracking": False,
                "enable_reverse_dependencies_autopkgtest": False,
                "enable_debdiff": False,
                "enable_blhc": False,
                "fail_on": QAFailOn.FAILURE,
                "lintian_backend": BackendType.AUTO,
                "lintian_fail_on_severity": LintianFailOnSeverity.ERROR,
                "piuparts_backend": BackendType.AUTO,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "reference_source_artifact": None,
                "source_artifact": self.source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
            },
        )

    def test_populate_qa_piuparts(self) -> None:
        """Test populate create qa: piuparts enabled."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_piuparts": True,
            }
        )

        self.assert_qa(
            workflow,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "autopkgtest_backend": BackendType.AUTO,
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "bookworm",
                "extra_repositories": None,
                "enable_autopkgtest": False,
                "enable_lintian": False,
                "enable_piuparts": True,
                "enable_regression_tracking": False,
                "enable_reverse_dependencies_autopkgtest": False,
                "enable_debdiff": False,
                "enable_blhc": False,
                "fail_on": QAFailOn.FAILURE,
                "lintian_backend": BackendType.AUTO,
                "lintian_fail_on_severity": LintianFailOnSeverity.ERROR,
                "piuparts_backend": BackendType.AUTO,
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "reference_source_artifact": None,
                "source_artifact": self.source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
            },
        )

    def test_populate_qa_debdiff(self) -> None:
        """Test populate create qa: debdiff enabled."""
        trixie = self.playground.create_collection(
            name="trixie",
            category=CollectionCategory.SUITE,
        )

        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_debdiff": True,
                "qa_suite": "trixie@debian:suite",
            }
        )

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

        self.assertEqual(
            qa.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "autopkgtest_backend": BackendType.AUTO,
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "bookworm",
                "extra_repositories": None,
                "enable_autopkgtest": False,
                "enable_lintian": False,
                "enable_piuparts": False,
                "enable_regression_tracking": False,
                "enable_reverse_dependencies_autopkgtest": False,
                "enable_debdiff": True,
                "enable_blhc": False,
                "fail_on": QAFailOn.FAILURE,
                "lintian_backend": BackendType.AUTO,
                "lintian_fail_on_severity": LintianFailOnSeverity.ERROR,
                "piuparts_backend": BackendType.AUTO,
                "prefix": "",
                "qa_suite": str(trixie),
                "reference_prefix": "",
                "reference_qa_results": None,
                "reference_source_artifact": None,
                "source_artifact": self.source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
            },
        )

        self.assertEqual(qa.status, WorkRequest.Statuses.RUNNING)

        self.assertEqual(
            qa.children.filter(
                task_name="debdiff", task_type=TaskTypes.WORKFLOW
            ).count(),
            1,
        )

    def test_populate_qa_piuparts_configuration(self) -> None:
        """Test populate create qa: piuparts non-default configuration."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_piuparts": True,
                "piuparts_backend": BackendType.INCUS_LXC,
                "piuparts_environment": "debian/match:codename=trixie",
            }
        )

        self.assert_qa(
            workflow,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["amd64"],
                "autopkgtest_backend": BackendType.AUTO,
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "bookworm",
                "extra_repositories": None,
                "enable_autopkgtest": False,
                "enable_lintian": False,
                "enable_piuparts": True,
                "enable_regression_tracking": False,
                "enable_reverse_dependencies_autopkgtest": False,
                "enable_debdiff": False,
                "enable_blhc": False,
                "fail_on": QAFailOn.FAILURE,
                "lintian_backend": BackendType.AUTO,
                "lintian_fail_on_severity": LintianFailOnSeverity.ERROR,
                "piuparts_backend": BackendType.INCUS_LXC,
                "piuparts_environment": "debian/match:codename=trixie",
                "prefix": "",
                "qa_suite": None,
                "reference_prefix": "",
                "reference_qa_results": None,
                "reference_source_artifact": None,
                "source_artifact": self.source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
                "vendor": "debian",
            },
        )

    def test_populate_qa_enable_regression_tracking_no_reference_source(
        self,
    ) -> None:
        self.playground.create_collection(
            name="trixie", category=CollectionCategory.SUITE
        )
        self.playground.create_collection(
            "trixie", CollectionCategory.QA_RESULTS
        )

        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "qa_suite": f"trixie@{CollectionCategory.SUITE}",
                "regression_tracking_qa_results": (
                    f"trixie@{CollectionCategory.QA_RESULTS}"
                ),
                "enable_regression_tracking": True,
                "enable_autopkgtest": True,
            }
        )

        normal_qa = workflow.children.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="qa",
            task_data__update_qa_results=RegressionTrackingUpdateQAResults.NO,
        )
        self.assertDictContainsAll(
            normal_qa.task_data,
            {
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "enable_regression_tracking": False,
                "fail_on": QAFailOn.FAILURE,
                "prefix": "",
                "reference_prefix": "",
                "reference_qa_results": None,
                "reference_source_artifact": None,
                "source_artifact": self.source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
            },
        )
        self.assertEqual(
            normal_qa.workflow_data_json, {"display_name": "QA", "step": "qa"}
        )

        self.assertFalse(
            workflow.children.filter(
                task_type=TaskTypes.WORKFLOW,
                task_name="qa",
                task_data__update_qa_results=(
                    RegressionTrackingUpdateQAResults.YES
                ),
            ).exists()
        )

    def test_populate_qa_enable_regression_tracking(self) -> None:
        reference_source_artifact = self.source_artifact
        reference_binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello",
                srcpkg_version="1.0-1",
                package="hello",
                version="1.0-1",
            )
        )
        new_source_artifact = self.playground.create_source_artifact(
            name="hello", version="1.0-2"
        )
        trixie = self.playground.create_collection(
            name="trixie", category=CollectionCategory.SUITE
        )
        trixie.manager.add_artifact(
            reference_source_artifact,
            user=self.playground.get_default_user(),
            variables={"component": "main", "section": "devel"},
        )
        trixie.manager.add_artifact(
            reference_binary_artifact,
            user=self.playground.get_default_user(),
            variables={
                "component": "main",
                "section": "devel",
                "priority": "optional",
            },
        )
        self.playground.create_collection(
            "trixie", CollectionCategory.QA_RESULTS
        )

        workflow = self.orchestrate(
            extra_data={
                "architectures": ["all", "amd64"],
                "source_artifact": new_source_artifact.id,
                "qa_suite": f"trixie@{CollectionCategory.SUITE}",
                "regression_tracking_qa_results": (
                    f"trixie@{CollectionCategory.QA_RESULTS}"
                ),
                "enable_regression_tracking": True,
                "enable_autopkgtest": True,
                "enable_debdiff": True,
            }
        )

        normal_qa = workflow.children.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="qa",
            task_data__update_qa_results=RegressionTrackingUpdateQAResults.NO,
        )
        self.assertDictContainsAll(
            normal_qa.task_data,
            {
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "enable_debdiff": True,
                "enable_regression_tracking": True,
                "fail_on": QAFailOn.REGRESSION,
                "prefix": "",
                "reference_prefix": "reference-qa-result|",
                "reference_qa_results": (
                    f"trixie@{CollectionCategory.QA_RESULTS}"
                ),
                "reference_source_artifact": reference_source_artifact.id,
                "source_artifact": new_source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.NO,
            },
        )
        self.assertEqual(
            normal_qa.workflow_data_json, {"display_name": "QA", "step": "qa"}
        )

        extra_qa = workflow.children.get(
            task_type=TaskTypes.WORKFLOW,
            task_name="qa",
            task_data__update_qa_results=RegressionTrackingUpdateQAResults.YES,
        )
        self.assertDictContainsAll(
            extra_qa.task_data,
            {
                "binary_artifacts": [
                    {
                        "category": ArtifactCategory.BINARY_PACKAGE,
                        "collection": f"trixie@{CollectionCategory.SUITE}",
                        "data_matchers": [
                            [
                                "srcpkg_name",
                                {
                                    "kind": CollectionItemMatcherKind.EXACT,
                                    "value": "hello",
                                },
                            ],
                            [
                                "srcpkg_version",
                                {
                                    "kind": CollectionItemMatcherKind.EXACT,
                                    "value": "1.0-1",
                                },
                            ],
                        ],
                    }
                ],
                "enable_debdiff": False,
                "enable_regression_tracking": False,
                "fail_on": QAFailOn.NEVER,
                "prefix": "reference-qa-result|",
                "reference_prefix": "",
                "reference_qa_results": (
                    f"trixie@{CollectionCategory.QA_RESULTS}"
                ),
                "reference_source_artifact": None,
                "source_artifact": reference_source_artifact.id,
                "update_qa_results": RegressionTrackingUpdateQAResults.YES,
            },
        )
        self.assertEqual(
            extra_qa.workflow_data_json,
            {
                "display_name": "QA (reference tests for hello_1.0-1)",
                "step": "qa-reference",
                "expanded": False,
            },
        )

        normal_autopkgtest = normal_qa.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="autopkgtest"
        )
        normal_autopkgtest_regression_analysis = (
            normal_autopkgtest.children.workflow_callbacks().get(
                workflow_data_json__step="regression-analysis"
            )
        )
        extra_autopkgtest = extra_qa.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="autopkgtest"
        )
        self.assertQuerySetEqual(
            normal_autopkgtest_regression_analysis.dependencies.all(),
            [
                normal_autopkgtest.children.get(
                    task_type=TaskTypes.WORKER, task_name="autopkgtest"
                ),
                extra_autopkgtest.children.get(
                    task_type=TaskTypes.WORKER, task_name="autopkgtest"
                ),
            ],
            ordered=False,
        )

    def test_populate_qa_with_suite(self) -> None:
        bookworm = self.playground.create_collection(
            name="bookworm", category=CollectionCategory.SUITE
        )
        bookworm.manager.add_artifact(
            self.playground.create_repository_index("InRelease"),
            user=self.playground.get_default_user(),
            variables={"path": "InRelease"},
        )

        workflow = self.orchestrate(
            extra_data={
                "suite": f"bookworm@{CollectionCategory.SUITE}",
                "architectures": ["all", "amd64"],
                "source_artifact": f"{self.source_artifact.id}@artifacts",
                "enable_lintian": True,
            }
        )

        qa = workflow.children.get(task_name="qa", task_type=TaskTypes.WORKFLOW)
        self.assertEqual(qa.status, WorkRequest.Statuses.RUNNING)
        self.assertEqual(
            qa.task_data["extra_repositories"],
            [{"suite": f"bookworm@{CollectionCategory.SUITE}"}],
        )

    def test_populate_architectures_allow_deny(self) -> None:
        """Populate uses architectures, architectures_allow/deny."""
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "architectures": ["amd64", "i386", "arm64"],
                "architectures_allowlist": ["amd64", "i386"],
                "architectures_denylist": ["amd64"],
            }
        )

        sbuild = workflow.children.get(
            task_name="sbuild", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(
            sbuild.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "architectures": ["i386"],
                "backend": BackendType.AUTO,
                "environment_variant": None,
                "extra_repositories": None,
                "input": {"source_artifact": self.source_artifact.id},
                "target_distribution": "debian:bookworm",
                "signing_template_names": {},
            },
        )

    def test_populate_make_signed_source_missing_signed_source_purpose(
        self,
    ) -> None:
        """Orchestrator error: missing required field."""
        msg = (
            'Orchestrator failed: "make_signed_source_purpose" must be '
            'set when signing the source'
        )
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "architectures": ["amd64"],
                "enable_make_signed_source": True,
                "signing_template_names": {"amd64": ["hello"]},
            },
            expect_success=False,
        )
        self.assertEqual(
            workflow.output_data,
            OutputData(
                errors=[
                    OutputDataError(message=msg, code="orchestrator-failed")
                ]
            ),
        )

    def test_populate_make_signed_source_missing_make_signed_source_key(
        self,
    ) -> None:
        """Orchestrator error: missing required field."""
        msg = (
            'Orchestrator failed: "make_signed_source_key" must be set '
            'when signing the source'
        )
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "architectures": ["amd64"],
                "enable_make_signed_source": True,
                "make_signed_source_purpose": KeyPurpose.UEFI,
                "signing_template_names": {"amd64": ["hello"]},
            },
            expect_success=False,
        )
        self.assertEqual(
            workflow.output_data,
            OutputData(
                errors=[
                    OutputDataError(message=msg, code="orchestrator-failed")
                ]
            ),
        )

    def test_populate_make_signed_source(self) -> None:
        """Populate create a make_signed_source workflow."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_make_signed_source": True,
                "make_signed_source_purpose": KeyPurpose.UEFI,
                "signing_template_names": {"amd64": ["hello"]},
                "make_signed_source_key": "ABC123",
            }
        )

        expected_binary_artifacts = [
            {
                "collection": "internal@collections",
                "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                "name_matcher": {
                    "kind": CollectionItemMatcherKind.STARTSWITH,
                    "value": "build-",
                },
            }
        ]
        make_signed_source = workflow.children.get(
            task_name="make_signed_source", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(
            make_signed_source.task_data,
            {
                "architectures": ["amd64"],
                "binary_artifacts": expected_binary_artifacts,
                "codename": "bookworm",
                "key": "ABC123",
                "purpose": "uefi",
                "sbuild_backend": BackendType.AUTO,
                "signing_template_artifacts": [
                    "internal@collections/name:signing-template-amd64-hello"
                ],
                "vendor": "debian",
            },
        )

        self.assertEqual(
            make_signed_source.workflow_data_json,
            {
                "display_name": "make signed source",
                "step": "make_signed_source",
            },
        )
        self.assertQuerySetEqual(make_signed_source.dependencies.all(), [])

        # MakeSignedSource.populate() was called and created its tasks.
        # They are all blocked, with the first one depending on sbuild and
        # the others depending on the previous one in sequence.
        self.assertQuerySetEqual(
            make_signed_source.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "extractforsigning",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                ),
                (
                    TaskTypes.SIGNING,
                    "sign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "extractforsigning",
                ),
                (
                    TaskTypes.WORKER,
                    "assemblesignedsource",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.SIGNING,
                    "sign",
                ),
                (
                    TaskTypes.WORKFLOW,
                    "sbuild",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "assemblesignedsource",
                ),
            ],
        )

    def test_populate_make_signed_source_qa_failure_policy_fail(self) -> None:
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_autopkgtest": True,
                "qa_failure_policy": QAFailurePolicy.FAIL,
                "enable_make_signed_source": True,
                "make_signed_source_purpose": KeyPurpose.UEFI,
                "signing_template_names": {"amd64": ["hello"]},
                "make_signed_source_key": "ABC123",
            }
        )

        self.assertFalse(
            workflow.children.filter(
                task_type=TaskTypes.WAIT, task_name="confirm"
            ).exists()
        )

        qa = workflow.children.get(task_type=TaskTypes.WORKFLOW, task_name="qa")
        make_signed_source = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="make_signed_source"
        )
        self.assertQuerySetEqual(make_signed_source.dependencies.all(), [qa])

    def test_populate_make_signed_source_qa_failure_policy_confirm(
        self,
    ) -> None:
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_autopkgtest": True,
                "qa_failure_policy": QAFailurePolicy.CONFIRM,
                "enable_confirmation": True,
                "enable_make_signed_source": True,
                "make_signed_source_purpose": KeyPurpose.UEFI,
                "signing_template_names": {"amd64": ["hello"]},
                "make_signed_source_key": "ABC123",
            }
        )

        qa = workflow.children.get(task_type=TaskTypes.WORKFLOW, task_name="qa")
        confirm = workflow.children.get(
            task_type=TaskTypes.WAIT, task_name="confirm"
        )
        self.assertEqual(
            confirm.task_data,
            {
                "auto_confirm_if_no_failure": True,
                "confirm_label": "Continue the workflow despite QA failures",
            },
        )
        self.assertEqual(
            confirm.workflow_data_json,
            {"display_name": "Confirm", "step": "confirm", "needs_input": True},
        )
        self.assertQuerySetEqual(confirm.dependencies.all(), [qa])

        make_signed_source = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="make_signed_source"
        )
        self.assertQuerySetEqual(
            make_signed_source.dependencies.all(), [confirm]
        )

    def test_populate_make_signed_source_confirmation(self) -> None:
        """If confirming, make_signed_source depends on the confirm task."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_confirmation": True,
                "enable_make_signed_source": True,
                "make_signed_source_purpose": KeyPurpose.UEFI,
                "signing_template_names": {"amd64": ["hello"]},
                "make_signed_source_key": "ABC123",
            }
        )

        confirm = workflow.children.get(
            task_type=TaskTypes.WAIT, task_name="confirm"
        )
        self.assertEqual(
            confirm.task_data,
            {
                "auto_confirm_if_no_failure": False,
                "confirm_label": "Continue the workflow",
            },
        )
        self.assertEqual(
            confirm.workflow_data_json,
            {"display_name": "Confirm", "step": "confirm", "needs_input": True},
        )
        self.assertQuerySetEqual(confirm.dependencies.all(), [])

        make_signed_source = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="make_signed_source"
        )
        self.assertQuerySetEqual(
            make_signed_source.dependencies.all(), [confirm]
        )

    def test_populate_package_upload(self) -> None:
        """Populate create an upload_package workflow."""
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "architectures": ["amd64", "i386"],
                "enable_upload": True,
                "vendor": "debian",
                "codename": "trixie",
                "upload_merge_uploads": False,
                "upload_since_version": "1.1",
                "upload_target": "ftp://user@example.org/pub/UploadQueue1/",
                "upload_target_distribution": "debian:testing",
            }
        )

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

        self.assertEqual(
            package_upload.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "trixie",
                "delayed_days": None,
                "key": None,
                "binary_key": None,
                "merge_uploads": False,
                "require_signature": True,
                "since_version": "1.1",
                "source_artifact": self.source_artifact.id,
                "target": "ftp://user@example.org/pub/UploadQueue1/",
                "target_distribution": "debian:testing",
                "vendor": "debian",
            },
        )

        self.assertEqual(
            package_upload.workflow_data_json,
            {
                "display_name": "package upload (hello)",
                "step": "package_upload/hello",
            },
        )
        self.assertQuerySetEqual(package_upload.dependencies.all(), [])

        # PackageUpload.populate() was called and created its tasks.
        # MakeSourcePackageUpload only requires the source artifact and so
        # can start immediately.  Everything else is blocked.
        self.assertQuerySetEqual(
            package_upload.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            # Work requests to construct, sign, and upload the source package.
            [
                (
                    TaskTypes.WORKER,
                    "makesourcepackageupload",
                    WorkRequest.Statuses.PENDING,
                    None,
                    None,
                ),
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "makesourcepackageupload",
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
            ]
            # Work requests to sign and upload each of the two binary packages.
            + [
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
            ]
            * 2,
        )

    def test_populate_package_upload_signed_source(self) -> None:
        """Populate creates package_upload workflows for signed source."""
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "architectures": ["amd64", "i386"],
                "enable_make_signed_source": True,
                "make_signed_source_purpose": KeyPurpose.UEFI,
                "signing_template_names": {
                    "amd64": ["hello-amd64-signed-template"]
                },
                "make_signed_source_key": "ABC123",
                "enable_upload": True,
                "vendor": "debian",
                "codename": "trixie",
                "upload_merge_uploads": False,
                "upload_since_version": "1.1",
                "upload_target": "ftp://user@example.org/pub/UploadQueue1/",
                "upload_target_distribution": "debian:testing",
            }
        )

        initial_sbuild_workflow = workflow.children.get(
            task_name="sbuild", task_type=TaskTypes.WORKFLOW
        )
        make_signed_source = workflow.children.get(
            task_name="make_signed_source", task_type=TaskTypes.WORKFLOW
        )
        extract_for_signing = make_signed_source.children.get(
            task_name="extractforsigning", task_type=TaskTypes.WORKER
        )
        sign = make_signed_source.children.get(
            task_name="sign", task_type=TaskTypes.SIGNING
        )
        assemble_signed_source = make_signed_source.children.get(
            task_name="assemblesignedsource", task_type=TaskTypes.WORKER
        )
        package_upload_source, package_upload_signed_amd64 = (
            workflow.children.filter(
                task_name="package_upload", task_type=TaskTypes.WORKFLOW
            ).order_by("id")
        )

        self.assertEqual(
            package_upload_source.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "trixie",
                "delayed_days": None,
                "key": None,
                "binary_key": None,
                "merge_uploads": False,
                "require_signature": True,
                "since_version": "1.1",
                "source_artifact": self.source_artifact.id,
                "target": "ftp://user@example.org/pub/UploadQueue1/",
                "target_distribution": "debian:testing",
                "vendor": "debian",
            },
        )

        self.assertEqual(
            package_upload_source.workflow_data_json,
            {
                "display_name": "package upload (hello)",
                "step": "package_upload/hello",
            },
        )
        self.assertQuerySetEqual(package_upload_source.dependencies.all(), [])

        # PackageUpload.populate() was called and created its tasks.
        # MakeSourcePackageUpload only requires the source artifact and so
        # can start immediately.  Everything else is blocked.
        self.assertQuerySetEqual(
            package_upload_source.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            # Work requests to construct, sign, and upload the source package.
            [
                (
                    TaskTypes.WORKER,
                    "makesourcepackageupload",
                    WorkRequest.Statuses.PENDING,
                    None,
                    None,
                ),
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "makesourcepackageupload",
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
            ]
            # Work requests to sign and upload each of the two binary packages.
            + [
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
            ]
            * 2,
        )

        self.assertEqual(
            package_upload_signed_amd64.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": (
                                "signed-source-amd64-"
                                "hello-amd64-signed-template|build-"
                            ),
                        },
                    }
                ],
                "codename": "trixie",
                "delayed_days": None,
                "key": None,
                "binary_key": None,
                "merge_uploads": False,
                "require_signature": True,
                "since_version": "1.1",
                "source_artifact": (
                    "internal@collections/name:"
                    "signed-source-amd64-hello-amd64-signed-template"
                ),
                "target": "ftp://user@example.org/pub/UploadQueue1/",
                "target_distribution": "debian:testing",
                "vendor": "debian",
            },
        )

        self.assertEqual(
            package_upload_signed_amd64.workflow_data_json,
            {
                "display_name": (
                    "package upload (amd64/hello-amd64-signed-template)"
                ),
                "step": "package_upload/amd64/hello-amd64-signed-template",
            },
        )

        # The workflow can't orchestrate uploads of the signed source
        # package and its binaries until the signed source package has been
        # assembled.
        self.assertEqual(
            package_upload_signed_amd64.status, WorkRequest.Statuses.BLOCKED
        )
        self.assertQuerySetEqual(
            package_upload_signed_amd64.dependencies.all(),
            [assemble_signed_source],
        )
        self.assertQuerySetEqual(package_upload_signed_amd64.children.all(), [])

        # Assembling the signed source package causes the remaining uploads
        # to be orchestrated.
        self.simulate_sbuild_workflow_completion(
            initial_sbuild_workflow, use_data=False
        )
        self.playground.create_signing_input_artifact(
            binary_package_name="hello-signed",
            workspace=extract_for_signing.workspace,
            work_request=extract_for_signing,
        )
        extract_for_signing.refresh_from_db()
        self.playground.advance_work_request(
            extract_for_signing,
            result=WorkRequest.Results.SUCCESS,
            use_data=False,
        )
        self.playground.create_signing_output_artifact(
            purpose=KeyPurpose.UEFI,
            fingerprint="ABC123",
            binary_package_name="hello-signed",
            workspace=sign.workspace,
            work_request=sign,
        )
        sign.refresh_from_db()
        self.playground.advance_work_request(
            sign, result=WorkRequest.Results.SUCCESS, use_data=False
        )
        self.playground.create_upload_artifacts(
            src_name="hello-signed-template",
            source=True,
            binary=False,
            workspace=assemble_signed_source.workspace,
            work_request=assemble_signed_source,
        )
        assemble_signed_source.refresh_from_db()
        self.playground.advance_work_request(
            assemble_signed_source,
            result=WorkRequest.Results.SUCCESS,
            use_data=False,
        )
        package_upload_signed_amd64.refresh_from_db()
        self.assertEqual(
            package_upload_signed_amd64.status, WorkRequest.Statuses.PENDING
        )
        self.schedule_and_run_workflow(workflow)

        package_upload_signed_amd64.refresh_from_db()
        self.assertEqual(
            package_upload_signed_amd64.status, WorkRequest.Statuses.RUNNING
        )
        self.assertQuerySetEqual(
            package_upload_signed_amd64.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                # Work requests to sign and upload the source package.
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.PENDING,
                    None,
                    None,
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
                # Work requests to sign and upload the binary package.
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
            ],
        )

    def test_populate_upload_package_no_include_source_binaries(self) -> None:
        """Populate create upload_workflow without source neither binaries."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_upload": True,
                "vendor": "debian",
                "codename": "trixie",
                "upload_merge_uploads": False,
                "upload_since_version": "1.1",
                "upload_target": "ftp://user@example.org/pub/UploadQueue1/",
                "upload_include_source": False,
                "upload_include_binaries": False,
            }
        )

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

        self.assertEqual(
            package_upload.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "binary_artifacts": [],
                "codename": "trixie",
                "delayed_days": None,
                "key": None,
                "binary_key": None,
                "merge_uploads": False,
                "require_signature": True,
                "since_version": "1.1",
                "source_artifact": None,
                "target": "ftp://user@example.org/pub/UploadQueue1/",
                "target_distribution": None,
                "vendor": "debian",
            },
        )

        self.assertEqual(
            package_upload.workflow_data_json,
            {
                "display_name": "package upload (hello)",
                "step": "package_upload/hello",
            },
        )

    def test_populate_package_upload_delayed(self) -> None:
        """Populate an upload_package workflow for a delayed upload."""
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "architectures": ["amd64", "i386"],
                "enable_upload": True,
                "vendor": "debian",
                "codename": "trixie",
                "upload_merge_uploads": False,
                "upload_since_version": "1.1",
                "upload_target": "ftp://user@example.org/pub/UploadQueue1/",
                "upload_target_distribution": "debian:testing",
                "upload_delayed_days": 3,
            }
        )

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

        self.assertEqual(
            package_upload.task_data,
            {
                "arch_all_build_architecture": "amd64",
                "binary_artifacts": [
                    {
                        "collection": "internal@collections",
                        "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                        "name_matcher": {
                            "kind": CollectionItemMatcherKind.STARTSWITH,
                            "value": "build-",
                        },
                    }
                ],
                "codename": "trixie",
                "delayed_days": 3,
                "key": None,
                "binary_key": None,
                "merge_uploads": False,
                "require_signature": True,
                "since_version": "1.1",
                "source_artifact": self.source_artifact.id,
                "target": "ftp://user@example.org/pub/UploadQueue1/",
                "target_distribution": "debian:testing",
                "vendor": "debian",
            },
        )

        self.assertEqual(
            package_upload.workflow_data_json,
            {
                "display_name": "package upload (hello)",
                "step": "package_upload/hello",
            },
        )
        self.assertQuerySetEqual(package_upload.dependencies.all(), [])

        # PackageUpload.populate() was called and created its tasks.
        # MakeSourcePackageUpload only requires the source artifact and so
        # can start immediately.  Everything else is blocked.
        self.assertQuerySetEqual(
            package_upload.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.WORKER,
                    "makesourcepackageupload",
                    WorkRequest.Statuses.PENDING,
                    None,
                    None,
                ),
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "makesourcepackageupload",
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
            ]
            + [
                (
                    TaskTypes.WAIT,
                    "externaldebsign",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WORKER,
                    "sbuild",
                ),
                (
                    TaskTypes.SERVER,
                    "packageupload",
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.WAIT,
                    "externaldebsign",
                ),
            ]
            * 2,
        )

    def test_populate_package_upload_qa_failure_policy_fail(self) -> None:
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_autopkgtest": True,
                "qa_failure_policy": QAFailurePolicy.FAIL,
                "enable_upload": True,
            }
        )

        self.assertFalse(
            workflow.children.filter(
                task_type=TaskTypes.WAIT, task_name="confirm"
            ).exists()
        )

        qa = workflow.children.get(task_type=TaskTypes.WORKFLOW, task_name="qa")
        package_upload = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="package_upload"
        )
        self.assertQuerySetEqual(package_upload.dependencies.all(), [qa])

    def test_populate_package_upload_qa_failure_policy_confirm(self) -> None:
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_autopkgtest": True,
                "qa_failure_policy": QAFailurePolicy.CONFIRM,
                "enable_upload": True,
            }
        )

        qa = workflow.children.get(task_type=TaskTypes.WORKFLOW, task_name="qa")
        confirm = workflow.children.get(
            task_type=TaskTypes.WAIT, task_name="confirm"
        )
        self.assertEqual(
            confirm.task_data,
            {
                "auto_confirm_if_no_failure": True,
                "confirm_label": "Continue the workflow despite QA failures",
            },
        )
        self.assertEqual(
            confirm.workflow_data_json,
            {"display_name": "Confirm", "step": "confirm", "needs_input": True},
        )
        self.assertQuerySetEqual(confirm.dependencies.all(), [qa])

        package_upload = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="package_upload"
        )
        self.assertQuerySetEqual(package_upload.dependencies.all(), [confirm])

    def test_populate_package_upload_confirmation(self) -> None:
        """If confirming, package_upload depends on the confirm task."""
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "enable_confirmation": True,
                "enable_upload": True,
            }
        )

        confirm = workflow.children.get(
            task_type=TaskTypes.WAIT, task_name="confirm"
        )
        self.assertEqual(
            confirm.task_data,
            {
                "auto_confirm_if_no_failure": False,
                "confirm_label": "Continue the workflow",
            },
        )
        self.assertEqual(
            confirm.workflow_data_json,
            {"display_name": "Confirm", "step": "confirm", "needs_input": True},
        )
        self.assertQuerySetEqual(confirm.dependencies.all(), [])

        package_upload = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="package_upload"
        )
        self.assertQuerySetEqual(package_upload.dependencies.all(), [confirm])

    def test_populate_package_publish(self) -> None:
        """``populate`` creates a ``package_publish`` workflow."""
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        architectures = ["amd64", "i386"]
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "suite": f"sid@{CollectionCategory.SUITE}",
                "architectures": architectures,
            }
        )

        expected_binary_artifacts = {
            "collection": "internal@collections",
            "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
            "name_matcher": {
                "kind": CollectionItemMatcherKind.STARTSWITH,
                "value": "build-",
            },
        }
        sbuild_workflow = workflow.children.get(
            task_name="sbuild", task_type=TaskTypes.WORKFLOW
        )
        package_publish = workflow.children.get(
            task_name="package_publish", task_type=TaskTypes.WORKFLOW
        )
        self.assertEqual(
            package_publish.task_data,
            {
                "binary_artifacts": [expected_binary_artifacts],
                "replace": False,
                "source_artifact": self.source_artifact.id,
                "target_suite": f"sid@{CollectionCategory.SUITE}",
            },
        )
        self.assertEqual(
            package_publish.workflow_data_json,
            {
                "display_name": "publish packages (hello)",
                "step": "package_publish/hello",
            },
        )

        # The workflow can't orchestrate publication until binaries have
        # been built.
        self.assertEqual(package_publish.status, WorkRequest.Statuses.BLOCKED)
        self.assertQuerySetEqual(
            package_publish.dependencies.all(),
            # https://github.com/typeddjango/django-stubs/pull/3087
            sbuild_workflow.children.all(),  # type: ignore[arg-type]
        )
        self.assertQuerySetEqual(package_publish.children.all(), [])

        # Completing the builds causes the publishing workflow to be
        # orchestrated.
        binaries, _ = self.simulate_sbuild_workflow_completion(
            sbuild_workflow, use_data=False
        )
        package_publish.refresh_from_db()
        self.assertEqual(package_publish.status, WorkRequest.Statuses.PENDING)
        self.schedule_and_run_workflow(workflow)

        # PackagePublish.populate() was called and created its task.
        expected_copies = [
            {
                "source_items": [f"{self.source_artifact.id}@artifacts"],
                "target_collection": sid.id,
                "unembargo": False,
                "replace": False,
                "variables": {"component": "main", "section": "misc"},
            }
        ]
        for arch in architectures:
            expected_copies += [
                {
                    "source_items": [f"{binary_artifact.id}@artifacts"],
                    "target_collection": sid.id,
                    "unembargo": False,
                    "replace": False,
                    "variables": {
                        "component": "main",
                        "section": "misc",
                        "priority": "optional",
                    },
                }
                for binary_artifact in binaries[arch]
            ]
        self.assertQuerySetEqual(
            package_publish.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "task_data",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.SERVER,
                    "copycollectionitems",
                    {"copies": expected_copies},
                    WorkRequest.Statuses.PENDING,
                    None,
                    None,
                ),
                (
                    TaskTypes.INTERNAL,
                    "workflow",
                    {},
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.SERVER,
                    "copycollectionitems",
                ),
            ],
        )

    def test_populate_package_publish_signed_source(self) -> None:
        """``populate`` publishes signed source."""
        sid = self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        architectures = ["amd64", "i386"]
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "suite": f"sid@{CollectionCategory.SUITE}",
                "architectures": architectures,
                "enable_make_signed_source": True,
                "make_signed_source_purpose": KeyPurpose.UEFI,
                "signing_template_names": {
                    "amd64": ["hello-amd64-signed-template"]
                },
                "make_signed_source_key": "ABC123",
            }
        )

        initial_sbuild_workflow = workflow.children.get(
            task_name="sbuild", task_type=TaskTypes.WORKFLOW
        )
        make_signed_source = workflow.children.get(
            task_name="make_signed_source", task_type=TaskTypes.WORKFLOW
        )
        extract_for_signing = make_signed_source.children.get(
            task_name="extractforsigning", task_type=TaskTypes.WORKER
        )
        sign = make_signed_source.children.get(
            task_name="sign", task_type=TaskTypes.SIGNING
        )
        assemble_signed_source = make_signed_source.children.get(
            task_name="assemblesignedsource", task_type=TaskTypes.WORKER
        )
        package_publish_source, package_publish_signed_amd64 = (
            workflow.children.filter(
                task_name="package_publish", task_type=TaskTypes.WORKFLOW
            ).order_by("id")
        )
        expected_binary_artifacts = {
            "collection": "internal@collections",
            "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
            "name_matcher": {
                "kind": CollectionItemMatcherKind.STARTSWITH,
                "value": "build-",
            },
        }
        self.assertEqual(
            package_publish_source.task_data,
            {
                "binary_artifacts": [expected_binary_artifacts],
                "replace": False,
                "source_artifact": self.source_artifact.id,
                "target_suite": f"sid@{CollectionCategory.SUITE}",
            },
        )
        self.assertEqual(
            package_publish_source.workflow_data_json,
            {
                "display_name": "publish packages (hello)",
                "step": "package_publish/hello",
            },
        )

        # The workflow can't orchestrate publication until binaries have
        # been built.
        self.assertEqual(
            package_publish_source.status, WorkRequest.Statuses.BLOCKED
        )
        self.assertQuerySetEqual(
            package_publish_source.dependencies.all(),
            # https://github.com/typeddjango/django-stubs/pull/3087
            initial_sbuild_workflow.children.all(),  # type: ignore[arg-type]
        )
        self.assertQuerySetEqual(package_publish_source.children.all(), [])

        # Completing the builds causes the publishing workflow to be
        # orchestrated.
        binaries, _ = self.simulate_sbuild_workflow_completion(
            initial_sbuild_workflow, use_data=False
        )
        self.playground.create_signing_input_artifact(
            binary_package_name="hello-signed",
            workspace=extract_for_signing.workspace,
            work_request=extract_for_signing,
        )
        package_publish_source.refresh_from_db()
        self.assertEqual(
            package_publish_source.status, WorkRequest.Statuses.PENDING
        )
        self.schedule_and_run_workflow(workflow)

        # PackagePublish.populate() was called and created its task.
        expected_copies = [
            {
                "source_items": [f"{self.source_artifact.id}@artifacts"],
                "target_collection": sid.id,
                "unembargo": False,
                "replace": False,
                "variables": {"component": "main", "section": "misc"},
            }
        ]
        for arch in architectures:
            for binary_artifact in binaries[arch]:
                copy_item = {
                    "source_items": [f"{binary_artifact.id}@artifacts"],
                    "target_collection": sid.id,
                    "unembargo": False,
                    "replace": False,
                    "variables": {
                        "component": "main",
                        "section": "misc",
                        "priority": "optional",
                    },
                }
                expected_copies.append(copy_item)
        self.assertQuerySetEqual(
            package_publish_source.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "task_data",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.SERVER,
                    "copycollectionitems",
                    {"copies": expected_copies},
                    WorkRequest.Statuses.PENDING,
                    None,
                    None,
                ),
                (
                    TaskTypes.INTERNAL,
                    "workflow",
                    {},
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.SERVER,
                    "copycollectionitems",
                ),
            ],
        )

        expected_signed_source_artifact = (
            "internal@collections/name:"
            "signed-source-amd64-hello-amd64-signed-template"
        )
        expected_signed_binary_artifacts = {
            "collection": "internal@collections",
            "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
            "name_matcher": {
                "kind": CollectionItemMatcherKind.STARTSWITH,
                "value": (
                    "signed-source-amd64-hello-amd64-signed-template|build-"
                ),
            },
        }
        self.assertEqual(
            package_publish_signed_amd64.task_data,
            {
                "binary_artifacts": [expected_signed_binary_artifacts],
                "replace": False,
                "source_artifact": expected_signed_source_artifact,
                "target_suite": f"sid@{CollectionCategory.SUITE}",
            },
        )

        self.assertEqual(
            package_publish_signed_amd64.workflow_data_json,
            {
                "display_name": (
                    "publish packages (amd64/hello-amd64-signed-template)"
                ),
                "step": "package_publish/amd64/hello-amd64-signed-template",
            },
        )

        # The workflow can't orchestrate publishing the signed source
        # package and its binaries until the signed source package has been
        # assembled.
        package_publish_signed_amd64.refresh_from_db()
        self.assertEqual(
            package_publish_signed_amd64.status, WorkRequest.Statuses.BLOCKED
        )
        self.assertQuerySetEqual(
            package_publish_signed_amd64.dependencies.all(),
            [assemble_signed_source],
        )
        self.assertQuerySetEqual(
            package_publish_signed_amd64.children.all(), []
        )

        # Assembling the signed source package makes progress.
        extract_for_signing.refresh_from_db()
        self.playground.advance_work_request(
            extract_for_signing,
            result=WorkRequest.Results.SUCCESS,
            use_data=False,
        )
        self.playground.create_signing_output_artifact(
            purpose=KeyPurpose.UEFI,
            fingerprint="ABC123",
            binary_package_name="hello-signed",
            workspace=sign.workspace,
            work_request=sign,
        )
        sign.refresh_from_db()
        self.playground.advance_work_request(
            sign, result=WorkRequest.Results.SUCCESS, use_data=False
        )
        signed_source = self.playground.create_upload_artifacts(
            src_name="hello-signed-template",
            source=True,
            binary=False,
            section="admin",
            workspace=assemble_signed_source.workspace,
            work_request=assemble_signed_source,
        ).source
        assemble_signed_source.refresh_from_db()
        self.playground.advance_work_request(
            assemble_signed_source,
            result=WorkRequest.Results.SUCCESS,
            use_data=False,
        )
        package_publish_signed_amd64.refresh_from_db()
        self.assertEqual(
            package_publish_signed_amd64.status, WorkRequest.Statuses.PENDING
        )
        self.schedule_and_run_workflow(workflow)

        # The workflow needs the binaries as well.
        package_publish_signed_amd64.refresh_from_db()
        self.assertEqual(
            package_publish_signed_amd64.status, WorkRequest.Statuses.BLOCKED
        )
        sbuild_signed_amd64 = make_signed_source.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        )
        sbuild_signed_amd64_task = sbuild_signed_amd64.children.get(
            task_type=TaskTypes.WORKER, task_name="sbuild"
        )
        self.assertQuerySetEqual(
            package_publish_signed_amd64.dependencies.all(),
            [assemble_signed_source, sbuild_signed_amd64_task],
        )

        # Completing the build causes the signed publishing workflow to be
        # orchestrated.
        signed_binaries, _ = self.simulate_sbuild_workflow_completion(
            sbuild_signed_amd64,
            section="admin",
            priority="extra",
            use_data=False,
        )
        package_publish_signed_amd64.refresh_from_db()
        self.assertEqual(
            package_publish_signed_amd64.status, WorkRequest.Statuses.PENDING
        )
        self.schedule_and_run_workflow(workflow)

        expected_signed_copies = [
            {
                "source_items": [f"{signed_source.id}@artifacts"],
                "target_collection": sid.id,
                "unembargo": False,
                "replace": False,
                "variables": {"component": "main", "section": "admin"},
            },
            {
                "source_items": [f"{signed_binaries['amd64'][0].id}@artifacts"],
                "target_collection": sid.id,
                "unembargo": False,
                "replace": False,
                "variables": {
                    "component": "main",
                    "section": "admin",
                    "priority": "extra",
                },
            },
        ]
        self.assertQuerySetEqual(
            package_publish_signed_amd64.children.order_by("id").values_list(
                "task_type",
                "task_name",
                "task_data",
                "status",
                "dependencies__task_type",
                "dependencies__task_name",
            ),
            [
                (
                    TaskTypes.SERVER,
                    "copycollectionitems",
                    {"copies": expected_signed_copies},
                    WorkRequest.Statuses.PENDING,
                    None,
                    None,
                ),
                (
                    TaskTypes.INTERNAL,
                    "workflow",
                    {},
                    WorkRequest.Statuses.BLOCKED,
                    TaskTypes.SERVER,
                    "copycollectionitems",
                ),
            ],
        )

    def test_populate_package_publish_replace(self) -> None:
        """``populate`` can publish with replacement."""
        self.playground.create_collection(
            name="sid",
            category=CollectionCategory.SUITE,
            data={"may_reuse_versions": True},
        )
        workflow = self.orchestrate(
            extra_data={
                "source_artifact": self.source_artifact.id,
                "suite": f"sid@{CollectionCategory.SUITE}",
                "architectures": ["amd64"],
                "publish_replace": True,
            }
        )

        package_publish = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="package_publish"
        )
        self.assertTrue(package_publish.task_data["replace"])

        source_data = self.source_artifact.create_data()
        assert isinstance(source_data, DebianSourcePackage)
        sbuild_workflow = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        )
        self.simulate_sbuild_workflow_completion(
            sbuild_workflow, use_data=False
        )
        self.schedule_and_run_workflow(workflow)

        copy_collection_items = package_publish.children.get(
            task_type=TaskTypes.SERVER, task_name="copycollectionitems"
        )
        self.assertEqual(
            [
                copy["replace"]
                for copy in copy_collection_items.task_data["copies"]
            ],
            [True, True],
        )

    def test_populate_package_publish_qa_failure_policy_fail(self) -> None:
        self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "suite": f"sid@{CollectionCategory.SUITE}",
                "enable_autopkgtest": True,
                "qa_failure_policy": QAFailurePolicy.FAIL,
            }
        )

        self.assertFalse(
            workflow.children.filter(
                task_type=TaskTypes.WAIT, task_name="confirm"
            ).exists()
        )

        sbuild = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        )
        qa = workflow.children.get(task_type=TaskTypes.WORKFLOW, task_name="qa")
        package_publish = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="package_publish"
        )
        self.assertQuerySetEqual(
            package_publish.dependencies.all(),
            {*sbuild.children.all(), qa},
            ordered=False,
        )

    def test_populate_package_publish_qa_failure_policy_confirm(self) -> None:
        self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "suite": f"sid@{CollectionCategory.SUITE}",
                "enable_autopkgtest": True,
                "qa_failure_policy": QAFailurePolicy.CONFIRM,
            }
        )

        qa = workflow.children.get(task_type=TaskTypes.WORKFLOW, task_name="qa")
        confirm = workflow.children.get(
            task_type=TaskTypes.WAIT, task_name="confirm"
        )
        self.assertEqual(
            confirm.task_data,
            {
                "auto_confirm_if_no_failure": True,
                "confirm_label": "Continue the workflow despite QA failures",
            },
        )
        self.assertEqual(
            confirm.workflow_data_json,
            {"display_name": "Confirm", "step": "confirm", "needs_input": True},
        )
        self.assertQuerySetEqual(confirm.dependencies.all(), [qa])

        sbuild = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        )
        package_publish = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="package_publish"
        )
        self.assertQuerySetEqual(
            package_publish.dependencies.all(),
            {*sbuild.children.all(), confirm},
            ordered=False,
        )

    def test_populate_package_publish_confirmation(self) -> None:
        """If confirming, package_publish depends on the confirm task."""
        self.playground.create_collection(
            name="sid", category=CollectionCategory.SUITE
        )
        workflow = self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": self.source_artifact.id,
                "suite": f"sid@{CollectionCategory.SUITE}",
                "enable_confirmation": True,
            }
        )

        confirm = workflow.children.get(
            task_type=TaskTypes.WAIT, task_name="confirm"
        )
        self.assertEqual(
            confirm.task_data,
            {
                "auto_confirm_if_no_failure": False,
                "confirm_label": "Continue the workflow",
            },
        )
        self.assertEqual(
            confirm.workflow_data_json,
            {"display_name": "Confirm", "step": "confirm", "needs_input": True},
        )
        self.assertQuerySetEqual(confirm.dependencies.all(), [])

        sbuild = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="sbuild"
        )
        package_publish = workflow.children.get(
            task_type=TaskTypes.WORKFLOW, task_name="package_publish"
        )
        self.assertQuerySetEqual(
            package_publish.dependencies.all(),
            {*sbuild.children.all(), confirm},
            ordered=False,
        )

    def test_populate_with_debian_upload(self) -> None:
        """An upload as input will populate."""
        upload_artifacts = self.playground.create_upload_artifacts()
        self.orchestrate(
            extra_data={
                "architectures": ["amd64"],
                "source_artifact": upload_artifacts.upload.id,
            }
        )
