# 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 base Task classes."""

import abc
import datetime as dt
import errno
import itertools
import math
import os
import select
import shlex
import signal
import subprocess
import sys
import tempfile
import textwrap
import time as tm
import traceback
from collections.abc import Iterator
from pathlib import Path
from typing import Any, cast, override
from unittest import mock
from unittest.mock import MagicMock

import psutil
from debian.debian_support import Version

import debusine.worker.tags as wtags
from debusine.artifacts import WorkRequestDebugLogs
from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianBinaryPackage,
    DebianSourcePackage,
    TaskTypes,
    WorkRequestResults,
)
from debusine.client.debusine import Debusine
from debusine.tasks import (
    BaseExternalTask,
    ExtraRepositoryMixin,
    RunCommandTask,
    Sbuild,
    TaskConfigError,
    analyze_external_worker_tasks,
    ensure_artifact_categories,
    ensure_collection_category,
    get_provided_external_worker_tags,
    inputs,
)
from debusine.tasks.executors import (
    ExecutorImageCategory,
    ExecutorInterface,
    InstanceInterface,
)
from debusine.tasks.inputs.mock_resolver import MockFieldResolver
from debusine.tasks.inputs.tests.test_inputs import ConcreteTaskInput
from debusine.tasks.models import (
    ArtifactInfo,
    BackendType,
    BaseDynamicTaskData,
    BaseDynamicTaskDataWithExecutor,
    BaseDynamicTaskDataWithExtraRepositories,
    BaseTaskData,
    BaseTaskDataWithExecutor,
    BaseTaskDataWithExtraRepositories,
    ExtraDebusineRepository,
    ExtraExternalRepository,
    ExtraRepository,
    WorkerType,
)
from debusine.tasks.tests.helpers import (
    ExternalTaskHelperMixin,
    SampleBaseExternalTask,
    SampleBaseTaskWithExecutor,
)
from debusine.test import TestCase
from debusine.test.test_utils import (
    create_artifact_response,
    preserve_task_registry,
)
from debusine.worker.system_information import native_architecture


class SampleBaseTask1(SampleBaseExternalTask[BaseTaskData]):
    """Sample class to test BaseExternalTask class."""

    @override
    def run(self, execute_directory: Path) -> WorkRequestResults:
        """Unused abstract method from BaseExternalTask."""
        raise NotImplementedError()


class SampleBaseTaskWithExecutor1(
    SampleBaseTaskWithExecutor[
        BaseTaskDataWithExecutor, BaseDynamicTaskDataWithExecutor
    ],
):
    """Sample class to test BaseTaskWithExecutor class."""

    @override
    def run(self, execute_directory: Path) -> WorkRequestResults:
        """Unused abstract method from BaseExternalTask."""
        raise NotImplementedError()


class SampleBaseTask2Data(BaseTaskData):
    """Data representation for SampleBaseTask2."""

    foo: str
    bar: float | None = None
    build_architecture: str | None = None


class SampleBaseTask2(
    SampleBaseExternalTask[SampleBaseTask2Data],
):
    """Test BaseExternalTask class with data validation."""

    TASK_VERSION = 1

    @override
    def run(self, execute_directory: Path) -> WorkRequestResults:
        """Unused abstract method from BaseExternalTask."""
        raise NotImplementedError()


class SampleBaseTaskDataWithExecutorAndArchitecture(BaseTaskDataWithExecutor):
    """BaseTaskDataWithExecutor with a build architecture."""

    build_architecture: str | None = None


class SampleBaseTaskWithExecutorAndArchitecture(
    SampleBaseTaskWithExecutor[
        SampleBaseTaskDataWithExecutorAndArchitecture,
        BaseDynamicTaskDataWithExecutor,
    ],
):
    """Test BaseTaskWithExecutor with a build architecture."""

    TASK_VERSION = 1

    @override
    def run(self, execute_directory: Path) -> WorkRequestResults:
        """Unused abstract method from BaseExternalTask."""
        raise NotImplementedError()


class SampleExtraRepositoryMixin(
    SampleBaseExternalTask[
        BaseTaskDataWithExtraRepositories,
        BaseDynamicTaskDataWithExtraRepositories,
    ],
    ExtraRepositoryMixin[
        BaseTaskDataWithExtraRepositories,
        BaseDynamicTaskDataWithExtraRepositories,
    ],
):
    """Test ExtraRepositoryMixin."""

    @override
    def compute_dynamic_data(self) -> BaseDynamicTaskDataWithExtraRepositories:
        return BaseDynamicTaskDataWithExtraRepositories(
            extra_repositories=self.extra_repositories
        )

    @override
    def run(self, execute_directory: Path) -> WorkRequestResults:
        """Unused abstract method from BaseExternalTask."""
        raise NotImplementedError()


