# 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 task models."""

import datetime as dt
import re
import textwrap
from typing import get_args
from unittest import TestCase

from pydantic import ValidationError

import debusine.tasks.models as task_models
from debusine.artifacts.models import (
    ArtifactCategory,
    CollectionCategory,
    DebianBinaryPackage,
    RuntimeStatistics,
)
from debusine.tasks.models import build_lookup_string, parse_lookup_string


class LookupTests(TestCase):
    """Tests for collection item lookups."""

    def test_string(self) -> None:
        """Single-item lookups accept strings."""
        self.assertIsInstance(
            "debian@debian:archive",
            get_args(task_models.LookupSingle),
        )

    def test_integer(self) -> None:
        """Single-item lookups accept integers."""
        self.assertIsInstance(1, get_args(task_models.LookupSingle))

    def test_dict(self) -> None:
        """Multiple-item lookups accept dicts."""
        raw_lookup = {"collection": "debian@debian:archive"}
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump(),
            (
                {
                    "collection": "debian@debian:archive",
                    "child_type": "artifact",
                    "category": None,
                    "name_matcher": None,
                    "data_matchers": (),
                    "lookup_filters": (),
                    "error_on_empty": False,
                },
            ),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_bad_type(self) -> None:
        with self.assertRaisesRegex(
            ValueError, r"Expected dict, got <class 'object'>"
        ):
            task_models.LookupDict.model_validate(object())

    def test_child_type(self) -> None:
        """`child_type` is accepted."""
        raw_lookup = {
            "collection": "debian@debian:archive",
            "child_type": "collection",
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump(),
            (
                {
                    "collection": "debian@debian:archive",
                    "child_type": "collection",
                    "category": None,
                    "name_matcher": None,
                    "data_matchers": (),
                    "lookup_filters": (),
                    "error_on_empty": False,
                },
            ),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_category(self) -> None:
        """`category` is accepted."""
        raw_lookup = {
            "collection": "bookworm@debian:suite",
            "category": ArtifactCategory.SOURCE_PACKAGE,
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump(),
            (
                {
                    "collection": "bookworm@debian:suite",
                    "child_type": "artifact",
                    "category": ArtifactCategory.SOURCE_PACKAGE,
                    "name_matcher": None,
                    "data_matchers": (),
                    "lookup_filters": (),
                    "error_on_empty": False,
                },
            ),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_name_matcher(self) -> None:
        """A `name` lookup is accepted."""
        raw_lookup = {"collection": "debian@debian:archive", "name": "foo"}
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump()[0]["name_matcher"],
            {
                "kind": task_models.CollectionItemMatcherKind.EXACT,
                "value": "foo",
            },
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_name_matcher_rejects_exact(self) -> None:
        """`name__exact` is rejected."""
        with self.assertRaisesRegex(
            ValueError, "Unrecognized matcher: name__exact"
        ):
            task_models.LookupMultiple.model_validate(
                {"collection": "debian@debian:archive", "name__exact": "foo"}
            )

    def test_name_matcher_lookup(self) -> None:
        """Lookups such as `name__contains` are accepted."""
        raw_lookup = {
            "collection": "debian@debian:archive",
            "name__contains": "foo",
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump()[0]["name_matcher"],
            {
                "kind": task_models.CollectionItemMatcherKind.CONTAINS,
                "value": "foo",
            },
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_name_matcher_rejects_extra_segments(self) -> None:
        """A `name` lookup followed by two segments is rejected."""
        with self.assertRaisesRegex(
            ValueError, "Unrecognized matcher: name__foo__bar"
        ):
            task_models.LookupMultiple.model_validate(
                {"collection": "debian@debian:archive", "name__foo__bar": "foo"}
            )

    def test_name_matcher_conflict(self) -> None:
        """Conflicting name lookups are rejected."""
        with self.assertRaisesRegex(
            ValueError,
            re.escape(
                "Conflicting matchers: ['name', 'name__contains', "
                "'name__endswith', 'name__in', 'name__startswith']"
            ),
        ):
            task_models.LookupMultiple.model_validate(
                {
                    "collection": "debian@debian:archive",
                    "name": "foo",
                    "name__contains": "foo",
                    "name__endswith": "foo",
                    "name__in": ("foo", "bar"),
                    "name__startswith": "foo",
                }
            )

    def test_data_matcher(self) -> None:
        """A `data__KEY` lookup is accepted."""
        raw_lookup = {
            "collection": "debian@debian:archive",
            "data__package": "foo",
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump()[0]["data_matchers"],
            (
                (
                    "package",
                    {
                        "kind": task_models.CollectionItemMatcherKind.EXACT,
                        "value": "foo",
                    },
                ),
            ),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_data_matcher_rejects_exact(self) -> None:
        """`data__KEY__exact` is rejected."""
        with self.assertRaisesRegex(
            ValueError, "Unrecognized matcher: data__package__exact"
        ):
            task_models.LookupMultiple.model_validate(
                {
                    "collection": "debian@debian:archive",
                    "data__package__exact": "foo",
                }
            )

    def test_data_matcher_lookup(self) -> None:
        """Lookups such as `data__KEY__contains` are accepted."""
        raw_lookup = {
            "collection": "debian@debian:archive",
            "data__package__contains": "foo",
            "data__version__startswith": "1.",
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump()[0]["data_matchers"],
            (
                (
                    "package",
                    {
                        "kind": task_models.CollectionItemMatcherKind.CONTAINS,
                        "value": "foo",
                    },
                ),
                (
                    "version",
                    {
                        "kind": (
                            task_models.CollectionItemMatcherKind.STARTSWITH
                        ),
                        "value": "1.",
                    },
                ),
            ),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_data_matcher_rejects_extra_segments(self) -> None:
        """A `data` lookup followed by three segments is rejected."""
        with self.assertRaisesRegex(
            ValueError, "Unrecognized matcher: data__package__foo__bar"
        ):
            task_models.LookupMultiple.model_validate(
                {
                    "collection": "debian@debian:archive",
                    "data__package__foo__bar": "foo",
                }
            )

    def test_data_matcher_conflict(self) -> None:
        """Conflicting data lookups are rejected."""
        with self.assertRaisesRegex(
            ValueError,
            re.escape(
                "Conflicting matchers: ['data__package', "
                "'data__package__contains', 'data__package__endswith', "
                "'data__package__in', 'data__package__startswith']"
            ),
        ):
            task_models.LookupMultiple.model_validate(
                {
                    "collection": "debian@debian:archive",
                    "data__package": "foo",
                    "data__package__contains": "foo",
                    "data__package__endswith": "foo",
                    "data__package__in": ("foo", "bar"),
                    "data__package__startswith": "foo",
                }
            )

    def test_lookup_filter_single(self) -> None:
        """`lookup__KEY` with a single-lookup value is accepted."""
        subordinate_lookup = "single"
        raw_lookup = {
            "collection": "_@debian:package-build-logs",
            "lookup__foo": subordinate_lookup,
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        assert isinstance(lookup.root[0], task_models.LookupDict)
        self.assertEqual(
            lookup.model_dump()[0]["lookup_filters"],
            (("foo", subordinate_lookup),),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_lookup_filter_multiple(self) -> None:
        """`lookup__KEY` with a multiple-lookup value is accepted."""
        subordinate_lookup = ["internal@collections/name:build-amd64"]
        raw_lookup = {
            "collection": "_@debian:package-build-logs",
            "lookup__same_work_request": subordinate_lookup,
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        assert isinstance(lookup.root[0], task_models.LookupDict)
        self.assertEqual(
            lookup.root[0].lookup_filters,
            (
                (
                    "same_work_request",
                    task_models.LookupMultiple.model_validate(
                        subordinate_lookup
                    ),
                ),
            ),
        )
        self.assertEqual(
            lookup.model_dump()[0]["lookup_filters"],
            (("same_work_request", tuple(subordinate_lookup)),),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_error_on_empty(self) -> None:
        """`error_on_empty` is accepted."""
        raw_lookup = {
            "collection": "trixie@debian:suite",
            "error_on_empty": True,
        }
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump(),
            (
                {
                    "collection": "trixie@debian:suite",
                    "child_type": "artifact",
                    "category": None,
                    "name_matcher": None,
                    "data_matchers": (),
                    "lookup_filters": (),
                    "error_on_empty": True,
                },
            ),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_list(self) -> None:
        """Multiple-item lookups accept lists."""
        raw_lookup = [
            {"collection": "debian@debian:archive"},
            {"collection": "kali@debian:archive"},
            "123@artifacts",
            124,
        ]
        lookup = task_models.LookupMultiple.model_validate(raw_lookup)
        self.assertEqual(
            lookup.model_dump(),
            (
                {
                    "collection": "debian@debian:archive",
                    "child_type": "artifact",
                    "category": None,
                    "name_matcher": None,
                    "data_matchers": (),
                    "lookup_filters": (),
                    "error_on_empty": False,
                },
                {
                    "collection": "kali@debian:archive",
                    "child_type": "artifact",
                    "category": None,
                    "name_matcher": None,
                    "data_matchers": (),
                    "lookup_filters": (),
                    "error_on_empty": False,
                },
                "123@artifacts",
                124,
            ),
        )
        self.assertEqual(lookup.export(), raw_lookup)

    def test_wrong_type(self) -> None:
        """Multiple-item lookups only accept dicts or lists."""
        with self.assertRaisesRegex(
            ValueError,
            "Lookup of multiple collection items must be a dictionary or a "
            "list",
        ):
            task_models.LookupMultiple.model_validate("foo")

    def test_parse_lookup_string_segments(self) -> None:
        segments, options = parse_lookup_string("a/b/c")
        self.assertEqual(segments, ["a", "b", "c"])
        self.assertIsNone(options)

    def test_parse_lookup_string_with_options(self) -> None:
        segments, options = parse_lookup_string("//opt1=val1:opt2=val2/a/b/c")
        self.assertEqual(segments, ["a", "b", "c"])
        self.assertEqual(options, {"opt1": "val1", "opt2": "val2"})

    def test_build_lookup_string(self) -> None:
        lookup = build_lookup_string("a", "b", "c")
        self.assertEqual(lookup, "a/b/c")

    def test_build_lookup_string_with_options(self) -> None:
        options = {"opt1": "val1", "opt2": "val2"}
        lookup = build_lookup_string("a", "b", "c", options=options)
        self.assertEqual(lookup, "//opt1=val1:opt2=val2/a/b/c")

    def test_round_trip_parse_build_lookup_string(self) -> None:
        lookup = "//opt1=val1:opt2=val2/a/b/c"
        segments, options = parse_lookup_string(lookup)
        lookup2 = build_lookup_string(*segments, options=options)
        self.assertEqual(lookup, lookup2)


class ActionRetryWithDelaysTests(TestCase):
    """Tests for :py:class:`ActionRetryWithDelays`."""

    def test_delays(self) -> None:
        """A valid `delays` field is accepted."""
        delays = ["30m", "1h", "2d", "1w"]
        action = task_models.ActionRetryWithDelays(delays=delays)
        self.assertEqual(
            action.model_dump(),
            {"action": "retry-with-delays", "delays": delays},
        )

    def test_delays_bad_format(self) -> None:
        """Items in `delays` must be integers followed by m/h/d/w."""
        with self.assertRaisesRegex(
            ValueError, "String should match pattern.*input_value='2 weeks'"
        ):
            task_models.ActionRetryWithDelays(delays=["2 weeks"])


class EnumTests(TestCase):
    """Tests for SystemImageBuild task data."""

    def test_enums_str(self) -> None:
        """Test enum stringification."""
        for enum_cls in (
            task_models.AutopkgtestNeedsInternet,
            task_models.BlhcFlags,
            task_models.DebDiffFlags,
            task_models.DebootstrapVariant,
            task_models.DiskImageFormat,
            task_models.LintianFailOnSeverity,
            task_models.MmDebstrapVariant,
            task_models.SbuildBuildComponent,
            task_models.SystemBootstrapRepositoryCheckSignatureWith,
            task_models.SystemBootstrapRepositoryType,
        ):
            with self.subTest(enum_cls=repr(enum_cls)):
                for el in enum_cls:
                    with self.subTest(el=el.value):
                        self.assertEqual(str(el), el.value)


class BaseDynamicTaskDataTests(TestCase):
    """Tests for ``BaseDynamicTaskData``."""

    sample = task_models.BaseDynamicTaskData(
        parameter_summary="ps",
        subject="s",
        runtime_context="rc",
        configuration_context="cc",
        task_input_fields={"field1": "value1"},
    )

    def test_serialize(self) -> None:
        serialized = self.sample.model_dump(mode="json")
        self.assertEqual(
            serialized,
            {
                "parameter_summary": "ps",
                "subject": "s",
                "runtime_context": "rc",
                "configuration_context": "cc",
                "task_input_fields": {"field1": "value1"},
            },
        )

        deserialized = task_models.BaseDynamicTaskData.model_validate(
            serialized
        )
        self.assertEqual(deserialized, self.sample)

    def test_serialize_exclude_unset(self) -> None:
        serialized = self.sample.model_dump(mode="json", exclude_unset=True)
        self.assertEqual(
            serialized,
            {
                "parameter_summary": "ps",
                "subject": "s",
                "runtime_context": "rc",
                "configuration_context": "cc",
                "task_input_fields": {"field1": "value1"},
            },
        )

        deserialized = task_models.BaseDynamicTaskData.model_validate(
            serialized
        )
        self.assertEqual(deserialized, self.sample)


class OutputDataTests(TestCase):
    """Tests for ``OutputData``."""

    def test_merge_disjoint(self) -> None:
        """``merge`` sets all the fields explicitly set on each model."""
        left = task_models.OutputData(
            runtime_statistics=RuntimeStatistics(duration=1)
        )
        right = task_models.OutputData(
            errors=[task_models.OutputDataError(message="message", code="code")]
        )

        merged = left.merge(right)

        self.assertEqual(
            merged.model_fields_set, {"runtime_statistics", "errors"}
        )
        self.assertEqual(
            merged.runtime_statistics, RuntimeStatistics(duration=1)
        )
        self.assertEqual(
            merged.errors,
            [task_models.OutputDataError(message="message", code="code")],
        )

    def test_merge_other_wins(self) -> None:
        """``merge`` prefers values from the ``other`` model."""
        left = task_models.OutputData(
            runtime_statistics=RuntimeStatistics(duration=1)
        )
        right = task_models.OutputData(
            runtime_statistics=RuntimeStatistics(duration=1, cpu_time=1)
        )

        merged = left.merge(right)

        self.assertEqual(merged.model_fields_set, {"runtime_statistics"})
        self.assertEqual(
            merged.runtime_statistics, RuntimeStatistics(duration=1, cpu_time=1)
        )


class LintianDataTests(TestCase):
    """Tests for Lintian task data."""

    def test_input_validation(self) -> None:
        """Test LintianInput validation."""
        task_models.LintianInput(source_artifact=1)
        task_models.LintianInput(
            binary_artifacts=task_models.LookupMultiple((1,))
        )
        task_models.LintianInput(
            binary_artifacts=task_models.LookupMultiple((1, 2))
        )

        error_msg = "One of source_artifact or binary_artifacts must be set"
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.LintianInput()
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.LintianInput(
                binary_artifacts=task_models.LookupMultiple(())
            )


class LintianDynamicDataTests(TestCase):
    """Tests for :py:class:`LintianDynamicData`."""

    def test_get_source_package_name(self) -> None:
        for subject in ("foo", None):
            with self.subTest(subject=subject):
                d = task_models.LintianDynamicData(
                    input_source_artifact_id=1,
                    input_binary_artifacts_ids=[1],
                    subject=subject,
                )
                self.assertEqual(d.get_source_package_name(), subject)


class DebDiffDataTests(TestCase):
    """Tests for DebDiff task data."""

    def test_input_validation(self) -> None:
        """Test DebDiffInput validation."""
        task_models.DebDiffInput(source_artifacts=[1, 2])
        task_models.DebDiffInput(
            binary_artifacts=[
                task_models.LookupMultiple.model_validate([1]),
                task_models.LookupMultiple.model_validate([2, 3]),
            ]
        )

        error_msg = (
            "Exactly one of source_artifacts or binary_artifacts must be set"
        )
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.DebDiffInput()
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.DebDiffInput(
                source_artifacts=task_models.LookupMultiple.model_validate(
                    [1, 2]
                ),
                binary_artifacts=[
                    task_models.LookupMultiple.model_validate([1]),
                    task_models.LookupMultiple.model_validate([2]),
                ],
            )

        error_msg = "List should have at least 2 items after validation, not 1"
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.DebDiffInput(source_artifacts=[1])
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.DebDiffInput(
                binary_artifacts=[
                    task_models.LookupMultiple.model_validate([1])
                ]
            )

        error_msg = "List should have at most 2 items after validation, not 3"
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.DebDiffInput(source_artifacts=[1, 2, 3])
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.DebDiffInput(
                binary_artifacts=[
                    task_models.LookupMultiple.model_validate([1]),
                    task_models.LookupMultiple.model_validate([2]),
                    task_models.LookupMultiple.model_validate([3]),
                ]
            )


class MakeSourcePackageUploadDynamicDataTests(TestCase):
    """Tests for :py:class:`MakeSourcePackageUploadDynamicData`."""

    def test_get_source_package_name(self) -> None:
        d = task_models.MakeSourcePackageUploadDynamicData(
            environment_id=1,
            input_source_artifact_id=1,
            subject="foo",
        )
        self.assertEqual(d.get_source_package_name(), "foo")


class MergeUploadsDynamicDataTests(TestCase):
    """Tests for :py:class:`MergeUploadsDynamicData`."""

    def test_get_source_package_name(self) -> None:
        d = task_models.MergeUploadsDynamicData(
            environment_id=1,
            input_uploads_ids=[1],
            subject="foo",
        )
        self.assertEqual(d.get_source_package_name(), "foo")

    def test_input_uploads_ids_reject_duplicates(self) -> None:
        """Duplicate input_uploads_ids are rejected."""
        with self.assertRaisesRegex(ValueError, "contains duplicate items"):
            task_models.MergeUploadsDynamicData(
                environment_id=1,
                input_uploads_ids=[1, 1],
                subject="foo",
            )


class SystemBootstrapDataTests(TestCase):
    """Tests for SystemBootstrap task data."""

    def test_repository_validation(self) -> None:
        """Test SystemBootstrapRepository validation."""
        common_kwargs = {
            "mirror": "https://deb.debian.org/deb",
            "suite": "bookworm",
        }

        task_models.SystemBootstrapRepository(**common_kwargs)

        error_msg = "List should have at least 1 item"
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.SystemBootstrapRepository(types=[], **common_kwargs)

        task_models.SystemBootstrapRepository(
            check_signature_with="external",
            keyring={"url": "https://example.com/keyring_file.txt"},
            **common_kwargs,
        )

        error_msg = (
            "repository requires 'keyring':"
            " 'check_signature_with' is set to 'external'"
        )
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.SystemBootstrapRepository(
                check_signature_with="external", **common_kwargs
            )

    def test_repository_validation_duplicate_list_items(self) -> None:
        """Test SystemBootstrapRepository validation of duplicate list items."""
        common_kwargs = {
            "mirror": "https://deb.debian.org/deb",
            "suite": "bookworm",
        }

        task_models.SystemBootstrapRepository(**common_kwargs)

        error_msg = "contains duplicate items"
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.SystemBootstrapRepository(
                types=["deb", "deb"], **common_kwargs
            )
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.SystemBootstrapRepository(
                components=["main", "main"], **common_kwargs
            )

    def test_keyring_file_url_under_usr_share_keyrings(self) -> None:
        """Keyring file:// URLs under /usr/share/keyrings/ are allowed."""
        task_models.SystemBootstrapRepository(
            mirror="https://deb.debian.org/deb",
            suite="bookworm",
            check_signature_with="external",
            keyring={
                "url": "file:///usr/share/keyrings/debian-archive-keyring.gpg"
            },
        )

    def test_keyring_file_url_under_usr_local_share_keyrings(self) -> None:
        """Keyring file:// URLs under /usr/local/share/keyrings/ are allowed."""
        task_models.SystemBootstrapRepository(
            mirror="https://deb.debian.org/deb",
            suite="bookworm",
            check_signature_with="external",
            keyring={
                "url": "file:///usr/local/share/keyrings/local-keyring.gpg"
            },
        )

    def test_keyring_file_url_not_under_allowed_directory(self) -> None:
        """Keyring file:// URLs not under an allowed directory are rejected."""
        common_kwargs = {
            "mirror": "https://deb.debian.org/deb",
            "suite": "bookworm",
            "check_signature_with": "external",
        }

        error_msg = (
            "file:// URLs for keyrings must be under /usr/share/keyrings/ or "
            "/usr/local/share/keyrings/"
        )
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.SystemBootstrapRepository(
                keyring={"url": "file:///etc/passwd"}, **common_kwargs
            )
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.SystemBootstrapRepository(
                keyring={"url": "file:///usr/share/keyringssuffix"},
                **common_kwargs,
            )
        with self.assertRaisesRegex(ValueError, error_msg):
            task_models.SystemBootstrapRepository(
                keyring={"url": "file:///usr/share/keyrings/../escape"},
                **common_kwargs,
            )


class AutopkgtestDataTests(TestCase):
    """Tests for Autopkgtest task data."""

    def test_validation(self) -> None:
        """Test AutopkgtestData validation."""
        common_kwargs = {
            "input": {
                "source_artifact": 1,
                "binary_artifacts": [1, 2],
            },
            "build_architecture": "amd64",
            "environment": 1,
        }

        task_models.AutopkgtestData(**common_kwargs)

        for field in ("global", "factor", "short", "install", "test", "copy"):
            with self.subTest(field=field):
                error_msg = (
                    rf"timeout\.{field}\n"
                    rf"  Input should be greater than or equal to 0"
                )
                with self.assertRaisesRegex(ValueError, error_msg):
                    task_models.AutopkgtestData(
                        timeout={field: -1}, **common_kwargs
                    )


class AutopkgtestDynamicDataTests(TestCase):
    """Tests for :py:class:`AutopkgtestDynamicData`."""

    def test_get_source_package_name(self) -> None:
        d = task_models.AutopkgtestDynamicData(
            environment_id=1,
            input_source_artifact_id=1,
            input_binary_artifacts_ids=[1],
            subject="foo",
        )
        self.assertEqual(d.get_source_package_name(), "foo")


class BlhcDynamicDataTests(TestCase):
    """Tests for :py:class:`BlhcDynamicData`."""

    def test_get_source_package_name(self) -> None:
        d = task_models.BlhcDynamicData(
            environment_id=1,
            input_artifact_id=1,
            subject="foo",
        )
        self.assertEqual(d.get_source_package_name(), "foo")


class ExtraExternalRepositoryTests(TestCase):
    """Tests for ExtraExternalRepository."""

    def test_valid_sources(self) -> None:
        """Accept a typical valid source."""
        task_models.ExtraExternalRepository(
            url="http://deb.debian.org/debian",
            suite="bookworm",
            components=["main", "non-free-firmware"],
        )

    def test_source_with_key(self) -> None:
        """Accept a valid flat source with a key."""
        task_models.ExtraExternalRepository(
            url="http://example.com/repo",
            suite="./",
            signing_key="PUBLIC KEY",
        )

    def test_flat_source_with_empty_components_list(self) -> None:
        """Rewrite an empty components list to None."""
        repo = task_models.ExtraExternalRepository(
            url="http://example.com/repo", suite="./", components=[]
        )
        self.assertIsNone(repo.components)

    def test_file_url(self) -> None:
        """Reject a file:// URL."""
        with self.assertRaisesRegex(
            ValueError, r"Input should be a valid URL, empty host"
        ):
            task_models.ExtraExternalRepository(
                url="file:/etc/passwd", suite="bookworm", components=["main"]
            )

    def test_crazy_suite(self) -> None:
        """Reject an unreasonable suite."""
        with self.assertRaisesRegex(
            ValueError, r"String should match pattern.*input_value='b@d_suite'"
        ):
            task_models.ExtraExternalRepository(
                url="http://example.com/",
                suite="b@d_suite",
                components=["main"],
            )

    def test_flat_repository_with_component(self) -> None:
        """Reject an flat repository with a component."""
        with self.assertRaisesRegex(
            ValueError, "Components cannot be specified for a flat repository"
        ):
            task_models.ExtraExternalRepository(
                url="http://example.com/", suite="flat/", components=["main"]
            )

    def test_non_flat_repository_without_component(self) -> None:
        """Reject a non-flat repository without a component."""
        with self.assertRaisesRegex(ValueError, "Components must be specified"):
            task_models.ExtraExternalRepository(
                url="http://example.com/", suite="bookworm"
            )

    def test_crazy_component(self) -> None:
        """Reject an unreasonable component."""
        with self.assertRaisesRegex(
            ValueError, r"String should match pattern.*input_value='foo/bar'"
        ):
            task_models.ExtraExternalRepository(
                url="http://example.com/",
                suite="bookworm",
                components=["foo/bar"],
            )

    def test_as_deb822_source_flat(self) -> None:
        """Render a flat sources.list entry to a string."""
        repo = task_models.ExtraExternalRepository(
            url="http://example.com/", suite="flat/"
        )
        self.assertEqual(
            repo.as_deb822_source(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.com/
                Suites: flat/
                """
            ),
        )

    def test_as_deb822_source_components(self) -> None:
        """Render a regular sources.list entry to a string."""
        repo = task_models.ExtraExternalRepository(
            url="http://example.com/",
            suite="bookworm",
            components=["foo", "bar"],
        )
        self.assertEqual(
            repo.as_deb822_source(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.com/
                Suites: bookworm
                Components: foo bar
                """
            ),
        )

    def test_as_deb822_source_signed(self) -> None:
        """Render a signed-by sources.list entry to a string."""
        repo = task_models.ExtraExternalRepository(
            url="http://example.com/",
            suite="bookworm",
            components=["foo", "bar"],
            signing_key="\n".join(
                (
                    "-----BEGIN PGP PUBLIC KEY BLOCK-----",
                    "",
                    "ABCDEFGHI",
                    "-----END PGP PUBLIC KEY BLOCK-----",
                )
            ),
        )
        self.assertEqual(
            repo.as_deb822_source(),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.com/
                Suites: bookworm
                Components: foo bar
                Signed-By:
                 -----BEGIN PGP PUBLIC KEY BLOCK-----
                 .
                 ABCDEFGHI
                 -----END PGP PUBLIC KEY BLOCK-----
                """
            ),
        )
        self.assertEqual(
            repo.as_deb822_source(signed_by_filename="/signature.asc"),
            textwrap.dedent(
                """\
                Types: deb
                URIs: http://example.com/
                Suites: bookworm
                Components: foo bar
                Signed-By: /signature.asc
                """
            ),
        )


class SbuildDynamicDataTests(TestCase):
    """Tests for SbuildDynamicData."""

    def test_binnmu_maintainer_not_strict(self) -> None:
        """binnmu_maintainer does not have to be a deliverable email address."""
        task_models.SbuildDynamicData(
            input_source_artifact_id=1,
            binnmu_maintainer="Debusine <noreply@debusine-dev>",
        )

    def test_get_source_package_name(self) -> None:
        d = task_models.SbuildDynamicData(
            input_source_artifact_id=1,
            binnmu_maintainer="Debusine <noreply@debusine-dev>",
            subject="foo",
        )
        self.assertEqual(d.get_source_package_name(), "foo")


class ImageCacheUsageLogEntryTests(TestCase):
    """Tests for ImageCacheUsageLogEntry."""

    def test_timestamp_validation_succeeds(self) -> None:
        """Test ImageCacheUsageLogEntry accepts aware timestamp."""
        task_models.ImageCacheUsageLogEntry(
            filename="foo", timestamp=dt.datetime.now(dt.UTC)
        )

    def test_timestamp_validation_fails(self) -> None:
        """Test ImageCacheUsageLogEntry rejects naive timestamp."""
        with self.assertRaisesRegex(ValueError, "timestamp is TZ-naive"):
            task_models.ImageCacheUsageLogEntry(
                filename="foo",
                timestamp=dt.datetime.now(),  # noqa: DTZ005
            )


class ImageCacheUsageLogTests(TestCase):
    """Tests for ImageCacheUsageLog."""

    def test_version_validation_succeeds(self) -> None:
        """Test ImageCacheUsageLog accepts version 1."""
        task_models.ImageCacheUsageLog(version=1)

    def test_version_validation_fails(self) -> None:
        """Test ImageCacheUsageLog rejects version 99."""
        with self.assertRaisesRegex(ValueError, "Unknown usage log version 99"):
            task_models.ImageCacheUsageLog(version=99)


class ExtractForSigningDataTests(TestCase):
    """Tests for ExtractForSigningData."""

    def test_backend_validation_succeeds(self) -> None:
        """Test ExtractForSigningData accepts no backend or backend=auto."""
        task_models.ExtractForSigningData(
            environment=1,
            input={"template_artifact": 2, "binary_artifacts": [3]},
        )
        task_models.ExtractForSigningData(
            environment=1,
            backend=task_models.BackendType.AUTO,
            input={"template_artifact": 2, "binary_artifacts": [3]},
        )

    def test_backend_validation_fails(self) -> None:
        """Test ExtractForSigningData rejects backends other than auto."""
        with self.assertRaisesRegex(
            ValueError,
            'ExtractForSigning only accepts backend "auto", not "qemu"',
        ):
            task_models.ExtractForSigningData(
                environment=1,
                backend=task_models.BackendType.QEMU,
                input={"template_artifact": 2, "binary_artifacts": [3]},
            )


class ExtractForSigningDynamicDataTests(TestCase):
    """Tests for :py:class:`ExtractForSigningDynamicData`."""

    def test_get_source_package_name(self) -> None:
        d = task_models.ExtractForSigningDynamicData(
            environment_id=1,
            input_template_artifact_id=1,
            input_binary_artifacts_ids=[1],
            subject="foo",
        )
        self.assertEqual(d.get_source_package_name(), None)


class AssembleSignedSourceDataTests(TestCase):
    """Tests for AssembleSignedSourceData."""

    def test_backend_validation_succeeds(self) -> None:
        """Test AssembleSignedSourceData accepts no backend or backend=auto."""
        task_models.AssembleSignedSourceData(
            environment=1, template=2, signed=[3]
        )
        task_models.AssembleSignedSourceData(
            environment=1,
            backend=task_models.BackendType.AUTO,
            template=2,
            signed=[3],
        )

    def test_backend_validation_fails(self) -> None:
        """Test AssembleSignedSourceData rejects backends other than auto."""
        with self.assertRaisesRegex(
            ValueError,
            'AssembleSignedSource only accepts backend "auto", not "qemu"',
        ):
            task_models.AssembleSignedSourceData(
                environment=1,
                backend=task_models.BackendType.QEMU,
                template=2,
                signed=[3],
            )


class AssembleSignedSourceDynamicDataTests(TestCase):
    """Tests for :py:class:`AssembleSignedSourceDynamicData."""

    def test_get_source_package_name(self) -> None:
        d = task_models.AssembleSignedSourceDynamicData(
            environment_id=1,
            template_id=1,
            signed_ids=[1],
            unsigned_ids=[1],
            subject="foo",
        )
        # In this case, subject is a binary package name, not a source package
        # name
        self.assertIsNone(d.get_source_package_name())


class EventReactionsTests(TestCase):
    """Tests for EventReactions."""

    def test_serialize_exclude_unset(self) -> None:
        update_collection_with_artifacts = (
            task_models.ActionUpdateCollectionWithArtifacts(
                collection="internal@collections",
                name_template="test",
                artifact_filters={"category": ArtifactCategory.TEST},
            )
        )
        value = task_models.EventReactions(
            on_success=[update_collection_with_artifacts]
        )
        self.assertEqual(
            value.model_dump(exclude_unset=True),
            {
                "on_success": [
                    update_collection_with_artifacts.model_dump(
                        exclude_unset=True
                    )
                ],
            },
        )

        # Try deserializing the serialized version
        value1 = task_models.EventReactions(
            **value.model_dump(exclude_unset=True)
        )
        self.assertEqual(value1, value)

    def test_add_reaction(self) -> None:
        update_collection_with_artifacts = (
            task_models.ActionUpdateCollectionWithArtifacts(
                collection="internal@collections",
                name_template="test",
                artifact_filters={"category": ArtifactCategory.TEST},
            )
        )

        value = task_models.EventReactions()
        value.add_reaction("on_success", update_collection_with_artifacts)

        self.assertEqual(
            value.model_dump(exclude_unset=True),
            {
                "on_success": [
                    update_collection_with_artifacts.model_dump(
                        exclude_unset=True
                    )
                ],
            },
        )

    def test_deserialize(self) -> None:
        """Test basic deserialization."""
        task_models.EventReactions.model_validate_json(
            """
            {
                "on_success": [
                    {
                        "action": "update-collection-with-artifacts",
                        "collection": "internal@collections",
                        "artifact_filters": {
                            "category": "debian:binary-package",
                            "data__deb_fields__Section": "python"
                        },
                        "name_template": "{package}_{version}",
                        "variables": {
                            "$package": "deb_fields.Package",
                            "$version": "deb_fields.Version"
                        }
                    }
                ],
                "on_failure": [
                    {
                        "action": "send-notification",
                        "channel": "admin-team",
                        "data": {
                            "cc": [ "qa-team@example.org " ],
                            "subject": "Work request ${work_request_id}"
                        }
                    },
                    {
                        "action": "send-notification",
                        "channel": "security-team"
                    }
                ]
            }
        """
        )


class DebDiffDynamicDataTests(TestCase):
    """Tests for DebDiffDynamicData."""

    def test_source_artifacts_and_binary_artifacts_set(self) -> None:
        with self.assertRaisesRegex(
            ValidationError,
            re.escape(
                "Only one of 'input_source_artifacts_ids' or "
                "'input_binary_artifacts_ids' may be set (not both)."
            ),
        ):
            task_models.DebDiffDynamicData(
                input_source_artifacts_ids=[1, 2],
                input_binary_artifacts_ids=[[3], [4]],
                environment_id=3,
            )

    def test_source_artifacts_and_binary_artifacts_none(self) -> None:
        # Does not raise
        task_models.DebDiffDynamicData(
            input_source_artifacts_ids=None,
            input_binary_artifacts_ids=None,
            environment_id=3,
        )

    def test_source_artifacts_only(self) -> None:
        task_models.DebDiffDynamicData(
            input_source_artifacts_ids=[1, 2], environment_id=3
        )

    def test_binary_artifacts_only(self) -> None:
        # Does not raise
        task_models.DebDiffDynamicData(
            input_binary_artifacts_ids=[[1, 2, 3], [10, 20, 30]],
            environment_id=3,
        )

    def test_get_source_package_name(self) -> None:
        for subject, expected in (
            ("foo", None),
            ("source:foo", "foo"),
            ("binary:foo", None),
        ):
            with self.subTest(subject=subject):
                d = task_models.DebDiffDynamicData(
                    environment_id=1,
                    subject=subject,
                )
                self.assertEqual(d.get_source_package_name(), expected)


class BaseTaskInputArtifactTests(TestCase):
    """Tests for :py:class:`BaseTaskInputArtifact`."""

    data = DebianBinaryPackage(
        srcpkg_name="unsigned",
        srcpkg_version="1.0",
        deb_fields={
            "Package": "hello",
            "Version": "1.0-1",
            "Architecture": "amd64",
        },
        deb_control_files=[],
    )

    def test_serialize_data_none(self) -> None:
        value = task_models.BaseTaskInputArtifact(
            artifact_id=1, category=ArtifactCategory.BINARY_PACKAGE, data=None
        )
        self.assertEqual(
            value.model_dump(),
            {
                "artifact_id": 1,
                "category": "debian:binary-package",
                "data": None,
            },
        )

    def test_serialize_category_none(self) -> None:
        with self.assertRaisesRegex(
            ValueError, "Category must be present if data is supplied"
        ):
            task_models.BaseTaskInputArtifact(
                artifact_id=1, category=None, data=self.data
            )

    def test_serialize_invalid_category(self) -> None:
        with self.assertRaisesRegex(
            ValueError, "Category 'foo' is not supported"
        ):
            task_models.BaseTaskInputArtifact(
                artifact_id=1, category="foo", data=self.data
            )

    def test_serialize_category_data_none(self) -> None:
        value = task_models.BaseTaskInputArtifact(
            artifact_id=1, category=None, data=None
        )
        self.assertEqual(
            value.model_dump(),
            {"artifact_id": 1, "category": None, "data": None},
        )

    def test_deserialize_data_none(self) -> None:
        serialized = {
            "artifact_id": 1,
            "category": "debian:binary-package",
            "data": None,
        }
        deserialized = task_models.BaseTaskInputArtifact.model_validate(
            serialized
        )
        self.assertEqual(
            deserialized,
            task_models.BaseTaskInputArtifact(
                artifact_id=1,
                category=ArtifactCategory.BINARY_PACKAGE,
                data=None,
            ),
        )

    def test_deserialize_category_none(self) -> None:
        serialized = {
            "artifact_id": 1,
            "category": None,
            "data": self.data,
        }
        with self.assertRaisesRegex(
            ValueError, "Category must be present if data is supplied"
        ):
            task_models.BaseTaskInputArtifact.model_validate(serialized)

    def test_deserialize_category_data_none(self) -> None:
        serialized = {"artifact_id": 1, "category": None, "data": None}
        deserialized = task_models.BaseTaskInputArtifact.model_validate(
            serialized
        )
        self.assertEqual(
            deserialized,
            task_models.BaseTaskInputArtifact(
                artifact_id=1,
                category=None,
                data=None,
            ),
        )


class InputArtifactSingleTests(TestCase):
    """Tests for :py:class:`InputArtifactSingle`."""

    value = task_models.InputArtifactSingle(
        lookup="test",
        label="field",
        artifact_id=1,
        category=ArtifactCategory.BINARY_PACKAGE,
        data=DebianBinaryPackage(
            srcpkg_name="unsigned",
            srcpkg_version="1.0",
            deb_fields={
                "Package": "hello",
                "Version": "1.0-1",
                "Architecture": "amd64",
            },
            deb_control_files=[],
        ),
    )

    def test_serialize(self) -> None:
        assert self.value.data is not None
        self.assertEqual(
            self.value.model_dump(),
            {
                "lookup": "test",
                "label": "field",
                "artifact_id": 1,
                "category": "debian:binary-package",
                "data": self.value.data.model_dump(),
            },
        )

    def test_deserialize(self) -> None:
        serialized = self.value.model_dump()
        deserialized = task_models.InputArtifactSingle.model_validate(
            serialized
        )
        self.assertEqual(deserialized, self.value)


class InputArtifactMultipleTests(TestCase):
    """Tests for :py:class:`InputArtifactMultiple`."""

    value = task_models.InputArtifactMultiple(
        lookup=["test"],
        label="field",
        artifacts=[
            task_models.BaseTaskInputArtifact(
                artifact_id=1,
                category=ArtifactCategory.BINARY_PACKAGE,
                data=DebianBinaryPackage(
                    srcpkg_name="unsigned",
                    srcpkg_version="1.0",
                    deb_fields={
                        "Package": "hello",
                        "Version": "1.0-1",
                        "Architecture": "amd64",
                    },
                    deb_control_files=[],
                ),
            )
        ],
    )

    def test_serialize(self) -> None:
        self.assertEqual(
            self.value.model_dump(mode="json"),
            {
                "lookup": ["test"],
                "label": "field",
                "artifacts": [
                    self.value.artifacts[0].model_dump(),
                ],
            },
        )

    def test_deserialize(self) -> None:
        serialized = self.value.model_dump()
        deserialized = task_models.InputArtifactMultiple.model_validate(
            serialized
        )
        self.assertEqual(deserialized, self.value)


class InputCollectionSingleTests(TestCase):
    """Tests for :py:class:`InputCollectionSingle`."""

    value = task_models.InputCollectionSingle(
        lookup="test",
        label="field",
        collection_id=1,
        scope_name="scope",
        workspace_name="workspace",
        category=CollectionCategory.SUITE,
        name="name",
        data={"present": True},
    )

    def test_serialize(self) -> None:
        self.assertEqual(
            self.value.model_dump(mode="json"),
            {
                "lookup": "test",
                "label": "field",
                "collection_id": 1,
                "category": "debian:suite",
                "scope_name": "scope",
                "workspace_name": "workspace",
                "name": "name",
                "data": {"present": True},
            },
        )

    def test_deserialize(self) -> None:
        serialized = self.value.model_dump()
        deserialized = task_models.InputCollectionSingle.model_validate(
            serialized
        )
        self.assertEqual(deserialized, self.value)
