# 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 piuparts task support on the worker."""

import datetime as dt
import shlex
import tarfile
import textwrap
from collections.abc import Iterable
from io import BytesIO
from pathlib import Path
from typing import override
from unittest import mock
from unittest.mock import MagicMock, call

from debian.debian_support import Version

import debusine.worker.tags as wtags
from debusine.artifacts import PiupartsArtifact
from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianPiuparts,
    DebianSystemTarball,
    DebianUpload,
    WorkRequestResults,
)
from debusine.client.models import (
    ArtifactResponse,
    FileResponse,
    FilesResponseType,
    RelationType,
)
from debusine.tasks import Piuparts, TaskConfigError
from debusine.tasks.models import (
    BackendType,
    BaseTaskInputArtifact,
    ExtraExternalRepository,
    PiupartsDynamicData,
)
from debusine.tasks.tests.helpers import ExternalTaskHelperMixin
from debusine.test import TestCase
from debusine.test.test_utils import (
    create_artifact_response,
    create_remote_artifact,
    create_system_tarball_data,
)


class PiupartsTests(ExternalTaskHelperMixin[Piuparts], TestCase):
    """Test the Piuparts class."""

    SAMPLE_TASK_DATA = {
        "input": {"binary_artifacts": [421]},
        "build_architecture": "amd64",
        "environment": "debian/match:codename=bookworm",
        "base_tgz": 44,
    }

    _binary_upload_data = DebianUpload(
        type="dpkg",
        changes_fields={
            "Architecture": "amd64",
            "Version": "1.0",
            "Files": [{"name": "hello_1.0_amd64.deb"}],
            "Source": "hello",
        },
    )

    @override
    def setUp(self) -> None:
        super().setUp()
        self.resolver.set(extra_repositories=None)
        self.resolver.set_field_artifact(
            "environment",
            id=1,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(),
        )
        self.resolver.set_field_artifact(
            "base_tgz",
            id=44,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(),
        )
        self.resolver.set_field_artifacts(
            "input_binary_artifacts",
            BaseTaskInputArtifact(
                artifact_id=421,
                category=ArtifactCategory.UPLOAD,
                data=self._binary_upload_data,
            ),
        )

    def test_configure_fails_with_missing_required_data(
        self,
    ) -> None:
        with self.assertRaises(TaskConfigError):
            self.configure_task(override={"input": {"artifact": 1}})

    def test_configure_environment_is_required(self) -> None:
        """Missing required field "environment": raise exception."""
        with self.assertRaises(TaskConfigError):
            self.configure_task(remove=["environment"])

    def test_configure_base_tgz_is_required(self) -> None:
        """Missing required field "base_tgz": raise exception."""
        with self.assertRaises(TaskConfigError):
            self.configure_task(remove=["base_tgz"])

    def test_get_provided_worker_tags(self) -> None:
        self.assertCountEqual(
            self.task.get_provided_worker_tags(),
            [
                wtags.TASK_PREFIX + "worker:piuparts:version:1",
            ],
        )

    def test_compute_scheduler_tags(self) -> None:
        self.assertComputesSchedulerTags(
            require_system={
                wtags.BUILD_ARCH_PREFIX + "amd64",
                wtags.EXECUTOR_PREFIX + "unshare",
            }
        )

    def test_compute_dynamic_data(self) -> None:
        """Dynamic data receives relevant artifact IDs."""
        self.assertEqual(
            self.compute_dynamic_data(self.task),
            PiupartsDynamicData(
                environment_id=1,
                input_binary_artifacts_ids=[421],
                base_tgz_id=44,
                subject="hello",
                parameter_summary="hello_1.0 (amd64)",
                runtime_context="sid:amd64",
                configuration_context="sid",
            ),
        )

    def test_get_subject(self) -> None:
        self.assertEqual(self.get_subject(), "hello")

    def test_get_subject_multiple_binaries_same_source(self) -> None:
        self.configure_task(
            override={"input": {"binary_artifacts": [421, 422, 423]}}
        )

        self.resolver.set_field_artifact(
            "environment",
            id=44,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(),
        )

        self.resolver.set_field_artifact(
            "base_tgz",
            id=45,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(codename="bookworm"),
        )

        binary_1 = DebianUpload(
            type="dpkg",
            changes_fields={
                "Architecture": "amd64",
                "Version": "1.0",
                "Files": [{"name": "linux-cpupower_1.0_amd64.deb"}],
                "Source": "linux",
            },
        )

        binary_2 = DebianUpload(
            type="dpkg",
            changes_fields={
                "Architecture": "amd64",
                "Version": "1.0",
                "Files": [{"name": "linux-image-1.0_amd64.deb"}],
                "Source": "linux",
            },
        )

        binary_3 = DebianUpload(
            type="dpkg",
            changes_fields={
                "Architecture": "amd64",
                "Version": "1.0",
                "Files": [{"name": "hello-1.0.0_amd64.deb"}],
                "Source": "hello",
            },
        )

        self.resolver.set_field_artifacts(
            "input_binary_artifacts",
            BaseTaskInputArtifact(
                artifact_id=421,
                category=ArtifactCategory.UPLOAD,
                data=binary_1,
            ),
            BaseTaskInputArtifact(
                artifact_id=422,
                category=ArtifactCategory.UPLOAD,
                data=binary_2,
            ),
            BaseTaskInputArtifact(
                artifact_id=423,
                category=ArtifactCategory.UPLOAD,
                data=binary_3,
            ),
        )

        self.assertEqual(self.get_subject(), "hello linux")

    def test_get_configuration_context(self) -> None:
        self.assertEqual(self.get_configuration_context(), "sid")

    def test_compute_dynamic_data_base_tgz_constraints(self) -> None:
        """Dynamic data applies environment-like constraints to base_tgz."""
        self.configure_task(
            override={"base_tgz": "debian/match:codename=bookworm"}
        )
        self.resolver.set_field_artifact(
            "environment",
            id=1,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(),
        )
        self.resolver.set_field_artifact(
            "base_tgz",
            id=44,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(codename="bookworm"),
        )

        self.assertEqual(
            self.compute_dynamic_data(self.task),
            PiupartsDynamicData(
                environment_id=1,
                input_binary_artifacts_ids=[421],
                base_tgz_id=44,
                subject="hello",
                parameter_summary="hello_1.0 (amd64)",
                runtime_context="bookworm:amd64",
                configuration_context="bookworm",
            ),
        )

    def test_compute_subject_parameter_summary_multiple_binaries_same_source(
        self,
    ) -> None:
        self.configure_task(
            override={"input": {"binary_artifacts": [421, 422]}}
        )
        self.resolver.set_field_artifact(
            "environment",
            id=44,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(),
        )

        self.resolver.set_field_artifact(
            "base_tgz",
            id=45,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(codename="bookworm"),
        )

        binary_1 = DebianUpload(
            type="dpkg",
            changes_fields={
                "Architecture": "amd64",
                "Version": "1.0",
                "Files": [{"name": "linux-cpupower_1.0_amd64.deb"}],
                "Source": "linux",
            },
        )

        binary_2 = DebianUpload(
            type="dpkg",
            changes_fields={
                "Architecture": "amd64",
                "Version": "1.0",
                "Files": [{"name": "linux-image-1.0_amd64.deb"}],
                "Source": "linux",
            },
        )

        self.resolver.set_field_artifacts(
            "input_binary_artifacts",
            BaseTaskInputArtifact(
                artifact_id=421,
                category=ArtifactCategory.UPLOAD,
                data=binary_1,
            ),
            BaseTaskInputArtifact(
                artifact_id=422,
                category=ArtifactCategory.UPLOAD,
                data=binary_2,
            ),
        )

        dynamic_data = self.compute_dynamic_data(self.task)

        self.assertEqual(dynamic_data.subject, "linux")
        self.assertEqual(dynamic_data.parameter_summary, "linux_1.0 (amd64)")

    def test_compute_dynamic_data_multiple_source_packages(self) -> None:
        self.configure_task(
            override={"input": {"binary_artifacts": [421, 422]}}
        )

        self.resolver.set_field_artifact(
            "environment",
            id=44,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(),
        )
        self.resolver.set_field_artifact(
            "base_tgz",
            id=45,
            category=ArtifactCategory.SYSTEM_TARBALL,
            data=create_system_tarball_data(codename="bookworm"),
        )
        self.resolver.set_field_artifacts(
            "input_binary_artifacts",
            BaseTaskInputArtifact(
                artifact_id=421,
                category=ArtifactCategory.UPLOAD,
                data=self._binary_upload_data,
            ),
            BaseTaskInputArtifact(
                artifact_id=422,
                category=ArtifactCategory.UPLOAD,
                data=DebianUpload(
                    type="dpkg",
                    changes_fields={
                        "Architecture": "amd64",
                        "Version": "2.0",
                        "Files": [{"name": "bye_1.0-1_amd64.deb"}],
                        "Source": "bye",
                    },
                ),
            ),
        )
        dynamic_data = self.compute_dynamic_data(self.task)

        self.assertEqual(dynamic_data.subject, "bye hello")
        self.assertEqual(dynamic_data.parameter_summary, "bye hello (amd64)")

    def test_acceptable_input_categories(self) -> None:
        self.assertEqual(
            Piuparts.inputs["base_tgz"].categories,  # type: ignore[attr-defined]
            (ArtifactCategory.SYSTEM_TARBALL,),
        )
        self.assertEqual(
            Piuparts.inputs["input_binary_artifacts"].categories,  # type: ignore[attr-defined]
            (
                ArtifactCategory.BINARY_PACKAGE,
                ArtifactCategory.UPLOAD,
            ),
        )

    def test_get_input_artifacts_ids(self) -> None:
        """Test get_input_artifacts_ids."""
        self.task.dynamic_data = None
        self.assertEqual(self.task.get_input_artifacts_ids(), [])

        self.task.dynamic_data = PiupartsDynamicData(
            environment_id=1,
            base_tgz_id=2,
            input_binary_artifacts_ids=[3, 4],
        )
        self.assertEqual(self.task.get_input_artifacts_ids(), [1, 2, 3, 4])

    def test_cmdline_as_root(self) -> None:
        """_cmdline_as_root() return True."""
        self.assertTrue(self.task._cmdline_as_root())

    def create_basetgz_tarfile(
        self,
        destination: Path,
        directories: Iterable[Path] = (),
        symlinks: Iterable[tuple[Path, Path]] = (),
        empty_files: Iterable[Path] = (),
    ) -> None:
        """Create a tar file that would make a useful test for base_tgz."""
        assert destination.suffix == ".xz"
        with tarfile.open(destination, "w:xz") as tar:
            tarinfo = tarfile.TarInfo(name="./foo")
            tarinfo.size = 12
            tar.addfile(tarinfo, BytesIO(b"A tar member"))

            # Add this even if with_dev=False, so that we can test
            # whether the preparation step removes it.
            tarinfo = tarfile.TarInfo(name="./dev/null")
            tarinfo.type = tarfile.CHRTYPE
            tarinfo.devmajor = 1
            tarinfo.devminor = 3
            tar.addfile(tarinfo)

            # We're skipping parent directories, but that's fine, this will
            # never be extracted.
            for path in directories:
                tarinfo = tarfile.TarInfo(name=f".{path}")
                tarinfo.type = tarfile.DIRTYPE
                tar.addfile(tarinfo)

            for src, dst in symlinks:
                tarinfo = tarfile.TarInfo(name=f".{dst}")
                tarinfo.type = tarfile.SYMTYPE
                tarinfo.linkname = str(src)
                tar.addfile(tarinfo)

            for path in empty_files:
                tarinfo = tarfile.TarInfo(name=f".{path}")
                tarinfo.size = 0
                tar.addfile(tarinfo, BytesIO(b""))

    def patch_fetch_artifact_for_basetgz(
        self,
        *,
        with_dev: bool,
        with_resolvconf_symlink: bool = False,
        with_dpkg_available: bool = True,
        codename: str = "bookworm",
        filename: str = "system.tar.xz",
    ) -> MagicMock:
        """Patch self.task.fetch_artifact(), pretend to download basetgz."""
        patcher = mock.patch.object(self.task, "fetch_artifact", autospec=True)
        mocked = patcher.start()

        def _create_basetgz_file(
            artifact_id: int,
            destination: Path,
            default_category: (  # noqa: ARG001, U100
                CollectionCategory | None
            ) = None,
        ) -> ArtifactResponse:
            symlinks: list[tuple[Path, Path]] = []
            empty_files: list[Path] = []
            if with_resolvconf_symlink:
                symlinks.append(
                    (
                        Path("../run/systemd/resolve/stub-resolv.conf"),
                        Path("/etc/resolv.conf"),
                    )
                )
            if with_dpkg_available:
                empty_files.append(Path("/var/lib/dpkg/available"))
            self.create_basetgz_tarfile(
                destination / filename,
                symlinks=symlinks,
                empty_files=empty_files,
            )

            fake_url = "https://not-used"

            data = DebianSystemTarball(
                filename=filename,
                vendor="debian",
                codename=codename,
                components=["main"],
                mirror=fake_url,
                variant=None,
                pkglist={},
                architecture="amd64",
                with_dev=with_dev,
                with_init=False,
            )

            file_model = FileResponse(
                size=100,
                checksums={"sha256": "not-used"},
                type="file",
                url=fake_url,
            )

            files = FilesResponseType({"system.tar.xz": file_model})

            return create_artifact_response(
                id=artifact_id,
                workspace="System",
                category=ArtifactCategory.SYSTEM_TARBALL,
                created_at=dt.datetime.now(tz=dt.UTC),
                data=data.model_dump(),
                download_tar_gz_url=(
                    f"http://localhost/artifact/{artifact_id}?archive=tar.gz"
                ),
                files_to_upload=[],
                files=files,
            )

        mocked.side_effect = _create_basetgz_file

        self.addCleanup(patcher.stop)

        return mocked

    def test_configure_for_execution_from_artifact_id_error_no_debs(
        self,
    ) -> None:
        """configure_for_execution() no .deb-s: return False."""
        download_directory = self.create_temporary_directory()
        (file1 := download_directory / "file1.dsc").write_text("")
        (file2 := download_directory / "file2.changes").write_text("")

        self.assertFalse(self.task.configure_for_execution(download_directory))

        files = sorted([str(file1), str(file2)])
        log_file_contents = (
            self.task.log_directory / "configure_for_execution.log"
        ).read_text()
        self.assertEqual(
            log_file_contents,
            f"There must be at least one *.deb file. Current files: {files}\n",
        )

    def test_configure_for_execution_prepare_executor_base_tgz(self) -> None:
        """configure_for_execution() prepare the executor and base tgz."""
        download_directory = self.create_temporary_directory()
        self.write_deb_file(download_directory / "package1_1.0_all.deb")

        self.patch_prepare_executor_instance()

        with (
            mock.patch.object(self.task, "_base_tar_data", autospec=True),
            mock.patch.object(
                self.task, "_prepare_base_tgz", autospec=True
            ) as prepare_base_tgz_mocked,
            mock.patch.object(
                self.task, "_check_piuparts_version", autospec=True
            ) as check_piuparts_version_mocked,
        ):
            self.assertTrue(
                self.task.configure_for_execution(download_directory)
            )

        run_called_with = [
            call(
                ["apt-get", "update"],
                run_as_root=True,
                check=True,
                stdout=mock.ANY,
                stderr=mock.ANY,
            ),
            call(
                ["apt-get", "--yes", "install", "piuparts"],
                run_as_root=True,
                check=True,
                stdout=mock.ANY,
                stderr=mock.ANY,
            ),
        ]
        assert isinstance(self.task.executor_instance, MagicMock)
        self.task.executor_instance.run.assert_has_calls(run_called_with)

        prepare_base_tgz_mocked.assert_called_with(download_directory)

        check_piuparts_version_mocked.assert_called_once()

    def test_check_piuparts_version_stretch(self) -> None:
        with (
            mock.patch.object(
                self.task, "get_package_version", return_value=Version("0.77")
            ),
            self.assertRaisesRegex(
                TaskConfigError,
                (
                    r"The environment contains a version of piuparts \(0\.77\) "
                    r"that is too old\. Container-based backends require "
                    r"piuparts >= 1\.5\. We recommend a >= trixie environment "
                    r"with an older base_tgz\."
                ),
            ),
        ):
            self.task._check_piuparts_version()

    def test_check_piuparts_version_trixie(self) -> None:
        with mock.patch.object(
            self.task, "get_package_version", return_value=Version("1.6.0")
        ):
            self.task._check_piuparts_version()

    def test_check_piuparts_version_incus_vm(self) -> None:
        self.task = Piuparts(
            {
                **self.SAMPLE_TASK_DATA,
                "backend": BackendType.INCUS_VM,
            }
        )

        with mock.patch.object(
            self.task, "get_package_version"
        ) as mock_get_package_version:
            self.task._check_piuparts_version()

        mock_get_package_version.assert_not_called()

    def test_prepare_base_tgz_raise_assertion_error(self) -> None:
        """_prepare_base_tgz raise AssertionError: executor_instance is None."""
        msg = r"^self\.executor_instance cannot be None$"
        self.assertRaisesRegex(
            AssertionError,
            msg,
            self.task._prepare_base_tgz,
            self.create_temporary_directory(),
        )

    def test_prepare_base_tgz_download_base_tgz_no_processing(self) -> None:
        """
        _prepare_base_tgz does not do any artifact processing (with_dev=False).

        * Download the artifact to the correct directory
        * Set self.task._base_tar to it
        """
        codename = "bullseye"
        filename = "system-image.tar.xz"
        fetch_artifact_mocked = self.patch_fetch_artifact_for_basetgz(
            with_dev=False, codename=codename, filename=filename
        )
        self.set_dynamic_data()

        # Use mock's side effect to set self.task.executor_instance
        self.patch_prepare_executor_instance()
        self.task._prepare_executor_instance()

        download_directory = self.create_temporary_directory()

        self.task._prepare_base_tgz(download_directory)

        destination_dir = download_directory / "base_tar"

        fetch_artifact_mocked.assert_called_with(
            self.task.data.base_tgz, destination_dir
        )
        self.assertEqual(self.task._base_tar, destination_dir / filename)

        with tarfile.open(self.task._base_tar) as tar:
            self.assertEqual(
                tar.getnames(),
                ["./foo", "./dev/null", "./var/lib/dpkg/available"],
            )

    def test_prepare_base_tgz_download_base_tgz_remove_dev_files(self) -> None:
        r"""_prepare_base_tgz download the artifact and remove /dev/\*."""
        self.set_dynamic_data()

        # Use mock's side effect to set self.task.executor_instance
        self.patch_prepare_executor_instance()
        self.task._prepare_executor_instance()

        download_directory = self.create_temporary_directory()
        destination_dir = download_directory / "base_tar"

        fetch_artifact_mocked = self.patch_fetch_artifact_for_basetgz(
            with_dev=True
        )

        system_tar = destination_dir / "system.tar.gz"

        self.task._prepare_base_tgz(download_directory)

        fetch_artifact_mocked.assert_called_with(
            self.task.data.base_tgz, destination_dir
        )

        # Downloaded system.tar.xz, then processed it and wrote the result
        # to system.tar.gz
        self.assertEqual(self.task._base_tar, system_tar)

        with tarfile.open(system_tar, "r:gz") as tar:
            self.assertEqual(
                tar.getnames(), ["./foo", "./var/lib/dpkg/available"]
            )

    def test_prepare_base_tgz_download_base_tgz_resolveconf(self) -> None:
        r"""_prepare_base_tgz download the artifact and install resolv.conf."""
        self.set_dynamic_data()

        # Use mock's side effect to set self.task.executor_instance
        self.patch_prepare_executor_instance()
        self.task._prepare_executor_instance()

        download_directory = self.create_temporary_directory()
        destination_dir = download_directory / "base_tar"

        fetch_artifact_mocked = self.patch_fetch_artifact_for_basetgz(
            with_dev=False, with_resolvconf_symlink=True
        )

        self.task._prepare_base_tgz(download_directory)

        fetch_artifact_mocked.assert_called_with(
            self.task.data.base_tgz, destination_dir
        )

        # Downloaded system.tar.xz, then processed it and wrote the result
        # to system.tar.gz
        system_tar = destination_dir / "system.tar.gz"
        self.assertEqual(self.task._base_tar, system_tar)

        with tarfile.open(system_tar, "r:gz") as tar:
            self.assertEqual(
                tar.getnames(),
                [
                    "./foo",
                    "./dev/null",
                    "./var/lib/dpkg/available",
                    "./etc",
                    "./etc/resolv.conf",
                ],
            )
            self.assertTrue(tar.getmember("./etc/resolv.conf").isreg())

    def test_prepare_base_tgz_download_base_tgz_dpkg_available(self) -> None:
        r"""_prepare_base_tgz download the artifact and touch dpkg available."""
        self.set_dynamic_data()

        # Use mock's side effect to set self.task.executor_instance
        self.patch_prepare_executor_instance()
        self.task._prepare_executor_instance()

        download_directory = self.create_temporary_directory()
        destination_dir = download_directory / "base_tar"

        fetch_artifact_mocked = self.patch_fetch_artifact_for_basetgz(
            with_dev=False, with_dpkg_available=False
        )

        self.task._prepare_base_tgz(download_directory)

        fetch_artifact_mocked.assert_called_with(
            self.task.data.base_tgz, destination_dir
        )

        # Downloaded system.tar.xz, then processed it and wrote the result
        # to system.tar.gz
        system_tar = destination_dir / "system.tar.gz"
        self.assertEqual(self.task._base_tar, system_tar)

        with tarfile.open(system_tar, "r:gz") as tar:
            self.assertEqual(
                tar.getnames(),
                [
                    "./foo",
                    "./dev/null",
                    "./var",
                    "./var/lib",
                    "./var/lib/dpkg",
                    "./var/lib/dpkg/available",
                ],
            )
            self.assertTrue(tar.getmember("./var/lib/dpkg/available").isreg())

    def test_filter_tar_existing_directory(self) -> None:
        directory = self.create_temporary_directory()
        filename = directory / "input.tar.xz"
        self.create_basetgz_tarfile(filename, directories=[Path("/var")])
        output = self.task._filter_tar(
            filename, create_empty_files=[Path("/var/test")]
        )
        with tarfile.open(output, "r:gz") as tar:
            self.assertEqual(
                tar.getnames(),
                [
                    "./foo",
                    "./dev/null",
                    "./var",
                    "./var/test",
                ],
            )

    def prepare_script(self, codename: str, script_name: str) -> str:
        """Build scripts and return the contents of script_name."""
        directory = self.create_temporary_directory()
        with mock.patch.object(self.task, "_base_tar_data", autospec=True):
            assert self.task._base_tar_data
            self.task._base_tar_data.codename = codename
            self.task._prepare_scripts(directory)
        script = directory / script_name
        self.assertTrue(script.exists())
        return script.read_text()

    def test_prepare_scripts_bad_permissions(self) -> None:
        """_prepare_scripts overrides pre_remove_50_find_bad_permissions."""
        self.set_dynamic_data()
        script = self.prepare_script(
            "trixie", "pre_remove_50_find_bad_permissions"
        )
        self.assertIn("Skipping", script)

    def test_prepare_scripts_install_useradd(self) -> None:
        """_prepare_scripts adds a script to install useradd."""
        self.set_dynamic_data()
        script = self.prepare_script("trixie", "post_setup_00_passwd")
        self.assertIn("apt-get -y install passwd", script)

    def prepare_extra_repo_script(self, codename: str) -> str:
        """Build and read the post_chroot_unpack_debusine_extra_repos script."""
        return self.prepare_script(
            codename, "post_chroot_unpack_debusine_extra_repos"
        )

    def test_prepare_scripts_buster(self) -> None:
        """_prepare_scripts with deb822 signed-by apt sources."""
        self.configure_task(
            override={
                "extra_repositories": [
                    {
                        "url": "http://example.net",
                        "suite": "bookworm",
                        "components": ["main"],
                        "signing_key": "KEY A",
                    },
                ]
            }
        )
        self.resolver.field_values["extra_repositories"] = [
            ExtraExternalRepository(
                url="http://example.net",
                suite="bookworm",
                components=["main"],
                signing_key="KEY A",
            )
        ]
        self.set_dynamic_data()
        script = self.prepare_extra_repo_script("buster")
        self.assertIn("mkdir -p /etc/apt/keyrings\n", script)
        key = shlex.quote("KEY A\n")
        self.assertIn(
            f"printf %s {key} > /etc/apt/keyrings/extra_apt_key_0.asc\n",
            script,
        )
        source = shlex.quote(
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.net/
                Suites: bookworm
                Components: main
                Signed-By: /etc/apt/keyrings/extra_apt_key_0.asc
                """
            )
        )
        self.assertIn(
            (
                f"printf %s {source} > "
                f"/etc/apt/sources.list.d/extra_repository_0.sources\n"
            ),
            script,
        )

    def test_prepare_scripts_bookworm(self) -> None:
        """_prepare_scripts with deb822 signed-by apt sources."""
        self.configure_task(
            override={
                "extra_repositories": [
                    {
                        "url": "http://example.net",
                        "suite": "bookworm",
                        "components": ["main"],
                        "signing_key": "KEY A",
                    },
                ]
            }
        )
        self.resolver.field_values["extra_repositories"] = [
            ExtraExternalRepository(
                url="http://example.net",
                suite="bookworm",
                components=["main"],
                signing_key="KEY A",
            )
        ]
        self.set_dynamic_data()
        script = self.prepare_extra_repo_script("bookworm")
        self.assertNotIn("/etc/apt/keyrings", script)
        source = shlex.quote(
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.net/
                Suites: bookworm
                Components: main
                Signed-By:
                 KEY A
                """
            )
        )
        self.assertIn(
            (
                f"printf %s {source} > "
                f"/etc/apt/sources.list.d/extra_repository_0.sources\n"
            ),
            script,
        )

    def test_prepare_scripts_extra_debusine_repositories(self) -> None:
        """_prepare_scripts with extra repositories hosted by Debusine."""
        signing_key = "\n".join(
            (
                "-----BEGIN PGP PUBLIC KEY BLOCK-----",
                "",
                "ABCDEFGHI",
                "-----END PGP PUBLIC KEY BLOCK-----",
            )
        )
        self.configure_task(
            override={
                "extra_repositories": [{"suite": "trixie-local@debian:suite"}]
            }
        )
        self.resolver.field_values["extra_repositories"] = [
            ExtraExternalRepository(
                url="http://deb.example.com/scope/workspace",
                suite="trixie-local",
                components=["main"],
                signing_key=signing_key,
            )
        ]
        self.set_dynamic_data()
        script = self.prepare_extra_repo_script("trixie")
        self.assertNotIn("/etc/apt/keyrings", script)
        source = shlex.quote(
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://deb.example.com/scope/workspace
                Suites: trixie-local
                Components: main
                Signed-By:
                 -----BEGIN PGP PUBLIC KEY BLOCK-----
                 .
                 ABCDEFGHI
                 -----END PGP PUBLIC KEY BLOCK-----
                """
            )
        )
        self.assertIn(
            (
                f"printf %s {source} > "
                f"/etc/apt/sources.list.d/extra_repository_0.sources\n"
            ),
            script,
        )

    def test_check_directory_for_consistency_errors_none(self) -> None:
        """There are no consistency errors."""
        build_directory = self.create_temporary_directory()
        (build_directory / self.task.CAPTURE_OUTPUT_FILENAME).touch()
        self.assertEqual(
            self.task.check_directory_for_consistency_errors(build_directory),
            [],
        )

    def test_check_directory_for_consistency_errors(self) -> None:
        """There is one consistency error: no piuparts.txt."""
        build_directory = self.create_temporary_directory()

        expected = (
            f"{self.task.CAPTURE_OUTPUT_FILENAME} not in {build_directory}"
        )
        self.assertEqual(
            self.task.check_directory_for_consistency_errors(build_directory),
            [expected],
        )

    def test_upload_artifacts(self) -> None:
        """``upload_artifacts`` uploads an artifact and creates relations."""
        execute_directory = self.create_temporary_directory()
        piuparts_output = execute_directory / self.task.CAPTURE_OUTPUT_FILENAME
        piuparts_output.write_text("piuparts output\n")

        self.task._source_artifacts_ids = [1, 2]
        self.task._deb_versions = {"hello": "1.0", "hello-doc": "1.0"}
        piuparts_version = "1.6.0"
        with mock.patch.object(
            self.task,
            "get_package_version",
            return_value=Version(piuparts_version),
        ):
            self.task._check_piuparts_version()
        codename = "trixie"
        self.task._base_tar_data = create_system_tarball_data(codename=codename)

        debusine_mock = self.mock_debusine()
        workspace_name = "testing"
        uploaded_artifact = create_remote_artifact(
            id=10, workspace=workspace_name
        )
        debusine_mock.upload_artifact.side_effect = [uploaded_artifact]

        self.task.workspace_name = workspace_name
        self.task.work_request_id = work_request_id = 147

        self.task.upload_artifacts(
            execute_directory, execution_result=WorkRequestResults.SUCCESS
        )

        debusine_mock.upload_artifact.assert_called_once_with(
            PiupartsArtifact.create(
                piuparts_output=piuparts_output,
                data=DebianPiuparts(
                    piuparts_version=piuparts_version,
                    binary_packages=self.task._deb_versions,
                    architecture="amd64",
                    distribution=f"debian:{codename}",
                ),
            ),
            workspace=workspace_name,
            work_request=work_request_id,
        )
        expected_relation_create = [
            call(
                uploaded_artifact.id,
                source_artifact_id,
                RelationType.RELATES_TO,
            )
            for source_artifact_id in self.task._source_artifacts_ids
        ]
        self.assertEqual(
            debusine_mock.relation_create.call_count,
            len(expected_relation_create),
        )
        debusine_mock.relation_create.assert_has_calls(expected_relation_create)

    def test_execute(self) -> None:
        """Test full (mocked) execution."""
        self.configure_task(override={"input": {'binary_artifacts': [1, 2]}})
        self.task.dynamic_data = PiupartsDynamicData(
            environment_id=7, input_binary_artifacts_ids=[1, 2], base_tgz_id=42
        )
        debusine_mock = self.mock_debusine()
        self.mock_image_download(debusine_mock)
        self.patch_prepare_executor_instance()
        download_directory = self.create_temporary_directory()

        fetch_artifact_mocked = self.patch_fetch_artifact_for_basetgz(
            with_dev=True, codename="trixie"
        )
        self.assertTrue(self.task.fetch_input(download_directory))
        fetch_artifact_mocked.assert_any_call(1, download_directory)
        fetch_artifact_mocked.assert_any_call(2, download_directory)

        self.write_deb_file(file2 := download_directory / "file2_1.0_amd64.deb")
        self.write_deb_file(file1 := download_directory / "file1_1.0_amd64.deb")
        self.write_deb_file(
            file3 := download_directory / "makedev_2.3.1-97_all.deb"
        )

        with mock.patch.object(
            self.task,
            "get_package_version",
            autospec=True,
            return_value="1.6.0",
        ) as check_piuparts_version_mocked:
            self.assertTrue(
                self.task.configure_for_execution(download_directory)
            )

        check_piuparts_version_mocked.assert_called_once()

        self.assertEqual(self.task._deb_files, [file1, file2, file3])
        self.assertEqual(
            self.task._deb_versions,
            {"file1": "1.0", "file2": "1.0", "makedev": "2.3.1-97"},
        )
        self.assertEqual(
            self.task._cmdline(),
            [
                "piuparts",
                "--keep-sources-list",
                "--allow-database",
                "--warn-on-leftovers-after-purge",
                f"--basetgz={self.task._base_tar}",
                "--scriptsdir=/etc/piuparts/scripts",
                f"--scriptsdir={self.task._scripts_dir}",
                str(file1),
                str(file2),
                str(file3),
            ],
        )

        (download_directory / self.task.CAPTURE_OUTPUT_FILENAME).write_text(
            "piuparts output\n"
        )

        assert self.task.executor is not None
        self.task.executor.system_image = self.fake_system_tarball_artifact()

        self.task.upload_artifacts(
            download_directory, execution_result=WorkRequestResults.SUCCESS
        )