class BaseTaskTests(TestCase):
    """Unit tests for BaseExternalTask class."""

    worker_metadata = {"system:worker_type": WorkerType.EXTERNAL}

    @override
    def setUp(self) -> None:
        """Create the shared attributes."""
        super().setUp()
        self.task = SampleBaseTask1({})
        self.task.set_working_directory(self.create_temporary_directory())
        self.task2 = SampleBaseTask2({"foo": "bar"})
        self.task2.set_working_directory(self.create_temporary_directory())

    def test_task_initialize_member_variables(self) -> None:
        """Member variables expected used in other methods are initialized."""
        self.assertIsNone(self.task.workspace_name)
        self.assertEqual(self.task._source_artifacts_ids, [])

    def test_name_is_set(self) -> None:
        """task.name is built from class name."""
        self.assertEqual(self.task.name, "samplebasetask1")

    def test_logger_is_configured(self) -> None:
        """task.logger is available."""
        self.assertIsNotNone(self.task.logger)

    @preserve_task_registry()
    def test_prefix_with_task_name(self) -> None:
        """task.prefix_with_task_name does what it says."""
        self.assertEqual(
            self.task.prefix_with_task_name("foobar"),
            f"{self.task.name}:foobar",
        )

        # Cannot import ServerNoop or NoopWorkflow because sometimes these
        # tests are run without the django test runner

        class SampleServer(BaseExternalTask[BaseTaskData, BaseDynamicTaskData]):
            TASK_TYPE = TaskTypes.SERVER

            @override
            def _execute(self) -> WorkRequestResults:
                raise NotImplementedError()

            @override
            def _upload_work_request_debug_logs(self) -> None:
                raise NotImplementedError()

        self.assertEqual(
            SampleServer.prefix_with_task_name("foo"), "server:sampleserver:foo"
        )

        class SampleInternal(
            BaseExternalTask[BaseTaskData, BaseDynamicTaskData]
        ):
            TASK_TYPE = TaskTypes.INTERNAL

            @override
            def _execute(self) -> WorkRequestResults:
                raise NotImplementedError()

            @override
            def _upload_work_request_debug_logs(self) -> None:
                raise NotImplementedError()

        self.assertEqual(
            SampleInternal.prefix_with_task_name("foo"),
            "internal:sampleinternal:foo",
        )

        class SampleWorkflow(
            BaseExternalTask[BaseTaskData, BaseDynamicTaskData]
        ):
            TASK_TYPE = TaskTypes.WORKFLOW

            @override
            def _execute(self) -> WorkRequestResults:
                raise NotImplementedError()

            @override
            def _upload_work_request_debug_logs(self) -> None:
                raise NotImplementedError()

        self.assertEqual(
            SampleWorkflow.prefix_with_task_name("foo"),
            "workflow:sampleworkflow:foo",
        )

        class SampleSigning(
            BaseExternalTask[BaseTaskData, BaseDynamicTaskData]
        ):
            TASK_TYPE = TaskTypes.SIGNING

            @override
            def _execute(self) -> WorkRequestResults:
                raise NotImplementedError()

            @override
            def _upload_work_request_debug_logs(self) -> None:
                raise NotImplementedError()

        self.assertEqual(
            SampleSigning.prefix_with_task_name("foo"),
            "signing:samplesigning:foo",
        )

        class SampleWait(BaseExternalTask[BaseTaskData, BaseDynamicTaskData]):
            TASK_TYPE = TaskTypes.WAIT

            @override
            def _execute(self) -> WorkRequestResults:
                raise NotImplementedError()

            @override
            def _upload_work_request_debug_logs(self) -> None:
                raise NotImplementedError()

        self.assertEqual(
            SampleWait.prefix_with_task_name("foo"), "wait:samplewait:foo"
        )

    def test_configure_sets_data_attribute(self) -> None:
        """task.data is set with configure."""
        task_data = {
            "foo": "bar",
        }

        data = SampleBaseTask2Data.model_validate(task_data)

        self.assertEqual(
            data,
            SampleBaseTask2Data(
                foo="bar",
            ),
        )

    def test_configure_task_data_notification_drop(self) -> None:
        """Raise TaskConfigError: "notifications" is not valid anymore."""
        task_data = SampleBaseTask2Data(foo="bar")
        task_data_notification = {
            "notifications": {"on_failure": [{"channel": "test"}]}
        }

        with self.assertRaisesRegex(TaskConfigError, "notifications"):
            SampleBaseTask2(
                {**task_data.model_dump(), **task_data_notification}
            )

    def test_configure_fails_with_bad_schema(self) -> None:
        """configure() raises TaskConfigError if schema is not respected."""
        task_data = {"nonexistent": "bar"}
        with self.assertRaises(TaskConfigError):
            SampleBaseTask2(task_data)

    def test_configure_works_with_good_schema(self) -> None:
        """configure() doesn't raise TaskConfigError."""
        task_data = SampleBaseTask2Data(foo="bar", bar=3.14)

        task2 = SampleBaseTask2(task_data.model_dump())

        self.assertEqual(task2.data.foo, "bar")
        self.assertEqual(task2.data.bar, 3.14)

    def test_get_provided_worker_tags(self) -> None:
        self.assertEqual(
            self.task2.get_provided_worker_tags(),
            {wtags.TASK_PREFIX + "worker:samplebasetask2:version:1"},
        )

    def test_get_provided_external_worker_tags(self) -> None:
        self.enterContext(
            mock.patch.object(
                BaseExternalTask,
                "_sub_tasks",
                new={
                    TaskTypes.WORKER: {
                        "task": SampleBaseTask1,
                        "task2": SampleBaseTask2,
                    }
                },
            )
        )
        self.assertEqual(
            get_provided_external_worker_tags(),
            {
                wtags.TASK_PREFIX + "worker:samplebasetask1:version:None",
                wtags.TASK_PREFIX + "worker:samplebasetask2:version:1",
            },
        )

    def test_analyze_worker_without_task_version(self) -> None:
        """analyze_worker() reports an unknown task version."""
        metadata = self.task.analyze_worker()
        self.assertIsNone(metadata["samplebasetask1:version"])

    def test_analyze_worker_with_task_version(self) -> None:
        """analyze_worker() reports a task version."""
        metadata = self.task2.analyze_worker()
        self.assertEqual(metadata["samplebasetask2:version"], 1)

    def test_analyze_external_worker(self) -> None:
        patcher = mock.patch.object(
            BaseExternalTask,
            "_sub_tasks",
            new={
                TaskTypes.WORKER: {
                    "task": SampleBaseTask1,
                    "task2": SampleBaseTask2,
                }
            },
        )
        patcher.start()
        self.addCleanup(patcher.stop)

        metadata = analyze_external_worker_tasks()
        self.assertEqual(
            metadata,
            self.task2.analyze_worker() | self.task.analyze_worker(),
        )

        # Assert that metadata contains data
        self.assertNotEqual(metadata, {})

    def test_class_for_name_invalid_type(self) -> None:
        """class_from_name raises a ValueError exception."""
        with self.assertRaisesRegex(
            ValueError, "'not-existing-type' is not a registered task type"
        ):
            BaseExternalTask.class_from_name(
                cast(TaskTypes, "not-existing-type"), "non-existing-class"
            )

    def test_class_for_name_sbuild(self) -> None:
        """class_from_name returns Sbuild (case-insensitive)."""
        self.assertEqual(
            BaseExternalTask.class_from_name(TaskTypes.WORKER, 'sBuIlD'), Sbuild
        )

    def test_class_for_name_no_class(self) -> None:
        """class_from_name raises a ValueError exception."""
        with self.assertRaisesRegex(
            ValueError,
            "'non-existing-class' is not a registered Worker task_name",
        ):
            BaseExternalTask.class_from_name(
                TaskTypes.WORKER, 'non-existing-class'
            )

    @preserve_task_registry()
    def test_class_for_name_no_duplicates(self) -> None:
        """
        __init_subclass__ raises AssertionError for duplicated names.

        __init_subclass__ uses the class name in lowercase: it can
        cause unexpected duplicated names. Check that are detected.
        """
        msg = "Two Tasks with the same name: 'somesubclassname'"
        with self.assertRaisesRegex(AssertionError, msg):

            class Somesubclassname(
                SampleBaseExternalTask[BaseTaskData, BaseDynamicTaskData],
            ):
                @override
                def run(
                    self,
                    execute_directory: Path,  # noqa: U100
                ) -> WorkRequestResults:
                    raise NotImplementedError()

            class SomeSubclassName(
                SampleBaseExternalTask[BaseTaskData, BaseDynamicTaskData],
            ):
                @override
                def run(
                    self,
                    execute_directory: Path,  # noqa: U100
                ) -> WorkRequestResults:
                    raise NotImplementedError()

    @preserve_task_registry()
    def test_conflict_names_server_worker(self) -> None:
        """Server and Worker tasks namespaces are shared."""
        msg = "'somesubclassname' already registered as a Server task"
        with self.assertRaisesRegex(AssertionError, msg):

            class Somesubclassname(
                SampleBaseExternalTask[BaseTaskData, BaseDynamicTaskData]
            ):
                TASK_TYPE = TaskTypes.SERVER

                @override
                def _execute(self) -> WorkRequestResults:
                    raise NotImplementedError()

                @override
                def _upload_work_request_debug_logs(self) -> None:
                    raise NotImplementedError()

            class SomeSubclassName(
                SampleBaseExternalTask[BaseTaskData, BaseDynamicTaskData],
            ):
                @override
                def run(
                    self,
                    execute_directory: Path,  # noqa: U100
                ) -> WorkRequestResults:
                    raise NotImplementedError()

        msg = "'somesubclassname1' already registered as a Worker task"
        with self.assertRaisesRegex(AssertionError, msg):

            class SomeSubclassName1(
                SampleBaseExternalTask[BaseTaskData, BaseDynamicTaskData],
            ):
                @override
                def run(
                    self,
                    execute_directory: Path,  # noqa: U100
                ) -> WorkRequestResults:
                    raise NotImplementedError()

            class Somesubclassname1(
                SampleBaseExternalTask[BaseTaskData, BaseDynamicTaskData]
            ):
                TASK_TYPE = TaskTypes.SERVER

                @override
                def _execute(self) -> WorkRequestResults:
                    raise NotImplementedError()

                @override
                def _upload_work_request_debug_logs(self) -> None:
                    raise NotImplementedError()

    def test_init_subclass_no_register_abstract_class(self) -> None:
        class Abstract(
            BaseExternalTask[BaseTaskData, BaseDynamicTaskData], abc.ABC
        ):
            @abc.abstractmethod
            def _do_something(self) -> None:
                raise NotImplementedError()

        for task_type, registry in BaseExternalTask._sub_tasks.items():
            with self.subTest(type=task_type.value):
                self.assertNotIn("abstract", registry)
                self.assertNotIn("basetask", registry)
                self.assertNotIn("baseexternaltask", registry)
                self.assertNotIn("basetaskwithexecutor", registry)
                self.assertNotIn("runcommandtask", registry)

    def test_execute_logging_exceptions_execute(self) -> None:
        patcher = mock.patch.object(self.task, "execute", autospec=True)
        mocker = patcher.start()
        self.addCleanup(patcher.stop)

        return_values = [True, False]
        for return_value in return_values:
            mocker.return_value = return_value
            self.assertEqual(
                self.task.execute_logging_exceptions(), return_value
            )

        self.assertEqual(mocker.call_count, len(return_values))

    def test_execute_logging_exceptions_handle_exception(self) -> None:
        """BaseExternalTask.execute_logging_exceptions() logs the exception."""
        patcher = mock.patch.object(self.task, "execute", autospec=True)
        mocker = patcher.start()
        msg = "Exception message"
        mocker.side_effect = ValueError(msg)
        self.addCleanup(patcher.stop)

        with (
            self.assertRaisesRegex(ValueError, msg),
            self.assertLogs(self.task.logger) as log,
        ):
            self.task.execute_logging_exceptions()

        self.assertEqual(len(log.output), 1)
        lines = log.output[0].splitlines()
        self.assertEqual(
            lines[:2],
            [
                f"INFO:debusine.tasks:Exception in Task {self.task.name}",
                "Traceback (most recent call last):",
            ],
        )
        self.assertEqual(lines[-1], f"ValueError: {msg}")

    def test_append_to_log_file(self) -> None:
        contents = ["a 1", "b 2"]
        contents_str = "\n".join(contents) + "\n"

        filename = "log.txt"

        # Add data into the file
        self.task.append_to_log_file(filename, contents)

        file = self.task.log_directory / filename

        self.assertEqual(file.read_text(), contents_str)

        # Add more data
        self.task.append_to_log_file(filename, contents)
        self.assertEqual(file.read_text(), contents_str * 2)

    def test_open_debug_log_file_default_is_append_mode(self) -> None:
        """
        BaseExternalTask.open_debug_log_file() uses a temporary directory.

        self.task._debug_log_files_directory is set to the temporary directory.

        Calling it again: return the same file (append mode by default).
        """
        line1 = "First line\n"
        filename = "test.log"
        with self.task.open_debug_log_file(filename) as file:
            file.write(line1)

        log_file = self.task.log_directory / filename

        self.assertEqual(log_file.read_text(), line1)

        # Verify by default is in append mode: we are adding to it
        line2 = "Second line\n"
        with self.task.open_debug_log_file(filename) as file:
            file.write(line2)

        self.assertEqual(log_file.read_text(), f"{line1}{line2}")

    def test_open_debug_log_file_mode_is_used(self) -> None:
        """BaseExternalTask.open_debug_log_file() use the specified mode."""
        filename = "test.log"

        with self.task.open_debug_log_file(filename) as f:
            f.write("something")

        line = "new line"
        with self.task.open_debug_log_file(filename, mode="w") as f:
            f.write(line)

        file = self.task.log_directory / filename

        # Only the second line is there, because we asked to overwrite
        self.assertEqual(file.read_text(), line)

    def patch_task__execute(self) -> MagicMock:
        """Return mocked object of self.task._execute()."""
        patcher = mock.patch.object(self.task, "_execute", autospec=True)
        mocked = patcher.start()
        self.addCleanup(patcher.stop)
        return mocked

    def patch_task__upload_work_request_debug_logs(self) -> MagicMock:
        """Return mocked object of self.task.upload_work_request_debug_logs."""
        patcher = mock.patch.object(
            self.task, "_upload_work_request_debug_logs", autospec=True
        )
        mocked = patcher.start()
        self.addCleanup(patcher.stop)
        return mocked

    def task_execute_and_assert(
        self, expected_value: WorkRequestResults
    ) -> None:
        """
        Call task.execute() and assert return value is expected_value.

        Set up mocks and ensure that mocks are used as expected.
        """
        _execute_mocked = self.patch_task__execute()
        _execute_mocked.return_value = expected_value
        upload_work_request_debug_logs = (
            self.patch_task__upload_work_request_debug_logs()
        )

        self.assertEqual(self.task.execute(), expected_value)

        _execute_mocked.assert_called_with()
        upload_work_request_debug_logs.assert_called_with()

    def test_execute_return_success(self) -> None:
        """BaseExternalTask.execute() returns SUCCESS and uploads logs."""
        self.task_execute_and_assert(WorkRequestResults.SUCCESS)

    def test_execute_return_failure(self) -> None:
        """BaseExternalTask.execute() returns FAILURE and uploads logs."""
        self.task_execute_and_assert(WorkRequestResults.FAILURE)

    def test_abortion_flag(self) -> None:
        """BaseExternalTask.abort() sets the aborted flag."""
        self.assertFalse(self.task.aborted)
        self.task.abort()
        self.assertTrue(self.task.aborted)

    def test_ensure_artifact_categories_artifact(self) -> None:
        ensure_artifact_categories(
            configuration_key="source_artifact",
            category=ArtifactCategory.TEST,
            expected=[ArtifactCategory.TEST, ArtifactCategory.SOURCE_PACKAGE],
        )

    def test_ensure_artifact_categories_raise(self) -> None:
        with self.assertRaisesRegex(
            TaskConfigError,
            r"^source_artifact: unexpected artifact category: 'debusine:test'. "
            r"Valid categories: \['debian:binary-package'\]$",
        ):
            ensure_artifact_categories(
                configuration_key="source_artifact",
                category=ArtifactCategory.TEST,
                expected=[
                    ArtifactCategory.BINARY_PACKAGE,
                ],
            )

    def test_ensure_collection_category_ok(self) -> None:
        ensure_collection_category(
            configuration_key="source_collection",
            category=CollectionCategory.TEST,
            expected=CollectionCategory.TEST,
        )

    def test_ensure_collection_category_raise(self) -> None:
        with self.assertRaisesRegex(
            TaskConfigError,
            rf"^source_collection: unexpected collection category: "
            rf"'{CollectionCategory.TEST}'\. "
            rf"Expected: '{CollectionCategory.SUITE}'$",
        ):
            ensure_collection_category(
                configuration_key="source_collection",
                category=CollectionCategory.TEST,
                expected=CollectionCategory.SUITE,
            )

    def test_compute_scheduler_tags(self) -> None:
        provided, required = self.task.compute_scheduler_tags()
        self.assertEqual(provided.sets, {})
        self.assertEqual(required.sets, {})

    def test_compute_dynamic_data(self) -> None:
        resolver = MockFieldResolver()
        resolver.resolve_inputs(self.task)
        self.assertEqual(
            self.task.compute_dynamic_data(), BaseDynamicTaskData()
        )

    @preserve_task_registry()
    def test_input_field_values_from_task_data(self) -> None:
        class TaskWithInputs(SampleBaseExternalTask[BaseTaskData]):
            """Concrete task class with an input field."""

            test_field = inputs.DebusineFQDNInput()

        test_value = "test_restore_inputs.example.org"
        task_data = BaseTaskData()
        dynamic_data = BaseDynamicTaskData()
        dynamic_data.task_input_fields["test_field"] = test_value

        with self.spy(
            "debusine.tasks.inputs.DebusineFQDNInput.deserialize_value"
        ) as deserialize_method:
            task = TaskWithInputs(
                task_data.model_dump(), dynamic_data.model_dump()
            )
        self.assertEqual(task.test_field, test_value)
        deserialize_method.assert_called_once()
        self.assertEqual(deserialize_method.call_args.args[1], test_value)

    def test_input_field_values_from_task_data_ignore_spurious(self) -> None:
        task_data = BaseTaskData()
        dynamic_data = BaseDynamicTaskData()
        dynamic_data.task_input_fields["test_field"] = "test"
        task = SampleBaseTask1(
            task_data.model_dump(), dynamic_data.model_dump()
        )
        self.assertFalse(hasattr(task, "test_field"))

    def test_setup_working_directory_creates_directory(self) -> None:
        with tempfile.TemporaryDirectory() as temp_directory:
            workdir = Path(temp_directory) / "workdir"
            self.assertFalse(workdir.exists())

            self.task.set_working_directory(workdir)

            self.assertTrue(workdir.exists())


