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

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

from debusine.artifacts.models import (
    ArtifactCategory,
    BareDataCategory,
    CollectionCategory,
    DebianUpload,
    DebusinePromise,
    TaskTypes,
)
from debusine.client.models import LookupChildType, RuntimeParameter
from debusine.db.models import (
    Artifact,
    Collection,
    CollectionItem,
    WorkRequest,
    WorkflowTemplate,
    default_workspace,
)
from debusine.server.workflows import (
    PackageUploadWorkflow,
    WorkflowValidationError,
)
from debusine.server.workflows.base import orchestrate_workflow
from debusine.server.workflows.models import (
    PackageUploadWorkflowDynamicData,
    WorkRequestWorkflowData,
)
from debusine.server.workflows.tests.helpers import WorkflowTestCase
from debusine.tasks.models import (
    ActionUpdateCollectionWithArtifacts,
    InputArtifactMultiple,
    InputArtifactSingle,
    LookupMultiple,
    SbuildData,
    SbuildInput,
)


class PackageUploadWorkflowTests(WorkflowTestCase[PackageUploadWorkflow]):
    """Unit tests for :py:class:`PackageUploadWorkflow`."""

    SAMPLE_TASK_DATA = {
        "target": "sftp://upload.example.org/queue/",
        "vendor": "debian",
        "codename": "trixie",
        "require_signature": False,
    }

    source_artifact: ClassVar[Artifact]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.source_artifact = (
            cls.playground.create_minimal_source_package_artifact()
        )

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

    def create_source_binary_artifacts(
        self, architectures: Sequence[str] | None = None
    ) -> tuple[Artifact, list[Artifact], Collection]:
        """Return one source and a list of binary packages (one per arch)."""
        if architectures is None:
            architectures = ("amd64",)
        collection = self.playground.create_collection(
            "test", CollectionCategory.TEST
        )

        source = self.playground.create_minimal_source_package_artifact()
        uploads = []
        for architecture in architectures:
            upload = self.playground.create_upload_artifacts(
                source=False,
                binary=True,
                binaries=[("hello", architecture)],
            )
            uploads.append(upload.upload)
            for binary in upload.binaries:
                CollectionItem.objects.create_from_artifact(
                    binary,
                    parent_collection=collection,
                    name=f"binary-{binary.data['deb_fields']['Architecture']}",
                    created_by_user=self.playground.get_default_user(),
                    data={},
                    created_by_workflow=None,
                )

        return source, uploads, collection

    def test_validate_input_source_binary_artifacts_not_set(self) -> None:
        """Invalid: source_artifact and binaries_artifacts not set."""
        del self.task_data["source_artifact"]
        self.validate = False
        with self.assertRaisesRegex(
            WorkflowValidationError,
            '"source_artifact" or "binary_artifacts" must be set',
        ):
            self.task.validate_input()

    def test_validate_input_source_package_vendor_codename_not_set(
        self,
    ) -> None:
        """Raise WorkflowValidationError: vendor and codename required."""
        self.validate = False
        del self.task_data["vendor"]
        del self.task_data["codename"]

        with self.assertRaisesRegex(
            WorkflowValidationError,
            '"vendor" and "codename" are required when '
            'source artifact category is debian:source-package',
        ):
            self.task.validate_input()

    def test_create_orchestrator(self) -> None:
        """A PackageUploadWorkflow can be instantiated."""
        source_artifact, [binary_artifact], _ = (
            self.create_source_binary_artifacts(["arm64"])
        )
        target_distribution = "debian:bookworm"
        self.task_data.update(
            {
                "source_artifact": source_artifact.id,
                "binary_artifacts": [binary_artifact.id],
                "target_distribution": target_distribution,
            }
        )
        w = self.task
        self.assertEqual(w.data.source_artifact, source_artifact.id)
        self.assertEqual(
            w.data.binary_artifacts, LookupMultiple((binary_artifact.id,))
        )

        self.assertEqual(w.data.target_distribution, target_distribution)

    def orchestrate(self, *, data: dict[str, Any]) -> WorkRequest:
        """Create a PackageUpload workflow and call orchestrate_workflow."""
        template = WorkflowTemplate.objects.create(
            name="package_upload",
            workspace=default_workspace(),
            task_name="package_upload",
            runtime_parameters=RuntimeParameter.ANY,
        )

        wr = self.playground.create_workflow(
            template,
            task_data={"target": "sftp://upload.example.org/queue", **data},
        )
        self.assertTrue(orchestrate_workflow(wr))

        return wr

    def test_populate_makesourcepackageupload(self) -> None:
        """
        Workflow create two children work request.

        Work requests created:
        -makesourcepackageupload
        -package_upload
        """
        source = self.playground.create_minimal_source_package_artifact()

        with patch.object(
            PackageUploadWorkflow,
            "requires_artifact",
            wraps=PackageUploadWorkflow.requires_artifact,
        ) as requires_artifact:
            wr = self.orchestrate(
                data={
                    "source_artifact": source.id,
                    "target_distribution": "debian:bookworm",
                    "since_version": "1.0",
                    "require_signature": False,
                    "vendor": "debian",
                    "codename": "bullseye",
                }
            )

        self.assertEqual(wr.children.count(), 2)

        [make_source] = wr.children.filter(task_name="makesourcepackageupload")

        self.assertEqual(
            make_source.workflow_data_json,
            {
                "display_name": "Make .changes file",
                "step": "source-package-upload",
            },
        )

        self.assertEqual(
            make_source.task_data,
            {
                "build_architecture": "amd64",
                "environment": "debian/match:codename=bullseye",
                "input": {"source_artifact": source.id},
                "since_version": "1.0",
                "target_distribution": "debian:bookworm",
            },
        )

        self.assertEqual(make_source.parent, wr)

        self.assert_work_request_event_reactions(
            make_source,
            on_success=[
                ActionUpdateCollectionWithArtifacts(
                    artifact_filters={"category": "debian:upload"},
                    collection="internal@collections",
                    name_template="package-upload-source",
                )
            ],
        )

        # Assert that self.requires_artifact() was called.
        # It cannot test the dependencies because source_artifact was
        # a real artifact and not a promise (so no dependencies were added)
        self.assertIn(
            call(make_source, source.id), requires_artifact.mock_calls
        )

        [package_upload] = wr.children.filter(task_name="packageupload")
        self.assertEqual(package_upload.task_type, TaskTypes.SERVER)
        self.assertEqual(
            package_upload.workflow_data_json,
            {
                "display_name": (
                    "Package upload internal@collections/"
                    "name:package-upload-source"
                ),
                "step": (
                    "package-upload-internal@collections/"
                    "name:package-upload-source"
                ),
            },
        )
        self.assertEqual(
            package_upload.task_data,
            {
                "input": {
                    "upload": "internal@collections/name:package-upload-source"
                },
                "target": "sftp://upload.example.org/queue",
                "delayed_days": None,
            },
        )

        self.assertQuerySetEqual(
            package_upload.dependencies.all(),
            list(
                WorkRequest.objects.filter(task_name="makesourcepackageupload")
            ),
        )

    def test_populate_source_is_package_upload(self) -> None:
        """Category source package is UPLOAD: skip makesourcepackageupload."""
        source = self.playground.create_upload_artifacts().upload

        wr = self.orchestrate(
            data={
                "source_artifact": source.id,
                "target_distribution": "debian:bookworm",
                "since_version": "1.0",
                "require_signature": False,
            }
        )

        self.assertEqual(wr.children.count(), 1)

        [package_upload] = wr.children.filter(task_name="packageupload")
        self.assertEqual(
            package_upload.workflow_data_json,
            {
                "display_name": f"Package upload {source.id}",
                "step": f"package-upload-{source.id}",
            },
        )
        self.assertEqual(
            package_upload.task_data,
            {
                "input": {"upload": source.id},
                "target": "sftp://upload.example.org/queue",
                "delayed_days": None,
            },
        )

    def test_populate_package_uploads_binaries_only(self) -> None:
        """The workflow create children package uploads: binaries only."""
        _, [binary], _ = self.create_source_binary_artifacts()

        wr = self.orchestrate(
            data={
                "binary_artifacts": [binary.id],
                "target_distribution": "debian:bookworm",
                "require_signature": False,
            },
        )

        self.assertEqual(
            wr.children.filter(task_name="packageupload").count(), 1
        )

        # Other properties of the created package is tested on different
        # unit tests

    def test_populate_merge_uploads(self) -> None:
        """Test mergeuploads task is created (merge_uploads=True)."""
        parent = self.playground.create_workflow()
        assert parent.internal_collection is not None

        architectures = ("amd64", "i386")
        source = self.playground.create_minimal_source_package_artifact()
        for architecture in architectures:
            sbuild = parent.create_child_worker(
                task_name="sbuild",
                task_data=SbuildData(
                    input=SbuildInput(source_artifact=source.id),
                    build_architecture=architecture,
                    environment="debian/match:codename=bookworm",
                ),
                workflow_data=WorkRequestWorkflowData(
                    display_name=architecture, step=architecture
                ),
            )
            self.playground.create_bare_data_item(
                parent.internal_collection,
                f"build-{architecture}",
                category=BareDataCategory.PROMISE,
                data=DebusinePromise(
                    promise_work_request_id=sbuild.id,
                    promise_workflow_id=parent.id,
                    promise_category=ArtifactCategory.UPLOAD,
                ),
            )

        wr = self.playground.create_workflow(
            task_name="package_upload",
            task_data={
                "source_artifact": source.id,
                "binary_artifacts": {
                    "collection": "internal@collections",
                    "child_type": LookupChildType.ARTIFACT_OR_PROMISE,
                    "name__startswith": "build-",
                },
                "require_signature": False,
                "target": "sftp://upload.example.org/queue",
                "target_distribution": "debian/match:codename=bookworm",
                "merge_uploads": True,
                "vendor": "debian",
                "codename": "bullseye",
            },
            parent=parent,
        )
        self.assertTrue(orchestrate_workflow(wr))

        self.assertEqual(
            wr.children.filter(task_name="mergeuploads").count(), 1
        )

        [merge_uploads] = wr.children.filter(task_name="mergeuploads")

        self.assertEqual(
            merge_uploads.task_data,
            {
                "input": {
                    "uploads": [
                        "internal@collections/name:package-upload-source"
                    ]
                    + [
                        f"internal@collections/name:build-{architecture}"
                        for architecture in architectures
                    ]
                },
            },
        )

        self.assertQuerySetEqual(
            merge_uploads.dependencies.all(),
            list(
                WorkRequest.objects.filter(
                    task_name__in={"makesourcepackageupload", "sbuild"}
                )
            ),
            ordered=False,
        )

        self.assertEqual(
            wr.children.filter(task_name="packageupload").count(), 1
        )
        [package_upload] = wr.children.filter(task_name="packageupload")
        self.assertEqual(
            package_upload.task_data,
            {
                "input": {
                    "upload": (
                        f"internal@collections/name:"
                        f"package-upload-merged-{wr.id}"
                    )
                },
                "target": "sftp://upload.example.org/queue",
                "delayed_days": None,
            },
        )

        # Replace one of the promises with an artifact and run the workflow
        # orchestrator again.  The result is stable: no new MergeUploads
        # work request is created.
        binary, _ = self.playground.create_artifact(
            category=ArtifactCategory.UPLOAD,
            data=DebianUpload(
                type="dpkg",
                changes_fields={
                    "Architecture": architectures[0],
                    "Files": [{"name": f"hello_1.0_{architectures[0]}.deb"}],
                },
            ),
        )
        parent.internal_collection.manager.add_artifact(
            binary,
            user=self.playground.get_default_user(),
            workflow=parent,
            name=f"build-{architectures[0]}",
            replace=True,
        )
        self.assertTrue(orchestrate_workflow(wr))

        self.assertEqual(
            wr.children.filter(task_name="mergeuploads").count(), 1
        )

    def test_upload_debsign(self) -> None:
        """Upload signed artifacts."""
        source, [binary], _ = self.create_source_binary_artifacts()

        wr = self.orchestrate(
            data={
                "source_artifact": source.id,
                "binary_artifacts": [binary.id],
                "require_signature": False,
                "target_distribution": "debian/match:codename=bookworm",
                "merge_uploads": True,
                "key": "ABC123",
                "vendor": "debian",
                "codename": "bookworm",
            },
        )

        [signer] = wr.children.filter(
            task_type=TaskTypes.SIGNING, task_name="debsign"
        )
        self.assertEqual(
            signer.task_data,
            {
                "key": "ABC123",
                "unsigned": (
                    f"internal@collections/name:package-upload-merged-{wr.id}"
                ),
            },
        )
        identifier = f"internal@collections/name:package-upload-merged-{wr.id}"
        self.assertEqual(
            signer.workflow_data_json,
            {
                "display_name": f"Sign upload for {identifier}",
                "step": f"debsign-{identifier}",
            },
        )

        self.assertQuerySetEqual(
            signer.dependencies.all(),
            list(
                WorkRequest.objects.filter(
                    task_type=TaskTypes.WORKER, task_name="mergeuploads"
                )
            ),
        )
        signed_name = (
            f"package-upload-signed-"
            f"internal_collections_name_package-upload-merged-{wr.id}"
        )
        self.assert_work_request_event_reactions(
            signer,
            on_success=[
                ActionUpdateCollectionWithArtifacts(
                    artifact_filters={"category": "debian:upload"},
                    collection="internal@collections",
                    created_at=None,
                    name_template=signed_name,
                    variables=None,
                )
            ],
        )

        [uploader] = wr.children.filter(task_name="packageupload")
        self.assertEqual(
            uploader.task_data,
            {
                "input": {"upload": f"internal@collections/name:{signed_name}"},
                "target": "sftp://upload.example.org/queue",
                "delayed_days": None,
            },
        )

        self.assertEqual(
            uploader.workflow_data_json,
            {
                "display_name": (
                    f"Package upload "
                    f"internal@collections/name:package-upload-merged-{wr.id}"
                ),
                "step": (
                    f"package-upload-internal@collections/"
                    f"name:package-upload-merged-{wr.id}"
                ),
            },
        )

    def test_upload_externaldebsign(self) -> None:
        """Upload external signed artifacts."""
        source, [binary], _ = self.create_source_binary_artifacts()

        wr = self.orchestrate(
            data={
                "source_artifact": source.id,
                "binary_artifacts": [binary.id],
                "target_distribution": "debian/match:codename=bookworm",
                "merge_uploads": True,
                "require_signature": True,
                "vendor": "debian",
                "codename": "bookworm",
            },
        )

        [signer] = wr.children.filter(
            task_type=TaskTypes.WAIT, task_name="externaldebsign"
        )
        self.assertEqual(
            signer.task_data,
            {
                "unsigned": (
                    f"internal@collections/name:package-upload-merged-{wr.id}"
                )
            },
        )
        identifier = f"internal@collections/name:package-upload-merged-{wr.id}"
        self.assertEqual(
            signer.workflow_data_json,
            {
                "display_name": (
                    f"Wait for signature on upload for {identifier}"
                ),
                "step": f"external-debsign-{identifier}",
                "needs_input": True,
            },
        )

        self.assertQuerySetEqual(
            signer.dependencies.all(),
            list(
                WorkRequest.objects.filter(
                    task_type=TaskTypes.WORKER, task_name="mergeuploads"
                )
            ),
        )
        signed_name = (
            f"package-upload-signed-"
            f"internal_collections_name_package-upload-merged-{wr.id}"
        )
        self.assert_work_request_event_reactions(
            signer,
            on_success=[
                ActionUpdateCollectionWithArtifacts(
                    artifact_filters={"category": "debian:upload"},
                    collection="internal@collections",
                    created_at=None,
                    name_template=signed_name,
                    variables=None,
                )
            ],
        )

        [uploader] = wr.children.filter(task_name="packageupload")
        self.assertEqual(
            uploader.task_data,
            {
                "input": {"upload": f"internal@collections/name:{signed_name}"},
                "target": "sftp://upload.example.org/queue",
                "delayed_days": None,
            },
        )

    def test_upload_split_signing(self) -> None:
        """Upload artifacts: developer signs source, Debusine signs binary."""
        source, [binary], _ = self.create_source_binary_artifacts()

        wr = self.orchestrate(
            data={
                "source_artifact": source.id,
                "binary_artifacts": [binary.id],
                "require_signature": True,
                "target_distribution": "debian/match:codename=bookworm",
                "binary_key": "ABC123",
                "vendor": "debian",
                "codename": "bookworm",
            },
        )

        [source_signer] = wr.children.filter(
            task_type=TaskTypes.WAIT, task_name="externaldebsign"
        )
        source_lookup = "internal@collections/name:package-upload-source"
        self.assertEqual(source_signer.task_data, {"unsigned": source_lookup})
        self.assertEqual(
            source_signer.workflow_data_json,
            {
                "display_name": (
                    f"Wait for signature on upload for {source_lookup}"
                ),
                "step": f"external-debsign-{source_lookup}",
                "needs_input": True,
            },
        )
        self.assertQuerySetEqual(
            source_signer.dependencies.all(),
            list(
                WorkRequest.objects.filter(
                    task_type=TaskTypes.WORKER,
                    task_name="makesourcepackageupload",
                )
            ),
        )
        source_signed_name = (
            "package-upload-signed-"
            "internal_collections_name_package-upload-source"
        )
        self.assert_work_request_event_reactions(
            source_signer,
            on_success=[
                ActionUpdateCollectionWithArtifacts(
                    artifact_filters={"category": "debian:upload"},
                    collection="internal@collections",
                    name_template=source_signed_name,
                )
            ],
        )

        [binary_signer] = wr.children.filter(
            task_type=TaskTypes.SIGNING, task_name="debsign"
        )
        self.assertEqual(
            binary_signer.task_data,
            {"key": "ABC123", "unsigned": binary.id},
        )
        self.assertEqual(
            binary_signer.workflow_data_json,
            {
                "display_name": f"Sign upload for {binary.id}",
                "step": f"debsign-{binary.id}",
            },
        )
        self.assertQuerySetEqual(
            binary_signer.dependencies.all(), [source_signer]
        )
        binary_signed_name = f"package-upload-signed-{binary.id}"
        self.assert_work_request_event_reactions(
            binary_signer,
            on_success=[
                ActionUpdateCollectionWithArtifacts(
                    artifact_filters={"category": "debian:upload"},
                    collection="internal@collections",
                    created_at=None,
                    name_template=binary_signed_name,
                    variables=None,
                )
            ],
        )

        [source_uploader, binary_uploader] = wr.children.filter(
            task_name="packageupload"
        ).order_by("id")
        self.assertEqual(
            source_uploader.task_data,
            {
                "input": {
                    "upload": f"internal@collections/name:{source_signed_name}"
                },
                "target": "sftp://upload.example.org/queue",
                "delayed_days": None,
            },
        )
        self.assertEqual(
            source_uploader.workflow_data_json,
            {
                "display_name": f"Package upload {source_lookup}",
                "step": f"package-upload-{source_lookup}",
            },
        )
        self.assertQuerySetEqual(
            source_uploader.dependencies.all(), [source_signer]
        )
        self.assertEqual(
            binary_uploader.task_data,
            {
                "input": {
                    "upload": f"internal@collections/name:{binary_signed_name}"
                },
                "target": "sftp://upload.example.org/queue",
                "delayed_days": None,
            },
        )
        self.assertEqual(
            binary_uploader.workflow_data_json,
            {
                "display_name": f"Package upload {binary.id}",
                "step": f"package-upload-{binary.id}",
            },
        )
        self.assertQuerySetEqual(
            binary_uploader.dependencies.all(),
            [binary_signer, source_uploader],
            ordered=False,
        )

    def test_upload_delayed(self) -> None:
        """Upload to a delayed queue."""
        source = self.playground.create_minimal_source_package_artifact()

        wr = self.orchestrate(
            data={
                "source_artifact": source.id,
                "delayed_days": 3,
                "vendor": "debian",
                "codename": "bookworm",
            }
        )

        signed_name = (
            "package-upload-signed-"
            "internal_collections_name_package-upload-source"
        )
        uploader = wr.children.filter(task_name="packageupload").get()
        self.assertEqual(
            uploader.task_data,
            {
                "input": {"upload": f"internal@collections/name:{signed_name}"},
                "target": "sftp://upload.example.org/queue",
                "delayed_days": 3,
            },
        )

    def test_orchestrate_idempotent(self) -> None:
        """Calling orchestrate twice does not create new work requests."""
        self.playground.create_collection(
            "test", CollectionCategory.TASK_CONFIGURATION
        )
        source, _, binaries_collection = self.create_source_binary_artifacts(
            architectures=("amd64", "i386")
        )

        # Force a predictable default ordering.
        with patch(
            "debusine.db.models.collections.CollectionItem._meta.ordering",
            ["id"],
        ):
            wr = self.orchestrate(
                data={
                    "task_configuration": (
                        f"test@{CollectionCategory.TASK_CONFIGURATION}"
                    ),
                    "source_artifact": source.id,
                    "binary_artifacts": {
                        "collection": binaries_collection.id,
                        "name__startswith": "binary-",
                    },
                    "vendor": "debian",
                    "codename": "bookworm",
                    "merge_uploads": True,
                },
            )

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

        # Force a different predictable default ordering.
        with patch(
            "debusine.db.models.collections.CollectionItem._meta.ordering",
            ["-id"],
        ):
            self.get_workflow(wr).populate()

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

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

    def test_compute_dynamic_data_source_artifact(self) -> None:
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            PackageUploadWorkflowDynamicData(
                subject="hello",
                parameter_summary="hello → sftp://upload.example.org/queue/",
                source_artifact_id=self.source_artifact.id,
                binary_artifacts_ids=[],
            ),
        )

    def test_compute_dynamic_data_source_is_promise(self) -> None:
        parent = self.playground.create_workflow()
        assert parent.internal_collection is not None

        signing = parent.create_child_worker(
            task_name="noop",
        )
        self.playground.create_bare_data_item(
            parent.internal_collection,
            "source",
            category=BareDataCategory.PROMISE,
            data=DebusinePromise(
                promise_work_request_id=signing.id,
                promise_workflow_id=parent.id,
                promise_category=ArtifactCategory.SOURCE_PACKAGE,
                **{"source_package_name": "hello"},  # noqa: PIE804
            ),
        )

        self.task_data["source_artifact"] = "internal@collections/name:source"
        self.parent = parent

        self.assertEqual(
            self.compute_dynamic_data(self.task),
            PackageUploadWorkflowDynamicData(
                parameter_summary="hello → sftp://upload.example.org/queue/",
                subject="hello",
                binary_artifacts_ids=[],
            ),
        )

    def test_compute_dynamic_data_binary_artifacts(self) -> None:
        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello"
            )
        )

        del self.task_data["source_artifact"]
        self.task_data["binary_artifacts"] = [binary_artifact.id]
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            PackageUploadWorkflowDynamicData(
                subject="hello",
                parameter_summary="hello → sftp://upload.example.org/queue/",
                source_artifact_id=None,
                binary_artifacts_ids=[binary_artifact.id],
            ),
        )

    def test_get_input_artifacts_source_artifact(self) -> None:
        workflow = self.task
        workflow.dynamic_data = workflow.compute_dynamic_data()
        assert workflow.data.source_artifact is not None
        self.assertEqual(
            workflow.get_input_artifacts(),
            [
                InputArtifactSingle(
                    lookup=workflow.data.source_artifact,
                    label="source_artifact",
                    artifact_id=self.source_artifact.id,
                ),
                InputArtifactMultiple.from_ids(
                    lookup=workflow.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=[],
                ),
            ],
        )

    def test_get_input_artifacts_binary_artifacts(self) -> None:
        binary_artifact = (
            self.playground.create_minimal_binary_package_artifact(
                srcpkg_name="hello"
            )
        )
        del self.task_data["source_artifact"]
        self.task_data["binary_artifacts"] = [binary_artifact.id]

        workflow = self.task
        workflow.dynamic_data = workflow.compute_dynamic_data()
        self.assertEqual(
            workflow.get_input_artifacts(),
            [
                InputArtifactMultiple.from_ids(
                    lookup=workflow.data.binary_artifacts,
                    label="binary_artifacts",
                    artifact_ids=[binary_artifact.id],
                )
            ],
        )