class SampleBaseExternalTask1(
    SampleBaseExternalTask[BaseTaskData, BaseDynamicTaskData]
):
    """Sample class to test BaseExternalTask class."""

    @override
    def run(self, execute_directory: Path) -> WorkRequestResults:
        """Unused abstract method from BaseExternalTask."""
        raise NotImplementedError()


class SampleBaseExternalTask2(
    SampleBaseExternalTask[SampleBaseTask2Data, BaseDynamicTaskData]
):
    """Test BaseExternalTask class with data validation."""

    TASK_VERSION = 1

    @override
    def run(self, execute_directory: Path) -> WorkRequestResults:
        """Unused abstract method from BaseExternalTask."""
        raise NotImplementedError()


class BaseExternalTaskTests(TestCase):
    """Unit tests for :py:class:`BaseExternalTask`."""

    @override
    def setUp(self) -> None:
        """Create the shared attributes."""
        super().setUp()
        self.task = SampleBaseExternalTask1({})
        self.task2 = SampleBaseExternalTask2({"foo": "bar"})
        tempdir = self.create_temporary_directory()
        self.task.set_working_directory(tempdir / "task1")
        self.task2.set_working_directory(tempdir / "task2")
        self.worker_metadata = {
            "system:worker_type": WorkerType.EXTERNAL,
            "system:architectures": native_architecture(),
        }

    def test_can_run_on_no_version(self) -> None:
        """Ensure can_run_on returns True if no version is specified."""
        self.enterContext(
            mock.patch(
                "debusine.tasks._task.FALLBACK_ARCHITECTURE",
                native_architecture(),
            )
        )
        self.assertIsNone(self.task.TASK_VERSION)
        metadata = {**self.worker_metadata, **self.task.analyze_worker()}
        self.assertTrue(self.task.can_run_on(metadata))

    def test_can_run_on_with_different_versions(self) -> None:
        """Ensure can_run_on returns False if versions differ."""
        self.assertIsNone(self.task.TASK_VERSION)
        metadata = {**self.worker_metadata, **self.task.analyze_worker()}
        metadata["samplebaseexternaltask1:version"] = 1
        self.assertFalse(self.task.can_run_on(metadata))

    def test_can_run_on_type_mismatch(self) -> None:
        """Ensure can_run_on returns False if the task/worker types mismatch."""
        metadata = {**self.worker_metadata, **self.task.analyze_worker()}
        metadata["system:worker_type"] = WorkerType.CELERY
        self.assertFalse(self.task.can_run_on(metadata))

    def test_can_run_on_architecture_not_available(self) -> None:
        """Task wants amd64, no amd64 in system:architectures."""
        metadata = {**self.worker_metadata, **self.task.analyze_worker()}
        metadata["system:architectures"] = ["arm64"]
        self.task2.data.build_architecture = "amd64"
        self.assertFalse(self.task2.can_run_on(metadata))

    def test_temporary_directory(self) -> None:
        """
        _temporary_directory() return a directory.

        The directory is deleted when the context finishes.
        """
        with tempfile.TemporaryDirectory() as basedir:
            workdir = Path(basedir)
            self.task.set_working_directory(workdir)
            with self.task._temporary_directory("foobar") as temp_dir:
                self.assertIsInstance(temp_dir, Path)
                self.assertTrue(temp_dir.is_dir())
                self.assertTrue(temp_dir.name.startswith("foobar-"))
                self.assertEqual(workdir, temp_dir.parent)

            # Ensure the temporary directory is deleted after the context is
            # exited
            self.assertFalse(temp_dir.exists())

    def test_fetch_artifact(self) -> None:
        """
        fetch_artifact() looks up and downloads a single artifact.

        Update: self._source_artifacts_ids
        """
        directory = self.create_temporary_directory()

        artifact_id = 5

        self.task.work_request_id = 1
        self.task.workspace_name = "test"
        self.task.debusine = mock.create_autospec(spec=Debusine)
        artifact_response = create_artifact_response(
            workspace=self.task.workspace_name,
            id=5,
            category=ArtifactCategory.TEST,
            created_at=dt.datetime.now(tz=dt.UTC),
            data={},
            download_tar_gz_url="https://example.com",
            files_to_upload=[],
        )
        assert isinstance(self.task.debusine, MagicMock)
        self.task.debusine.download_artifact.return_value = artifact_response

        self.assertEqual(
            self.task.fetch_artifact(artifact_id, directory), artifact_response
        )

        self.task.debusine.download_artifact.assert_called_once_with(
            artifact_id,
            directory,
            tarball=False,
            workspace=self.task.workspace_name,
        )

        self.assertEqual(self.task._source_artifacts_ids, [artifact_id])

    def test_fetch_artifact_without_configure_server_access(self) -> None:
        """fetch_artifact() asserts that self.debusine is set."""
        with self.assertRaisesRegex(AssertionError, r"^self\.debusine not set"):
            self.task.fetch_artifact(1, self.create_temporary_directory())

    def test_fetch_input_default_returns_true(self) -> None:
        """Default implementation returns True."""
        self.assertTrue(
            self.task.fetch_input(self.create_temporary_directory())
        )

    def test_check_directory_for_consistency(self) -> None:
        """Default implementation returns an empty list."""
        self.assertEqual(
            self.task.check_directory_for_consistency_errors(
                self.create_temporary_directory()
            ),
            [],
        )

    def test_upload_artifacts_default_returns(self) -> None:
        """Default implementation returns without raising an exception."""
        self.task.upload_artifacts(
            self.create_temporary_directory(),
            execution_result=WorkRequestResults.SUCCESS,
        )

    def test_execute_fetch_input_logs(self) -> None:
        """_execute(): if fetch_input() raises an exception: log it."""
        msg = "There is some exception"
        exc = Exception(msg)

        with (
            mock.patch.object(
                self.task,
                "fetch_input",
                autospec=True,
                side_effect=exc,
            ),
            mock.patch.object(self.task, "cleanup") as cleanup,
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)
            cleanup.assert_called_once()

        log = (self.task.log_directory / "fetch_input.log").read_text()

        self.assertRegex(log, f"^Exception type: Exception\nMessage: {msg}")

        traceback_str = "".join(traceback.format_exception(exc))
        self.assertIn(traceback_str, log)

    def test_execute_configure_for_execution_logs(self) -> None:
        """_execute(): if configure_for_execution() raises an exc: log it."""
        msg = "There is some exception"

        exc = Exception(msg)

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task,
                "configure_for_execution",
                autospec=True,
                side_effect=exc,
            ),
            mock.patch.object(self.task, "cleanup") as cleanup,
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)
            cleanup.assert_called_once()

        log = (
            self.task.log_directory / "configure_for_execution.log"
        ).read_text()

        self.assertRegex(log, f"^Exception type: Exception\nMessage: {msg}")

        traceback_str = "".join(traceback.format_exception(exc))
        self.assertIn(traceback_str, log)

    def test_execute_cmdline_logs(self) -> None:
        """_execute(): if run() raises an exc: log it."""
        msg = "There is some exception"

        exc = Exception(msg)

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(self.task, "run", autospec=True, side_effect=exc),
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)

        log = (self.task.log_directory / "execution.log").read_text()

        self.assertRegex(log, f"^Exception type: Exception\nMessage: {msg}")

        traceback_str = "".join(traceback.format_exception(exc))
        self.assertIn(traceback_str, log)

    def test_execute_check_directory_for_consistency_errors_logs(self) -> None:
        """_execute(): if check_directory_for_consistency_errors() fails."""
        msg = "There is some exception"

        exc = Exception(msg)

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(self.task, "run", autospec=True),
            mock.patch.object(
                self.task,
                "check_directory_for_consistency_errors",
                autospec=True,
                side_effect=exc,
            ),
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)

        log = (self.task.log_directory / "post_execution.log").read_text()

        self.assertRegex(log, f"^Exception type: Exception\nMessage: {msg}")

        traceback_str = "".join(traceback.format_exception(exc))
        self.assertIn(traceback_str, log)

    def test_execute_check_directory_for_consistency_errors_non_empty(
        self,
    ) -> None:
        """_execute(): if the output directory has consistency errors."""
        msg = "Consistency error"

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(self.task, "run", autospec=True),
            mock.patch.object(
                self.task,
                "check_directory_for_consistency_errors",
                autospec=True,
                return_value=[msg],
            ),
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)

        log = (self.task.log_directory / "consistency.log").read_text()

        self.assertRegex(log, r"^Consistency error$")

    def test_execute_upload_artifacts_logs(self) -> None:
        """_execute(): if upload_artifacts() raises an exception: log it."""
        msg = "There is some exception"

        exc = Exception(msg)

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(self.task, "run", autospec=True),
            mock.patch.object(
                self.task, "upload_artifacts", autospec=True, side_effect=exc
            ),
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)

        log = (self.task.log_directory / "post_execution.log").read_text()

        self.assertRegex(log, f"^Exception type: Exception\nMessage: {msg}")

        traceback_str = "".join(traceback.format_exception(exc))
        self.assertIn(traceback_str, log)

    def test_execute_cleanup_logs(self) -> None:
        """_execute(): if cleanup() raises an exception: log it."""
        msg = "There is some exception"

        exc = Exception(msg)

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(self.task, "run", autospec=True),
            mock.patch.object(self.task, "upload_artifacts", autospec=True),
            mock.patch.object(
                self.task, "cleanup", autospec=True, side_effect=exc
            ),
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)

        log = (self.task.log_directory / "post_execution.log").read_text()

        self.assertRegex(log, f"^Exception type: Exception\nMessage: {msg}")

        traceback_str = "".join(traceback.format_exception(exc))
        self.assertIn(traceback_str, log)

    def test_execute_logs_progress_success(self) -> None:
        """
        _execute() logs each of its main steps: success.

        For a success, there is no stages.log file.
        """
        self.task.work_request_id = 1

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(
                self.task,
                "run",
                autospec=True,
                return_value=WorkRequestResults.SUCCESS,
            ),
            mock.patch.object(self.task, "upload_artifacts", autospec=True),
            mock.patch.object(self.task, "cleanup", autospec=True),
            self.assertLogs(self.task.logger) as log,
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.SUCCESS)

        self.assertEqual(
            log.output,
            [
                "INFO:debusine.tasks:Work request 1: Fetching input",
                "INFO:debusine.tasks:Work request 1: Configuring for execution",
                "INFO:debusine.tasks:Work request 1: Preparing to run",
                "INFO:debusine.tasks:Work request 1: Running",
                "INFO:debusine.tasks:Work request 1: Checking output",
                "INFO:debusine.tasks:Work request 1: Uploading artifacts",
                "INFO:debusine.tasks:Work request 1: Cleaning up",
            ],
        )
        stages_log = self.task.log_directory / "stages.log"
        self.assertFalse(stages_log.exists())

    def test_execute_logs_progress_failure(self) -> None:
        """
        _execute() logs each of its main steps: failure.

        For a failure, there is an additional stages.log file.
        """
        self.task.work_request_id = 1

        with (
            mock.patch.object(self.task, "fetch_input", autospec=True),
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(
                self.task,
                "run",
                autospec=True,
                return_value=WorkRequestResults.FAILURE,
            ),
            mock.patch.object(self.task, "upload_artifacts", autospec=True),
            mock.patch.object(self.task, "cleanup", autospec=True),
            self.assertLogs(self.task.logger) as log,
        ):
            self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)

        self.assertEqual(
            log.output,
            [
                "INFO:debusine.tasks:Work request 1: Fetching input",
                "INFO:debusine.tasks:Work request 1: Configuring for execution",
                "INFO:debusine.tasks:Work request 1: Preparing to run",
                "INFO:debusine.tasks:Work request 1: Running",
                "INFO:debusine.tasks:Work request 1: Checking output",
                "INFO:debusine.tasks:Work request 1: Uploading artifacts",
                "INFO:debusine.tasks:Work request 1: Cleaning up",
            ],
        )
        stages_log = self.task.log_directory / "stages.log"
        stages_log_lines = []
        for line in stages_log.read_text().splitlines():
            d, message = line.split(maxsplit=1)
            dt.datetime.fromisoformat(d)
            stages_log_lines.append(message)
        self.assertEqual(
            stages_log_lines,
            [
                "Fetching input",
                "Configuring for execution",
                "Preparing to run",
                "Running",
                "Checking output",
                "Uploading artifacts",
                "Cleaning up",
            ],
        )

    def setup_upload_work_request_debug_logs(
        self, source_artifacts_ids: list[int] | None = None
    ) -> None:
        """Setup for upload_work_request_debug_logs tests."""  # noqa: D401
        self.task.debusine = mock.create_autospec(spec=Debusine)

        # Add a file to be uploaded
        with self.task.open_debug_log_file("test.log") as file:
            file.write("log")

        self.task.workspace_name = "System"
        self.task.work_request_id = 5

        if source_artifacts_ids is None:
            self.task._source_artifacts_ids = []
        else:
            self.task._source_artifacts_ids = source_artifacts_ids

    def test_upload_work_request_debug_logs_with_relation(self) -> None:
        """
        Artifact is created and uploaded. Relation is created.

        The relation is from the debug logs artifact to the source_artifact_id.
        """
        remote_artifact_id = 10
        source_artifact_id = 2

        self.setup_upload_work_request_debug_logs([source_artifact_id])

        assert isinstance(self.task.debusine, MagicMock)
        self.task.debusine.upload_artifact.return_value = MagicMock(
            id=remote_artifact_id
        )

        work_request_debug_logs_artifact = WorkRequestDebugLogs.create(
            files=self.task.log_directory.glob("*")
        )
        self.task._upload_work_request_debug_logs()

        self.task.debusine.upload_artifact.assert_called_with(
            work_request_debug_logs_artifact,
            workspace=self.task.workspace_name,
            work_request=self.task.work_request_id,
        )

        self.task.debusine.relation_create.assert_called_with(
            remote_artifact_id, source_artifact_id, "relates-to"
        )

    def test_upload_work_request_debug_logs(self) -> None:
        """Artifact is created and uploaded."""
        self.setup_upload_work_request_debug_logs()
        assert isinstance(self.task.debusine, MagicMock)

        work_request_debug_logs_artifact = WorkRequestDebugLogs.create(
            files=self.task.log_directory.glob("*")
        )

        self.task._upload_work_request_debug_logs()

        self.task.debusine.upload_artifact.assert_called_with(
            work_request_debug_logs_artifact,
            workspace=self.task.workspace_name,
            work_request=self.task.work_request_id,
        )

    def test_upload_work_request_no_log_files(self) -> None:
        """No log files: no artifact created."""
        self.task.debusine = mock.create_autospec(spec=Debusine)

        self.task._upload_work_request_debug_logs()

        assert isinstance(self.task.debusine, MagicMock)
        self.task.debusine.upload_artifact.assert_not_called()

    def test_get_input_artifacts_ids(self) -> None:
        self.task.dynamic_data = mock.Mock(spec=BaseDynamicTaskData)
        self.task.dynamic_data.get_input_artifacts_ids.return_value = [99]

        self.task.inputs = {
            "field1": ConcreteTaskInput("value1"),
            "field2": inputs.SingleInput(),
            "field3": inputs.SingleInput(),
        }

        self.task.__dict__["field2"] = ArtifactInfo(
            artifact_id=100,
            label="source",
            category=ArtifactCategory.SOURCE_PACKAGE,
            lookup="",
            data=DebianSourcePackage(
                name="hello",
                version="1.0.0",
                type="dpkg",
                dsc_fields={},
            ),
        )

        self.task.__dict__["field3"] = ArtifactInfo(
            artifact_id=101,
            label="binary",
            category=ArtifactCategory.BINARY_PACKAGE,
            lookup="",
            data=DebianBinaryPackage(
                srcpkg_name="hello",
                srcpkg_version="1.0.0",
                deb_fields={},
                deb_control_files=[],
            ),
        )

        self.assertEqual(self.task.get_input_artifacts_ids(), [99, 100, 101])


class TestsTaskConfigError(TestCase):
    """Tests for TaskConfigError."""

    def test_without_original_exception(self) -> None:
        """No original exception: pass through message."""
        msg = "This is an error message"

        error = TaskConfigError(msg)

        self.assertEqual(str(error), msg)
        self.assertIsNone(error.original_exception)

    def test_str_with_original_exception(self) -> None:
        """Message and original exception: combine both."""
        msg = "This is an error message"
        original_exc = Exception("something")

        error = TaskConfigError(msg, original_exception=original_exc)

        self.assertEqual(
            str(error), f"{msg} (Original exception: {original_exc})"
        )
        self.assertIs(error.original_exception, original_exc)

    def test_None_with_original_exception(self) -> None:
        """Original exception and no message: stringify exception."""
        original_exc = Exception("something")

        error = TaskConfigError(None, original_exception=original_exc)

        self.assertEqual(str(error), str(original_exc))
        self.assertIs(error.original_exception, original_exc)

    def test_add_parent_message(self) -> None:
        """Test add_parent_message."""
        for exc, expected in (
            (TaskConfigError(None), "test"),
            (TaskConfigError("msg"), "test: msg"),
        ):
            exc.add_parent_message("test")
            self.assertEqual(exc.args[0], expected)


class RunCommandTaskForTesting(
    RunCommandTask[BaseTaskData, BaseDynamicTaskData],
):
    """Used in TaskTests."""

    __test__ = False
    _popen: subprocess.Popen[Any] | None = None
    _pids_directory: Path | None = None

    def __init__(
        self,
        abort_after_aborted_calls: int | float = math.inf,
        wait_popen_timeouts: Iterator[int | float] | None = None,
        wait_popen_sigkill: bool = False,
        pids_directory: Path | None = None,
        send_signal_to_cmd: signal.Signals | None = None,
    ) -> None:
        """
        Initialize object used in the tests.

        This class can drive a test script that supports a simple test
        protocol, by setting pids_directory and send_signal_to_cmd. Once the
        test script is ready for execution, it creates its $PID.pid file and
        writes a single line to it. Then on the second iteration of the busy
        loop (via checks of the aborted property) send_signal_to_cmd will be
        sent to the test script, instructing it to succeed or fail.

        This replaces the use of signals to be able to integrate with
        pytest-xdist. execnet creates a separate thread that we have no
        control over, that could win a race receiving our signal, and crash the
        xdist worker.

        :param abort_after_aborted_calls: abort() method will return False
          until this number. Then will return True on each subsequent call
        :param wait_popen_timeouts: a sequence of timeouts to use for
          successive calls to _wait_popen
        :param wait_popen_sigkill: if True the first call to _wait_popen()
          will do super()._wait_popen() AND also kill the process group.
        :param pids_directory: the directory where the PID.log files are
          expected to appear.
        :param send_signal_to_cmd: When self.aborted is accessed twice, send
          this signal to the command.
        """
        super().__init__({})

        self._abort_after_aborted_calls = abort_after_aborted_calls
        self._wait_popen_timeouts = wait_popen_timeouts or itertools.repeat(
            0.001
        )
        self._wait_popen_sigkill = wait_popen_sigkill
        self._pids_directory = pids_directory
        self._send_signal_to_cmd = send_signal_to_cmd

        self._children_initialized = False
        self._popen = None

        self.aborted_call_count = 0

    @override
    @property
    def aborted(self) -> bool:
        """
        Return False (not aborted) or True.

        True when has been called more times than abort_after_aborted_calls
        parameter used in the __init__().

        If send_signal_to_cmd it will send the signal to the PID.
        """
        self.aborted_call_count += 1

        if self.aborted_call_count == 2 and self._send_signal_to_cmd:
            assert self._popen
            os.kill(self._popen.pid, self._send_signal_to_cmd)

        return self.aborted_call_count > self._abort_after_aborted_calls

    def _do_wait_for_pid_file(self, pid: int) -> None:
        """Will wait up to 5 seconds for the PID file to be written to."""
        if self._children_initialized:
            return

        assert self._pids_directory
        pid_file = self._pids_directory / f"{pid}.pid"
        # We're in a race with the test process. If we win the race, create a
        # FIFO so we can block until the test process is ready.
        try:
            os.mkfifo(pid_file)
        except OSError as e:  # pragma: no cover
            if e.errno != errno.EEXIST:
                raise
        # Open read-write so that select() will block on an empty FIFO
        with pid_file.open("r+b", buffering=0) as f:
            timeout = 5
            end_time = tm.monotonic() + timeout
            written = False
            while not written and tm.monotonic() < end_time:
                read_set, _, _ = select.select(
                    [f], [], [], end_time - tm.monotonic()
                )
                if f in read_set:  # pragma: no cover
                    # Empty the FIFO, we're only expecting a single line
                    while f.read(1) != b"\n":
                        pass
                    written = True

        if not written:  # pragma: no cover
            # No data received after the timeout. This should never
            # happen: signal-logger.sh should have sent on time.
            # (the timeout is to fail faster in case that signal-logger.sh
            # is not working)
            raise RuntimeError("PID file not written to")

        self._children_initialized = True

    def wait_cmd_zombie_process(self) -> None:
        """
        Wait and get the returncode for cmd.

        TaskMixins SIGKILLed cmd. cmd probably was in a non-interruptable
        call in the Kernel and TaskMixins did not wait and cmd is not
        left zombie. Here it collects the returncode, so it is
        not zombie and avoids a Popen.__del__ ResourceWarning
        """
        assert self._popen
        self._popen.wait(timeout=5)

    @override
    def _wait_popen(
        self,
        popen: subprocess.Popen[Any],
        timeout: float,  # noqa: U100
    ) -> int:
        # Calls TaskMixins._wait_popen() with a short timeout

        # Depending on self._wait_popen_sigkill will send a SIGKILL
        # to the process group

        self._popen = popen  # Used by wait_cmd_zombie_process()

        if self._pids_directory:
            self._do_wait_for_pid_file(popen.pid)

        try:
            return super()._wait_popen(popen, next(self._wait_popen_timeouts))
        except subprocess.TimeoutExpired as exc:
            raise exc
        finally:
            if self._wait_popen_sigkill:
                assert self._pids_directory
                pid = int(next(self._pids_directory.glob("*.pid")).stem)

                process_group = os.getpgid(pid)

                os.killpg(process_group, signal.SIGKILL)

                # If _wait_popen is called again: no killing again
                self._wait_popen_sigkill = False

    @override
    def _cmdline(self) -> list[str]:
        """Unused abstract method from RunCommandTask."""
        raise NotImplementedError()

    @override
    def get_subject(self) -> str | None:
        raise NotImplementedError()

    @override
    def compute_dynamic_data(self) -> BaseDynamicTaskData:
        raise NotImplementedError()


class RunCommandTaskTests(
    ExternalTaskHelperMixin[RunCommandTaskForTesting], TestCase
):
    """Tests for RunCommandTask methods in isolation."""

    stdout_output = "Something written to stdout"
    stderr_output = "Something written to stderr"

    @override
    def setUp(self) -> None:
        super().setUp()
        self.work_directory = self.create_temporary_directory()
        self.configure_task()

    def test_cmdline_as_root(self) -> None:
        """_cmdline_as_root return False."""
        self.assertFalse(self.task._cmdline_as_root())

    def test_cmd_env_default(self) -> None:
        """_cmd_env defaults to None."""
        self.assertIsNone(self.task._cmd_env())

    def test_task_result(self) -> None:
        """task_result() in the Mixin return SUCCESS by default."""
        self.assertEqual(
            self.task.task_result(0, self.work_directory),
            WorkRequestResults.SUCCESS,
        )

    def test_cmdline_raise_not_implemented(self) -> None:
        """Mixin raise NotImplementedError."""
        with self.assertRaises(NotImplementedError):
            self.task._cmdline()

    def test_run_cmd_run_as_root_raise_value_error(self) -> None:
        """run_cmd() raise ValueError: run_as_root requires executor."""
        with (
            self.create_temporary_file().open("wb") as cmd_log,
            self.create_temporary_file().open("wb") as out_file,
        ):
            msg = r"^run_as_root requires an executor$"
            self.assertRaisesRegex(
                ValueError,
                msg,
                self.task._run_cmd,
                ["true"],
                self.create_temporary_directory(),
                env=None,
                run_as_root=True,
                cmd_log=cmd_log,
                out_file=out_file,
            )

    def test_execute_cmdline_logs(self) -> None:
        """_execute(): if _cmdline() raise an exc: log it."""
        msg = "There is some exception"

        exc = Exception(msg)

        self.patch_task("fetch_input")
        self.patch_task("configure_for_execution")
        self.patch_task("_cmdline").side_effect = exc

        self.assertEqual(self.task._execute(), WorkRequestResults.FAILURE)

        log = (self.task.log_directory / "execution.log").read_text()

        self.assertRegex(log, f"^Exception type: Exception\nMessage: {msg}")

        traceback_str = "".join(traceback.format_exception(exc))
        self.assertIn(traceback_str, log)

    def patch_task(self, method: str) -> MagicMock:
        """Patch self.task.{method}, return its mock."""
        patcher = mock.patch.object(self.task, method, autospec=True)
        mocked = patcher.start()
        self.addCleanup(patcher.stop)
        return mocked

    def run_signal_logger(
        self, task: RunCommandTaskForTesting, options: list[str]
    ) -> int | None:
        """Run signal logger in task with options."""
        script_path = self.write_signal_logger()

        return task.run_cmd(
            [str(script_path), str(self.work_directory), *options],
            self.work_directory,
        )

    def run_success_failure(self, task: RunCommandTaskForTesting) -> int | None:
        """Run success/failure script in task."""
        script_path = self.write_success_failure()

        return task.run_cmd(
            [str(script_path), str(self.work_directory)],
            self.work_directory,
        )

    @staticmethod
    def pid_exist_not_zombie(pid: int) -> bool:
        """Return True if the pid exist and is not a zombie process."""
        try:
            status = psutil.Process(pid).status()
            # no cover: depends on the timings of the processes being terminated
            # this line in the test is never hit
            return status != "zombie"  # pragma: no cover
        except psutil.NoSuchProcess:
            return False

    def assert_processes_are_terminated(
        self, expected_number_files_to_check: int
    ) -> None:
        """
        Assert that the PIDs in self.temp_directory don't exist or a zombie.

        :param expected_number_files_to_check: number of expected processes
           (PID files) to check.
        """
        start_time = tm.time()
        time_out = 2

        files_to_check = list(self.work_directory.glob("*.pid"))
        number_of_files_to_check = len(files_to_check)

        self.assertEqual(
            expected_number_files_to_check,
            number_of_files_to_check,
            f"Expected {expected_number_files_to_check} "
            f"processes found {number_of_files_to_check}",
        )

        while True:
            alive_pids = False
            elapsed = tm.time() - start_time

            if elapsed > time_out:  # pragma: no cover
                # "no cover" because if the test pass this branch
                # is not used. Not having it the Test would be waiting
                # endless for the processes to disappear
                break

            for file in files_to_check:
                pid = int(file.stem)

                if self.pid_exist_not_zombie(pid):  # pragma: no cover
                    # Usually not triggered: the processes are killed very fast
                    # hence the no cover statement.
                    alive_pids = True

            if alive_pids:  # pragma: no cover
                # Usually not triggered because processes are killed very fast
                # hence the no cover statement.
                pass
            else:
                break  # All processes are gone

        self.assertLess(
            elapsed, time_out, "Timeout waiting for processes to disappear"
        )

    def _prepare_task(self, **kwargs: Any) -> None:
        self.task = RunCommandTaskForTesting(**kwargs)
        self.task.set_working_directory(self.work_directory)
        self.task_set_current(self.task)

    def test_execute_cmd_cancelled_killed_with_sigterm(self) -> None:
        """Execute script, kills with SIGTERM. No processes running after."""
        self._prepare_task(
            abort_after_aborted_calls=2,
            pids_directory=self.work_directory,
        )

        # Create three processes
        result = self.run_signal_logger(self.task, ["3"])

        # The processes were killed in RunCommandTask.run_cmd (the task
        # was aborted after two aborted calls)
        self.assert_processes_are_terminated(3)

        self.assertIsNone(result)

    def test_execute_cmd_cancelled_killed_with_sigkill(self) -> None:
        """Execute script, two children, killed by SIGKILL (not by SIGTERM)."""
        self._prepare_task(
            abort_after_aborted_calls=1,
            pids_directory=self.work_directory,
        )

        result = self.run_signal_logger(self.task, ["2", "TERM"])

        self.assert_processes_are_terminated(2)

        self.assertIsNone(result)

        self.task.wait_cmd_zombie_process()

    def test_execute_cmd_finished_before_killpg_sigterm(self) -> None:
        """Execute script. SIGTERM is called but processes were already dead."""
        self._prepare_task(
            abort_after_aborted_calls=1,
            wait_popen_sigkill=True,
            pids_directory=self.work_directory,
        )

        result = self.run_signal_logger(self.task, ["2"])

        self.assert_processes_are_terminated(2)

        self.assertIsNone(result)

    def test_execute_cmd_finished_before_killpg_sigkill(self) -> None:
        """Execute script. SIGKILL is called but processes were already dead."""
        self._prepare_task(
            abort_after_aborted_calls=2,
            wait_popen_timeouts=iter([0, 1, 1]),
            wait_popen_sigkill=True,
            pids_directory=self.work_directory,
        )

        result = self.run_signal_logger(self.task, ["2", "TERM"])

        self.assert_processes_are_terminated(2)

        self.assertIsNone(result)

        output_text = (
            self.task.log_directory / self.task.CMD_LOG_FILENAME
        ).read_text()

        self.assertIn(
            textwrap.dedent(
                """\
            output (contains stdout and stderr):

            aborted: True
            returncode: -9
            """
            ),
            output_text,
        )
        self.assertTrue(output_text.startswith("cmd: "))

    def test_command_finish_with_success(self) -> None:
        """Execute script. No abort, cmd return code is 0 (success)."""
        self._prepare_task(
            abort_after_aborted_calls=math.inf,
            send_signal_to_cmd=signal.SIGHUP,
            pids_directory=self.work_directory,
        )

        returncode = self.run_success_failure(self.task)

        self.assertEqual(returncode, 0)
        self.assertTrue(
            (self.task.log_directory / self.task.CMD_LOG_FILENAME)
            .read_text()
            .startswith("cmd: ")
        )

    def test_command_finish_with_failure(self) -> None:
        """Execute script. No abort, result and log file existence."""
        self._prepare_task(
            abort_after_aborted_calls=math.inf,
            send_signal_to_cmd=signal.SIGUSR2,
            pids_directory=self.work_directory,
        )

        returncode = self.run_success_failure(self.task)

        self.assertEqual(returncode, 1)
        self.assertTrue(
            (self.task.log_directory / self.task.CMD_LOG_FILENAME)
            .read_text()
            .startswith("cmd: ")
        )

    def write_success_failure(self) -> Path:
        """
        Write to self.work_directory a script that traps SIGHUP and SIGUSR2.

        For SIGHUP exits with exitcode == 0, SIGUSR2 exits with exitcode == 1.

        :return: script path
        """
        script_file = Path(self.work_directory) / "success-failure.sh"

        with script_file.open("w") as f:
            f.write(
                textwrap.dedent(
                    f"""\
                #!/bin/sh

                # Arguments:
                # $1: output directory for the log files. Must exist
                output_directory="$1"

                trap "exit 0" HUP
                trap "exit 1" USR2

                echo "{self.stdout_output}"
                echo "{self.stderr_output}" >&2

                echo Ready > "$output_directory/$$.pid"

                starttime=$(date +%s)
                timeout=10

                # Short sleep so we respond to signals fast
                while sleep 0.001
                do
                    if [ $(date +%s) -gt $((starttime + timeout)) ]
                    then
                        exit 2
                    fi
                done
                """
                )
            )

        script_file.chmod(0o700)

        return script_file

    def write_signal_logger(self) -> Path:
        """
        Write to self.work_directory a script to test finishing of processes.

        :return: script path
        """
        script_file = Path(self.work_directory) / "signal-logger.sh"

        with script_file.open("w") as f:
            f.write(
                textwrap.dedent(
                    """\
            #!/bin/sh

            # Arguments:
            # $1: output directory for the log files. Must exist
            # $2: number of generation of processes to launch
            # $3: (optional) If "TERM": traps TERM signal (logs it) and do not
            #                die

            log() {
                echo "$1" >> "$output_directory/$$.log"
            }

            output_directory="$1"
            to_spawn="$2"
            term="$3"

            if [ "$term" = "TERM" ]
            then
                trap "log SIGTERM" TERM
            fi

            if [ "$to_spawn" -gt 1 ]
            then
                to_spawn=$((to_spawn-1))
                cmd="$0 $output_directory $pid_to_notify $to_spawn $term"
                $cmd &
                while [ ! -s "$output_directory/$!.pid" ]
                    do
                    sleep 0.01
                done
                log "$$ launched $!"
            fi

            echo Ready > "$output_directory/$$.pid"

            starttime=$(date +%s)
            timeout=10

            while true
            do
                sleep 1
                if [ $(date +%s) -gt $((starttime + timeout)) ]
                then
                    exit 2
                fi
            done
            """
                )
            )

        script_file.chmod(0o700)

        return script_file

    def test_run_cmd_uses_executor(self) -> None:
        """_run_cmd() uses executor_instance to run the command."""
        cmd = ["echo", "test"]
        expected_returncode = 5

        self.task.executor_instance = MagicMock(spec=InstanceInterface)
        self.task.executor_instance.run.return_value.returncode = (
            expected_returncode
        )

        with (
            self.create_temporary_file().open("wb") as cmd_log,
            self.create_temporary_file().open("wb") as out_file,
        ):
            actual_returncode = self.task._run_cmd(
                cmd,
                self.work_directory,
                env=None,
                run_as_root=False,
                cmd_log=cmd_log,
                out_file=out_file,
            )

            self.task.executor_instance.run.assert_called_with(
                cmd,
                cwd=self.work_directory,
                env=None,
                run_as_root=self.task._cmdline_as_root(),
                stderr=cmd_log.fileno(),
                stdout=out_file.fileno(),
            )

            self.assertEqual(actual_returncode, expected_returncode)

    def test_run_cmd_large_stdout_stderr(self) -> None:
        """
        Ensure that a command generating large output is handled correctly.

        Generating more than 65536 bytes can be problematic depending on
        how subprocess.Popen's stdout/stderr is handed.
        """
        max_bytes = 70_000
        command = [
            sys.executable,
            "-u",
            "-c",
            textwrap.dedent(
                f"""\
                import sys
                for i in range({max_bytes}):
                    sys.stdout.write('o')
                    sys.stderr.write('e')
                """
            ),
        ]

        self._prepare_task()

        returncode = self.task.run_cmd(command, self.work_directory)

        self.assertEqual(returncode, 0)
        log_file = self.task.log_directory / self.task.CMD_LOG_FILENAME
        self.assertGreater(log_file.stat().st_size, max_bytes * 2)

    def test_run_cmd_invalid_utf8(self) -> None:
        """Ensure that binary output is handled correctly."""
        command = ["printf", r"\xFF\xFF"]
        self._prepare_task()
        returncode = self.task.run_cmd(
            command, self.work_directory, capture_stdout_filename="stdout.txt"
        )
        self.assertEqual(returncode, 0)
        log_file = self.task.log_directory / self.task.CMD_LOG_FILENAME
        self.assertIn(b"\xff\xff", log_file.read_bytes())

    def test_run_cmd_output_appended(self) -> None:
        """If two run commands are used: output log is appended."""
        self._prepare_task()

        self.task.run_cmd(["true"], self.work_directory)
        self.task.run_cmd(["false"], self.work_directory)

        log_file = self.task.log_directory / self.task.CMD_LOG_FILENAME
        contents = log_file.read_text()

        self.assertIn("cmd: true", contents)
        self.assertIn("cmd: false", contents)

    def test_run_cmd_capture_output(self) -> None:
        """When the parameter capture_output=True: stdout is in result.out."""
        stdout_output = "this is the output\nsecond line"
        stderr_output = "this is stderr\nsecond line"
        command = [
            "bash",
            "-c",
            f"echo -e '{stdout_output}'; echo -e '{stderr_output}' >&2",
        ]
        self._prepare_task(wait_popen_timeouts=itertools.repeat(5))

        stdout_filename = "stdout.txt"

        returncode = self.task.run_cmd(
            command,
            self.work_directory,
            capture_stdout_filename=stdout_filename,
        )

        self.assertEqual(returncode, 0)

        stdout_file = self.work_directory / stdout_filename

        self.assertEqual(stdout_file.read_text(), stdout_output + "\n")

        command_quoted = shlex.join(command)

        expected_log = (
            f"cmd: {command_quoted}\n"
            f"output (contains stderr only, stdout was captured):\n"
            f"{stderr_output}\n\n"
            f"aborted: False\n"
            f"returncode: 0\n"
            f"\n"
            f"Files in working directory:\n"
            f"stdout.txt\n"
            f"{RunCommandTask.CMD_LOG_SEPARATOR}\n"
            f"this is the output\n"
            f"second line\n"
            f"\n"
            f"{RunCommandTask.CMD_LOG_SEPARATOR}\n"
        )

        log_file = self.task.log_directory / self.task.CMD_LOG_FILENAME

        self.assertEqual(log_file.read_text(), expected_log)


class BaseTaskWithExecutorTests(
    ExternalTaskHelperMixin[SampleBaseTaskWithExecutor1], TestCase
):
    """Unit tests for BaseTaskWithExecutor class."""

    @override
    def setUp(self) -> None:
        """Create the shared attributes."""
        super().setUp()
        self.configure_task()

    def patch_executor(
        self, backend: BackendType, image_category: ExecutorImageCategory
    ) -> MagicMock:
        """Patch executor for `backend` and return its mock."""
        executor = MagicMock(spec=ExecutorInterface)
        executor.image_category = image_category
        patcher = mock.patch.dict(
            "debusine.tasks.executors.base._backends", {backend: executor}
        )
        patcher.start()
        self.addCleanup(patcher.stop)
        return executor

    def setup_task_with_mocked_executor(
        self,
        task: SampleBaseTaskWithExecutor1,
        image_category: ExecutorImageCategory,
    ) -> MagicMock:
        """
        Set up common task environment for preparing executor tests.

        Return mock of executor.
        """
        backend = BackendType.UNSHARE

        task.data.backend = backend
        self.mock_debusine(task)

        return self.patch_executor(backend, image_category)

    def test_backend_auto_default_backend(self) -> None:
        self.task.data.backend = BackendType.AUTO

        self.assertEqual(self.task.backend, self.task.DEFAULT_BACKEND)

    def test_backend_return_backend(self) -> None:
        """BaseExternalTask.backend return the backend."""
        backend = BackendType.INCUS_LXC
        self.task.data.backend = backend

        self.assertEqual(self.task.backend, backend)

    def test_prepare_executor(self) -> None:
        executor_mocked = self.setup_task_with_mocked_executor(
            self.task, ExecutorImageCategory.TARBALL
        )
        executor_instance_mock = executor_mocked.return_value
        self.task.dynamic_data = BaseDynamicTaskDataWithExecutor(
            environment_id=89
        )

        self.task._prepare_executor()

        # Executor is called with the right parameters
        executor_mocked.assert_called_with(
            self.task.debusine, 89, "testing", self.task.working_directory
        )

        # Executor's download_image() is called
        executor_instance_mock.download_image.assert_called_with()

        # self.task.executor is the executor
        self.assertIs(self.task.executor, executor_instance_mock)

    def test_prepare_executor_instance_executor_is_none(self) -> None:
        self.task.executor = None
        executor_mocked = self.setup_task_with_mocked_executor(
            self.task, ExecutorImageCategory.TARBALL
        )
        self.task.dynamic_data = BaseDynamicTaskDataWithExecutor(
            environment_id=89
        )

        # executor_mocked.create() will be called, set up mock to verify
        executor_instance_mock = MagicMock(spec=InstanceInterface)
        executor_mocked.return_value.create.return_value = (
            executor_instance_mock
        )

        self.task._prepare_executor_instance()

        executor_mocked.assert_called()
        self.assertIs(self.task.executor_instance, executor_instance_mock)
        executor_instance_mock.start.assert_called_with()

    def test_prepare_executor_instance(self) -> None:
        self.task.executor = MagicMock(spec=ExecutorInterface)
        executor_instance_mock = MagicMock(spec=InstanceInterface)
        self.task.executor.create.return_value = executor_instance_mock

        self.task._prepare_executor_instance()

        # In this case, executor_mocked should not be called
        self.task.executor.assert_not_called()

        self.assertIs(self.task.executor_instance, executor_instance_mock)
        executor_instance_mock.start.assert_called_with()

    def test_get_package_version_success(self) -> None:
        self.task.executor_instance = MagicMock(spec=InstanceInterface)
        self.task.executor_instance.run.return_value.returncode = 0
        self.task.executor_instance.run.return_value.stdout = "1.0"

        self.assertEqual(self.task.get_package_version("hello"), Version("1.0"))

        self.task.executor_instance.run.assert_called_once_with(
            ["dpkg-query", "--show", "--showformat=${Version}", "hello"],
            check=True,
            run_as_root=True,
            text=True,
        )

    def test_get_package_version_dpkg_query_error(self) -> None:
        self.task.executor_instance = MagicMock(spec=InstanceInterface)
        self.task.executor_instance.run.side_effect = (
            subprocess.CalledProcessError(
                returncode=1,
                cmd=[
                    "dpkg-query",
                    "--show",
                    "--showformat=${Version}",
                    "hello",
                ],
                output="STDOUT",
                stderr="STDERR",
            )
        )
        with self.assertRaises(subprocess.CalledProcessError):
            self.task.get_package_version("hello")

        log_file_contents = (
            self.task.log_directory / "install.log"
        ).read_text()

        self.assertEqual(
            log_file_contents,
            (
                "Failed to determine available hello version.\n"
                "command: dpkg-query --show '--showformat=${Version}' hello\n"
                "exitcode: 1\n"
                "stdout: STDOUT\n"
                "stderr: STDERR\n"
            ),
        )

    def test_execute_call_directory_push(self) -> None:
        """_execute(): self.executor is not None: call directory_push()."""
        self.task.executor_instance = MagicMock(spec=InstanceInterface)

        with (
            mock.patch.object(
                self.task, "fetch_input", autospec=True, return_value=True
            ) as fetch_input_mocked,
            mock.patch.object(
                self.task, "configure_for_execution", autospec=True
            ),
            mock.patch.object(
                self.task, "run", autospec=True, return_value=True
            ),
            mock.patch.object(self.task, "upload_artifacts", autospec=True),
        ):
            self.task._execute()

        download_directory = fetch_input_mocked.call_args[0][0]
        self.task.executor_instance.directory_push.assert_called_once_with(
            download_directory,
            download_directory.parent,
            user=self.task.executor_instance.non_root_user,
            group=self.task.executor_instance.non_root_user,
        )

    def test_execute_shuts_down_remaining_executors(self) -> None:
        """_execute() self.executor is not None: shut down executor."""
        self.task.executor_instance = MagicMock(spec=InstanceInterface)
        self.task.executor_instance.run.return_value.returncode = 0
        self.task.executor_instance.is_started.return_value = True

        with (
            mock.patch.object(
                self.task, "fetch_input", autospec=True, return_value=True
            ),
            mock.patch.object(self.task, "configure_for_execution"),
            mock.patch.object(
                self.task, "run", autospec=True, return_value=True
            ),
            mock.patch.object(self.task, "upload_artifacts", autospec=True),
        ):
            self.task._execute()

        self.task.executor_instance.stop.assert_called_once()

    def test_run_executor_command(self) -> None:
        """run_executor_command() runs the command."""
        self.task.executor_instance = MagicMock(spec=InstanceInterface)
        self.task.executor_instance.run.return_value.returncode = 0
        self.task.executor_instance.is_started.return_value = True

        self.task.run_executor_command(["true"], "prepare.log", check=True)

        self.task.executor_instance.run.assert_called_once_with(
            ["true"],
            run_as_root=False,
            stdout=mock.ANY,
            stderr=mock.ANY,
            check=True,
        )

        expected_log = "Executing: true\nExecution completed (exit code 0)\n"
        log_file = self.task.log_directory / "prepare.log"

        self.assertEqual(log_file.read_text(), expected_log)


class ExtraRepositoryMixinTests(TestCase):
    """Unit tests for BaseTaskWithExecutor class."""

    @override
    def setUp(self) -> None:
        """Create the shared attributes."""
        super().setUp()
        self.task = SampleExtraRepositoryMixin({"build_architecture": "amd64"})
        self.resolver = MockFieldResolver()
        self.directory = self.create_temporary_directory()

    def configure_extra_repositories(
        self, *extra_repositories: dict[str, Any]
    ) -> None:
        """Parse a list of dictionaries into ExtraRepository objects."""
        parsed_extra_repositories: list[ExtraRepository] = []
        for repo in extra_repositories:
            if repo.get("url"):
                parsed_extra_repositories.append(
                    ExtraExternalRepository.model_validate(repo)
                )
            else:
                parsed_extra_repositories.append(
                    ExtraDebusineRepository.model_validate(repo)
                )
        self.task.data.extra_repositories = parsed_extra_repositories
        self.resolver.field_values["extra_repositories"] = (
            parsed_extra_repositories
        )

    def set_dynamic_data(self) -> None:
        self.resolver.resolve_inputs(self.task)
        self.task.dynamic_data = self.task.compute_dynamic_data()

    def test_supports_inline_signed_by(self) -> None:
        """Test supports_inline_signed_by with 4 sample distributions."""
        self.assertFalse(self.task.supports_inline_signed_by("stretch"))
        self.assertFalse(self.task.supports_inline_signed_by("xenial"))
        self.assertTrue(self.task.supports_inline_signed_by("bookworm"))
        self.assertTrue(self.task.supports_inline_signed_by("jammy"))

    def test_write_extra_repository_config_no_keys(self) -> None:
        """Test repositories with no keys."""
        self.configure_extra_repositories(
            {
                "url": "http://example.net",
                "suite": "stretch",
                "components": ["main"],
            },
        )
        self.set_dynamic_data()

        self.task.write_extra_repository_config("stretch", self.directory)

        self.assertEqual(self.task.extra_repository_keys, [])
        self.assertEqual(len(self.task.extra_repository_sources), 1)
        self.assertEqual(
            self.task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.net/
                Suites: stretch
                Components: main
                """
            ),
        )

    def test_write_extra_repository_config_two_keys(self) -> None:
        """Test repositories with two keys."""
        self.configure_extra_repositories(
            {
                "url": "http://example.net/",
                "suite": "stretch",
                "components": ["a"],
                "signing_key": "KEY A",
            },
            {
                "url": "http://example.net/",
                "suite": "stretch/",
                "signing_key": "KEY B",
            },
        )
        self.set_dynamic_data()

        self.task.write_extra_repository_config("stretch", self.directory)

        self.assertEqual(len(self.task.extra_repository_keys), 2)
        expected_keys = ["KEY A\n", "KEY B\n"]
        for key_file, expected_key in zip(
            self.task.extra_repository_keys, expected_keys
        ):
            self.assertEqual(key_file.read_text(), expected_key)

        self.assertEqual(len(self.task.extra_repository_sources), 2)
        self.assertEqual(
            self.task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.net/
                Suites: stretch
                Components: a
                Signed-By: /etc/apt/keyrings/extra_apt_key_0.asc
                """
            ),
        )
        self.assertEqual(
            self.task.extra_repository_sources[1].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.net/
                Suites: stretch/
                Signed-By: /etc/apt/keyrings/extra_apt_key_1.asc
                """
            ),
        )

    def test_write_extra_repository_config_deb822_signed_keyring(self) -> None:
        """Test repositories in deb822 format with keys in apt < 2.3.10."""
        self.configure_extra_repositories(
            {
                "url": "http://example.net",
                "suite": "bullseye",
                "components": ["main", "non-free"],
                "signing_key": "\n".join(
                    (
                        "-----BEGIN PGP PUBLIC KEY BLOCK-----",
                        "",
                        "ABCDEFGHI",
                        "-----END PGP PUBLIC KEY BLOCK-----",
                    )
                ),
            },
        )
        self.set_dynamic_data()

        self.task.write_extra_repository_config("bullseye", self.directory)

        self.assertEqual(len(self.task.extra_repository_keys), 1)
        self.assertEqual(
            self.task.extra_repository_keys[0].read_text(),
            textwrap.dedent(
                """\
                -----BEGIN PGP PUBLIC KEY BLOCK-----

                ABCDEFGHI
                -----END PGP PUBLIC KEY BLOCK-----
                """
            ),
        )
        self.assertEqual(
            self.task.extra_repository_keys[0].name,
            "extra_apt_key_0.asc",
        )
        self.assertEqual(len(self.task.extra_repository_sources), 1)
        self.assertEqual(
            self.task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.net/
                Suites: bullseye
                Components: main non-free
                Signed-By: /etc/apt/keyrings/extra_apt_key_0.asc
                """
            ),
        )

    def test_write_extra_repository_config_deb822_signed_inline(self) -> None:
        """Test repositories in deb822 format with keys inline."""
        self.configure_extra_repositories(
            {
                "url": "http://example.net",
                "suite": "bookworm",
                "components": ["main", "non-free"],
                "signing_key": "\n".join(
                    (
                        "-----BEGIN PGP PUBLIC KEY BLOCK-----",
                        "",
                        "ABCDEFGHI",
                        "-----END PGP PUBLIC KEY BLOCK-----",
                    )
                ),
            },
        )
        self.set_dynamic_data()

        self.task.write_extra_repository_config("bookworm", self.directory)

        self.assertEqual(self.task.extra_repository_keys, [])
        self.assertEqual(len(self.task.extra_repository_sources), 1)
        self.assertEqual(
            self.task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.net/
                Suites: bookworm
                Components: main non-free
                Signed-By:
                 -----BEGIN PGP PUBLIC KEY BLOCK-----
                 .
                 ABCDEFGHI
                 -----END PGP PUBLIC KEY BLOCK-----
                """
            ),
        )

    def test_write_extra_debusine_repository_default_components(self) -> None:
        signing_key = "\n".join(
            (
                "-----BEGIN PGP PUBLIC KEY BLOCK-----",
                "",
                "ABCDEFGHI",
                "-----END PGP PUBLIC KEY BLOCK-----",
            )
        )
        self.configure_extra_repositories(
            {"suite": "bookworm-local@debian:suite"}
        )
        self.resolver.field_values["extra_repositories"] = [
            ExtraExternalRepository(
                url="http://deb.example.com/scope/workspace",
                suite="bookworm-local",
                components=["main"],
                signing_key=signing_key,
            )
        ]
        self.set_dynamic_data()

        self.task.write_extra_repository_config("bookworm", self.directory)

        self.assertEqual(self.task.extra_repository_keys, [])
        self.assertEqual(
            self.task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://deb.example.com/scope/workspace
                Suites: bookworm-local
                Components: main
                Signed-By:
                 -----BEGIN PGP PUBLIC KEY BLOCK-----
                 .
                 ABCDEFGHI
                 -----END PGP PUBLIC KEY BLOCK-----
                """
            ),
        )

    def test_write_extra_debusine_repository_explicit_components(self) -> None:
        signing_key = "\n".join(
            (
                "-----BEGIN PGP PUBLIC KEY BLOCK-----",
                "",
                "ABCDEFGHI",
                "-----END PGP PUBLIC KEY BLOCK-----",
            )
        )
        self.configure_extra_repositories(
            {
                "suite": "bookworm-local@debian:suite",
                "components": ["main", "contrib"],
            }
        )
        self.resolver.field_values["extra_repositories"] = [
            ExtraExternalRepository(
                url="http://deb.example.com/scope/workspace",
                suite="bookworm-local",
                components=["main", "contrib"],
                signing_key=signing_key,
            )
        ]
        self.set_dynamic_data()

        self.task.write_extra_repository_config("bookworm", self.directory)

        self.assertEqual(self.task.extra_repository_keys, [])
        self.assertEqual(
            self.task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://deb.example.com/scope/workspace
                Suites: bookworm-local
                Components: main contrib
                Signed-By:
                 -----BEGIN PGP PUBLIC KEY BLOCK-----
                 .
                 ABCDEFGHI
                 -----END PGP PUBLIC KEY BLOCK-----
                """
            ),
        )

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

    def test_write_extra_repository_config_architecture_filter_exclude(
        self,
    ) -> None:
        """Repos with non-matching architectures are excluded."""
        task = SampleExtraRepositoryMixin({"build_architecture": "amd64"})
        repos = [
            ExtraExternalRepository.model_validate(
                {
                    "url": "http://ports.ubuntu.com/ubuntu-ports",
                    "suite": "jammy",
                    "components": ["main"],
                    "architectures": ["arm64", "armhf"],
                }
            ),
            ExtraExternalRepository.model_validate(
                {
                    "url": "http://archive.ubuntu.com/ubuntu",
                    "suite": "jammy",
                    "components": ["main"],
                }
            ),
        ]
        task.data.extra_repositories = repos  # type: ignore[assignment]
        resolver = MockFieldResolver()
        resolver.field_values["extra_repositories"] = repos
        resolver.resolve_inputs(task)
        task.dynamic_data = task.compute_dynamic_data()

        task.write_extra_repository_config("jammy", self.directory)

        self.assertEqual(len(task.extra_repository_sources), 1)
        self.assertEqual(
            task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://archive.ubuntu.com/ubuntu
                Suites: jammy
                Components: main
                """
            ),
        )

    def test_write_extra_repository_config_architecture_filter_include(
        self,
    ) -> None:
        """Repos with matching architecture are included."""
        task = SampleExtraRepositoryMixin({"build_architecture": "arm64"})
        repos = [
            ExtraExternalRepository.model_validate(
                {
                    "url": "http://ports.ubuntu.com/ubuntu-ports",
                    "suite": "jammy",
                    "components": ["main"],
                    "architectures": ["arm64", "armhf"],
                }
            ),
        ]
        task.data.extra_repositories = repos  # type: ignore[assignment]
        resolver = MockFieldResolver()
        resolver.field_values["extra_repositories"] = repos
        resolver.resolve_inputs(task)
        task.dynamic_data = task.compute_dynamic_data()

        task.write_extra_repository_config("jammy", self.directory)

        self.assertEqual(len(task.extra_repository_sources), 1)
        self.assertEqual(
            task.extra_repository_sources[0].read_text(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://ports.ubuntu.com/ubuntu-ports
                Suites: jammy
                Components: main
                """
            ),
        )

    def test_write_extra_repository_config_no_architectures_always_included(
        self,
    ) -> None:
        """Repos without architectures field are included on any arch."""
        task = SampleExtraRepositoryMixin({"build_architecture": "riscv64"})
        repos = [
            ExtraExternalRepository.model_validate(
                {
                    "url": "http://archive.ubuntu.com/ubuntu",
                    "suite": "jammy",
                    "components": ["main"],
                }
            ),
        ]
        task.data.extra_repositories = repos  # type: ignore[assignment]
        resolver = MockFieldResolver()
        resolver.field_values["extra_repositories"] = repos
        resolver.resolve_inputs(task)
        task.dynamic_data = task.compute_dynamic_data()

        task.write_extra_repository_config("jammy", self.directory)

        self.assertEqual(len(task.extra_repository_sources), 1)
