# 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.

"""Test views for package archives."""

import base64
import datetime as dt
import hashlib
from abc import ABC
from collections.abc import Mapping
from contextlib import ExitStack
from operator import attrgetter
from pathlib import PurePath
from typing import Any, ClassVar, overload, override
from unittest import mock

import lxml
from debian.copyright import parse_multiline
from debian.deb822 import Deb822
from django.conf import settings
from django.test import override_settings
from django.utils import timezone
from django.utils.cache import has_vary_header
from pgpy import PGPKey, PGPKeyring
from pgpy.constants import EllipticCurveOID, PubKeyAlgorithm
from rest_framework import HTTP_HEADER_ENCODING, status

from debusine.artifacts.models import ArtifactCategory, CollectionCategory
from debusine.assets.models import KeyPurpose
from debusine.db.models import (
    ArtifactRelation,
    Collection,
    CollectionRelation,
    Token,
    Workspace,
)
from debusine.db.playground import scenarios
from debusine.test.django import TestCase, TestResponseType
from debusine.web.archives.views import RenderableSuite, SigningKeysResponse
from debusine.web.views.tests.utils import ViewTestMixin


class SigningKeysResponseTests(TestCase):
    """Tests for :py:class:`SigningKeysResponse`."""

    def test_render(self) -> None:
        keys = [
            PGPKey.new(PubKeyAlgorithm.EdDSA, EllipticCurveOID.Ed25519).pubkey
            for _ in range(2)
        ]
        response = SigningKeysResponse([bytes(key) for key in keys])
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            response.headers["Content-Type"],
            "application/pgp-keys; charset=utf-8",
        )
        assert response.content.startswith(
            b"-----BEGIN PGP PUBLIC KEY BLOCK-----\n"
        )
        self.assertEqual(
            PGPKeyring(response.content).fingerprints(),
            {key.fingerprint for key in keys},
        )


class ArchiveViewTests(ViewTestMixin, TestCase, ABC):
    """Helper methods for testing archive views."""

    playground_memory_file_store = False
    scenario = scenarios.DefaultContext()

    archive: ClassVar[Collection]
    suites: ClassVar[dict[str, Collection]]

    # Path used for some tests of generic behaviour in each subclass.
    example_path: str
    # Is the example path scoped to a suite?
    expect_in_suite: bool
    # Are paths tested here normally mutable?
    expect_mutable_no_archive: bool
    expect_mutable: bool
    # What Content-Type does this endpoint return on success?
    expect_content_type: str = "text/html; charset=utf-8"

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.archive = cls.scenario.workspace.get_singleton_collection(
            pc=cls.scenario.context_ui(), category=CollectionCategory.ARCHIVE
        )
        cls.suites = {
            name: cls.playground.create_collection(
                name, CollectionCategory.SUITE, workspace=cls.scenario.workspace
            )
            for name in ("bookworm", "trixie")
        }

    @overload
    def format_snapshot(self, snapshot: None) -> None: ...

    @overload
    def format_snapshot(self, snapshot: dt.datetime) -> str: ...

    def format_snapshot(self, snapshot: dt.datetime | None) -> str | None:
        """Format a snapshot timestamp as a URL segment."""
        return None if snapshot is None else snapshot.strftime("%Y%m%dT%H%M%SZ")

    def archive_file_get(
        self,
        *,
        workspace_name: str | None = None,
        snapshot: dt.datetime | None = None,
        path: str,
        headers: Mapping[str, Any] | None = None,
        token: Token | None = None,
    ) -> TestResponseType:
        """Get a file from an archive."""
        with ExitStack() as stack:
            fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_PRIMARY_FQDN
            scope_name = self.scenario.scope.name
            if workspace_name is None:
                workspace_name = self.scenario.workspace.name
            url = f"/{scope_name}/{workspace_name}"
            if snapshot is not None:
                url += "/" + self.format_snapshot(snapshot)
            url += f"/{path}"
            if token is not None:
                headers = {**(headers or {})}
                if token.user is not None:
                    token_username = token.user.username
                else:
                    assert token.external_username
                    token_username = token.external_username
                credentials = f"{token_username}:{token.key}"
                base64_credentials = base64.b64encode(
                    credentials.encode(HTTP_HEADER_ENCODING)
                ).decode("ascii")
                headers["Authorization"] = "Basic " + base64_credentials
                stack.enter_context(
                    mock.patch(
                        "debusine.server.middlewares.scopes."
                        "AuthorizationMiddleware._basic_auth_allowed",
                        return_value=True,
                    )
                )
            return self.client.get(url, HTTP_HOST=fqdn, headers=headers)

    def assert_has_caching_headers(
        self, response: TestResponseType, *, mutable: bool
    ) -> None:
        self.assertEqual(
            response.headers["Cache-Control"],
            "max-age=1800, proxy-revalidate" if mutable else "max-age=31536000",
        )
        self.assertTrue(has_vary_header(response, "Authorization"))

    def assertResponse404(
        self, response: TestResponseType, message: str
    ) -> None:
        """Check that the response is a 404 with the given message."""
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
        self.assertEqual(response.context["exception"], message)

    def test_workspace_not_found(self) -> None:
        for snapshot in (None, timezone.now()):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    workspace_name="nonexistent",
                    snapshot=snapshot,
                    path=self.example_path,
                )

                self.assertResponse404(
                    response,
                    "Workspace nonexistent not found in scope debusine,"
                    " or you are not authorized to see it",
                )
                self.assert_has_caching_headers(
                    response,
                    mutable=self.expect_mutable_no_archive and snapshot is None,
                )

    def test_archive_not_found(self) -> None:
        self.archive.child_items.all().delete()
        self.archive.delete()

        for snapshot in (None, timezone.now()):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    snapshot=snapshot, path=self.example_path
                )

                self.assertResponse404(
                    response,
                    f"No {CollectionCategory.ARCHIVE} collection found in "
                    f"{self.scenario.workspace}",
                )
                self.assert_has_caching_headers(
                    response,
                    mutable=self.expect_mutable_no_archive and snapshot is None,
                )

    def test_private_workspace_not_accessible(self) -> None:
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        response = self.archive_file_get(path=self.example_path)
        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)

    def test_private_workspace_logged_in(self) -> None:
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        token = self.scenario.create_user_token()
        with self.scenario.assign_role(
            self.scenario.workspace, Workspace.Roles.CONTRIBUTOR
        ):
            response = self.archive_file_get(
                path=self.example_path, token=token
            )
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            response.headers["Content-Type"], self.expect_content_type
        )
        if self.expect_content_type.startswith("text/html"):
            self.assertHTMLValid(response.content)

    def test_private_workspace_external_user_token_grants_archive(self) -> None:
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        token = self.playground.create_external_user_token(username="test")
        self.archive.grant_role(
            token, Collection.Roles.VIEWER, created_by=self.scenario.user
        )
        response = self.archive_file_get(path=self.example_path, token=token)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            response.headers["Content-Type"], self.expect_content_type
        )
        if self.expect_content_type.startswith("text/html"):
            self.assertHTMLValid(response.content)

    def test_private_workspace_external_user_token_grants_suite(self) -> None:
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        token = self.playground.create_external_user_token(username="test")
        self.suites["bookworm"].grant_role(
            token, Collection.Roles.VIEWER, created_by=self.scenario.user
        )
        response = self.archive_file_get(path=self.example_path, token=token)
        if self.expect_in_suite:
            self.assertEqual(response.status_code, status.HTTP_200_OK)
            self.assertEqual(
                response.headers["Content-Type"], self.expect_content_type
            )
            if self.expect_content_type.startswith("text/html"):
                self.assertHTMLValid(response.content)
        else:
            self.assertResponse404(
                response,
                f"Workspace {self.scenario.workspace.name} not found in scope"
                f" {self.scenario.scope}, or you are not authorized to see it",
            )

    def test_private_workspace_external_user_token_grants_wrong_suite(
        self,
    ) -> None:
        self.scenario.workspace.public = False
        self.scenario.workspace.save()
        token = self.playground.create_external_user_token(username="test")
        self.suites["trixie"].grant_role(
            token, Collection.Roles.VIEWER, created_by=self.scenario.user
        )
        response = self.archive_file_get(path=self.example_path, token=token)
        self.assertResponse404(
            response,
            f"Workspace {self.scenario.workspace.name} not found in scope"
            f" {self.scenario.scope}, or you are not authorized to see it",
        )


class ArchiveFileViewTests(ArchiveViewTests, ABC):
    """Helper methods for testing archive file views."""

    # A path that exercises the view under test, but that is not expected to
    # exist.
    example_path_nonexistent: str

    def assert_response_ok(
        self, response: TestResponseType, *, contents: bytes, mutable: bool
    ) -> None:
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            b"".join(getattr(response, "streaming_content")), contents
        )
        self.assert_has_caching_headers(response, mutable=mutable)

    def test_file_not_found(self) -> None:
        for snapshot in (None, timezone.now()):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    snapshot=snapshot, path=self.example_path_nonexistent
                )

                self.assertResponse404(
                    response, "No FileInArtifact matches the given query."
                )
                self.assert_has_caching_headers(
                    response, mutable=self.expect_mutable and snapshot is None
                )


class ArchiveRootViewTests(ArchiveViewTests):
    """Test retrieving the root of an archive."""

    example_path = ""
    expect_in_suite = False
    expect_mutable_no_archive = True
    expect_mutable = True

    description_path = "//div[@id='workspace-description']"

    def test_current(self) -> None:
        self.playground.create_collection(
            "not-exported",
            CollectionCategory.SUITE,
            workspace=self.scenario.workspace,
            data={"exported": False},
        )

        response = self.archive_file_get(path="")

        tree = self.assertResponseHTML(response)
        archive_suites = tree.xpath("//div[@id='archive-suites']")[0]
        self.assertTextContentEqual(
            archive_suites.p, "This archive contains the following suites:"
        )
        self.assertEqual(len(archive_suites.ul.li), len(self.suites))
        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_FQDN
        workspace = self.scenario.workspace
        for li, suite in zip(
            archive_suites.ul.li,
            sorted(self.suites.values(), key=attrgetter("name")),
        ):
            self.assertTextContentEqual(li.p, f"{suite.name}: ")
            self.assertEqual(
                Deb822("".join(li.pre.itertext())),
                {
                    "Types": "deb",
                    "URIs": (
                        f"http://{fqdn}/{workspace.scope.name}/{workspace.name}"
                    ),
                    "Suites": suite.name,
                },
            )
        self.assert_has_caching_headers(response, mutable=True)

        self.assertFalse(tree.xpath(self.description_path))

    def test_current_signing_keys(self) -> None:
        keys = [
            PGPKey.new(PubKeyAlgorithm.EdDSA, EllipticCurveOID.Ed25519).pubkey
            for _ in range(len(self.suites))
        ]
        sorted_suites = sorted(self.suites.values(), key=attrgetter("name"))
        for suite, key in zip(sorted_suites, keys):
            self.playground.create_asset_usage(
                resource=self.playground.create_signing_key_asset(
                    purpose=KeyPurpose.OPENPGP_REPOSITORY,
                    fingerprint=key.fingerprint,
                    public_key=str(key),
                    workspace=suite.workspace,
                ),
                workspace=suite.workspace,
            )
            suite.data["signing_keys"] = [key.fingerprint]
            suite.save()

        response = self.archive_file_get(path="")

        tree = self.assertResponseHTML(response)
        archive_suites = tree.xpath("//div[@id='archive-suites']")[0]
        self.assertTextContentEqual(
            archive_suites.p, "This archive contains the following suites:"
        )
        self.assertEqual(len(archive_suites.ul.li), len(self.suites))
        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_FQDN
        workspace = self.scenario.workspace
        for li, suite, key in zip(archive_suites.ul.li, sorted_suites, keys):
            self.assertTextContentEqual(li.p, f"{suite.name}: ")
            suite_deb822 = Deb822("".join(li.pre.itertext()))
            self.assertCountEqual(
                suite_deb822.keys(), ["Types", "URIs", "Suites", "Signed-By"]
            )
            self.assertDictContainsAll(
                suite_deb822,
                {
                    "Types": "deb",
                    "URIs": (
                        f"http://{fqdn}/{workspace.scope.name}/{workspace.name}"
                    ),
                    "Suites": suite.name,
                },
            )
            self.assertEqual(
                PGPKeyring(
                    parse_multiline(suite_deb822["Signed-By"])
                ).fingerprints(),
                {key.fingerprint},
            )
        self.assert_has_caching_headers(response, mutable=True)

    def test_snapshot(self) -> None:
        now = timezone.now()
        one_month_ago = (now - dt.timedelta(days=30)).replace(microsecond=0)

        response = self.archive_file_get(snapshot=one_month_ago, path="")

        # We always list all exported suites in the workspace, regardless of
        # when they were added to the archive, so we don't bother testing
        # different snapshot times here.
        tree = self.assertResponseHTML(response)
        archive_suites = tree.xpath("//div[@id='archive-suites']")[0]
        self.assertTextContentEqual(
            archive_suites.p, "This archive contains the following suites:"
        )
        self.assertEqual(len(archive_suites.ul.li), len(self.suites))
        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_FQDN
        workspace = self.scenario.workspace
        for li, suite in zip(
            archive_suites.ul.li,
            sorted(self.suites.values(), key=attrgetter("name")),
        ):
            self.assertTextContentEqual(li.p, f"{suite.name}: ")
            self.assertEqual(
                Deb822("".join(li.pre.itertext())),
                {
                    "Types": "deb",
                    "URIs": (
                        f"http://{fqdn}/{workspace.scope.name}/{workspace.name}"
                    ),
                    "Suites": suite.name,
                    "Snapshot": self.format_snapshot(one_month_ago),
                },
            )
        self.assert_has_caching_headers(response, mutable=False)

    def test_archive_empty(self) -> None:
        self.archive.child_items.all().delete()
        for suite in self.suites.values():
            suite.delete()

        for snapshot in (None, timezone.now()):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(snapshot=snapshot, path="")

                tree = self.assertResponseHTML(response)
                archive_suites = tree.xpath("//div[@id='archive-suites']")[0]
                self.assertTextContentEqual(
                    archive_suites, "This archive is empty."
                )
                self.assert_has_caching_headers(
                    response, mutable=snapshot is None
                )

    def test_description(self) -> None:
        workspace = self.playground.get_default_workspace()
        workspace.description = "This is **description**"
        workspace.save()

        response = self.archive_file_get(path="")

        tree = self.assertResponseHTML(response)

        description_el = self.assertHasElement(tree, self.description_path)

        description_html = lxml.etree.tostring(description_el)
        self.assertIn(b"<strong>description</strong>", description_html)

    def test_get_place(self) -> None:
        response = self.archive_file_get(path="")
        assert response.context_data is not None

        place = response.context_data["place"]
        workspace = self.scenario.workspace

        self.assertPlaceValueEqual(
            place.title, "Suites", page=f"Suites in <code>{workspace}</code>"
        )

    @override_settings(
        ALLOWED_HOSTS=["*"],
        DEBUSINE_DEBIAN_ARCHIVE_FQDN=["deb.example.com", "deb.example.org"],
        DEBUSINE_DEBIAN_ARCHIVE_PRIMARY_FQDN="deb.example.com",
    )
    def test_override_archive_fqdn(self) -> None:
        response = self.archive_file_get(path="")

        tree = self.assertResponseHTML(response)
        archive_suites = tree.xpath("//div[@id='archive-suites']")[0]
        workspace = self.scenario.workspace
        for li in archive_suites.ul.li:
            self.assertEqual(
                Deb822("".join(li.pre.itertext()))["URIs"],
                f"http://deb.example.com/"
                f"{workspace.scope.name}/{workspace.name}",
            )


class SuiteRootViewTests(ArchiveViewTests):
    """Test retrieving the root of a suite."""

    example_path = "dists/bookworm/"
    expect_in_suite = True
    expect_mutable_no_archive = True
    expect_mutable = True

    def test_current(self) -> None:
        suite = self.suites["bookworm"]
        suite.data["components"] = ["main", "contrib", "non-free"]
        suite.data["architectures"] = ["amd64"]
        suite.data["release_fields"] = {"Description": "Something special"}
        suite.save()

        response = self.archive_file_get(path="dists/bookworm/")

        tree = self.assertResponseHTML(response)
        self.assertTextContentEqual(
            tree.xpath("//div[@id='suite-summary']")[0],
            "bookworm Something special",
        )

        self.assertTextContentEqual(
            tree.xpath("//div[@id='suite-architectures']")[0],
            "Architectures amd64",
        )

        suite_release_fields = tree.xpath("//div[@id='suite-release-fields']")[
            0
        ]
        self.assertEqual(
            [
                (element.tag, element.text.strip())
                for element in suite_release_fields.dl.iterchildren()
            ],
            [("dt", "Description"), ("dd", "Something special")],
        )
        suite_apt = tree.xpath("//div[@id='suite-apt']")[0]
        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_FQDN
        workspace = self.scenario.workspace
        self.assertEqual(
            Deb822("".join(suite_apt.pre.itertext())),
            {
                "Types": "deb",
                "URIs": (
                    f"http://{fqdn}/{workspace.scope.name}/{workspace.name}"
                ),
                "Suites": "bookworm",
                "Components": "main contrib non-free",
            },
        )
        self.assert_has_caching_headers(response, mutable=True)

    def test_current_with_required_suites(self) -> None:
        personal = self.playground.create_collection(
            "r-personal",
            CollectionCategory.SUITE,
            workspace=self.scenario.workspace,
        )
        CollectionRelation.objects.create(
            source=personal,
            target=self.suites["trixie"],
            created_by_user=self.playground.get_default_user(),
            type=CollectionRelation.Types.SUITE_REQUIRES,
            position=1,
        )

        response = self.archive_file_get(path="dists/r-personal/")

        tree = self.assertResponseHTML(response)

        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_FQDN
        workspace = self.scenario.workspace
        uri = f"http://{fqdn}/{workspace.scope.name}/{workspace.name}"

        required_apt_sources = "".join(
            tree.xpath("//div[@id='suite-required-apt']/pre/code/text()")
        )

        self.assertEqual(
            list(Deb822.iter_paragraphs(required_apt_sources)),
            [
                {
                    "Types": "deb",
                    "URIs": uri,
                    "Suites": "trixie",
                },
            ],
        )

    def test_current_signing_keys(self) -> None:
        suite = self.suites["bookworm"]
        keys = [
            PGPKey.new(PubKeyAlgorithm.EdDSA, EllipticCurveOID.Ed25519).pubkey
            for _ in range(2)
        ]
        for key in keys:
            self.playground.create_asset_usage(
                resource=self.playground.create_signing_key_asset(
                    purpose=KeyPurpose.OPENPGP_REPOSITORY,
                    fingerprint=key.fingerprint,
                    public_key=str(key),
                    workspace=suite.workspace,
                ),
                workspace=suite.workspace,
            )
        suite.data["signing_keys"] = [key.fingerprint for key in keys]
        suite.save()

        response = self.archive_file_get(path="dists/bookworm/")

        tree = self.assertResponseHTML(response)
        self.assertTextContentEqual(
            tree.xpath("//div[@id='suite-summary']")[0], "bookworm"
        )
        self.assertFalse(tree.xpath("//div[@id='suite-architectures']"))
        self.assertFalse(tree.xpath("//div[@id='suite-release-fields']"))
        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_FQDN
        workspace = self.scenario.workspace
        suite_apt = tree.xpath("//div[@id='suite-apt']")[0]
        suite_deb822 = Deb822("".join(suite_apt.pre.itertext()))
        self.assertCountEqual(
            suite_deb822.keys(), ["Types", "URIs", "Suites", "Signed-By"]
        )
        self.assertDictContainsAll(
            suite_deb822,
            {
                "Types": "deb",
                "URIs": (
                    f"http://{fqdn}/{workspace.scope.name}/{workspace.name}"
                ),
                "Suites": "bookworm",
            },
        )
        self.assertEqual(
            PGPKeyring(
                parse_multiline(suite_deb822["Signed-By"])
            ).fingerprints(),
            {key.fingerprint for key in keys},
        )
        self.assert_has_caching_headers(response, mutable=True)

    def test_snapshot(self) -> None:
        now = timezone.now()
        one_month_ago = (now - dt.timedelta(days=30)).replace(microsecond=0)
        self.archive.child_items.filter(
            category=CollectionCategory.SUITE, name="bookworm"
        ).update(created_at=one_month_ago)

        # We always show an exported suite, regardless of when it was added
        # to the archive, so we don't bother testing different snapshot
        # times here.
        response = self.archive_file_get(
            snapshot=one_month_ago, path="dists/bookworm/"
        )

        tree = self.assertResponseHTML(response)
        self.assertTextContentEqual(
            tree.xpath("//div[@id='suite-summary']")[0], "bookworm"
        )
        self.assertFalse(tree.xpath("//div[@id='suite-architectures']"))
        self.assertFalse(tree.xpath("//div[@id='suite-release-fields']"))
        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_FQDN
        suite_apt = tree.xpath("//div[@id='suite-apt']")[0]
        workspace = self.scenario.workspace
        self.assertEqual(
            Deb822("".join(suite_apt.pre.itertext())),
            {
                "Types": "deb",
                "URIs": (
                    f"http://{fqdn}/{workspace.scope.name}/{workspace.name}"
                ),
                "Suites": "bookworm",
                "Snapshot": self.format_snapshot(one_month_ago),
            },
        )
        self.assert_has_caching_headers(response, mutable=False)

    def test_suite_not_found(self) -> None:
        for snapshot in (None, timezone.now()):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    snapshot=snapshot, path="dists/nonexistent/"
                )

                self.assertResponse404(response, "Suite not found")
                self.assert_has_caching_headers(
                    response, mutable=snapshot is None
                )

    def test_suite_not_exported(self) -> None:
        self.playground.create_collection(
            "not-exported",
            CollectionCategory.SUITE,
            workspace=self.scenario.workspace,
            data={"exported": False},
        )

        for snapshot in (None, timezone.now()):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    snapshot=snapshot, path="dists/not-exported/"
                )

                self.assertResponse404(response, "Suite not found")
                self.assert_has_caching_headers(
                    response, mutable=snapshot is None
                )

    def test_get_place(self) -> None:
        suite = self.suites["bookworm"]

        response = self.archive_file_get(path="dists/bookworm/")
        assert response.context_data is not None

        place = response.context_data["place"]
        self.assertPlaceValueEqual(
            place.title, str(suite), page=f"Suite <code>{suite}</code>"
        )


class SigningKeysViewArchiveTests(ArchiveViewTests):
    """Test retrieving signing keys for an archive."""

    keys: ClassVar[dict[str, PGPKey]]

    example_path = "signing-keys.asc"
    expect_in_suite = False
    expect_mutable_no_archive = True
    expect_mutable = True
    expect_content_type = "application/pgp-keys; charset=utf-8"

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.keys = {
            suite_name: PGPKey.new(
                PubKeyAlgorithm.EdDSA, EllipticCurveOID.Ed25519
            ).pubkey
            for suite_name in cls.suites
        }
        for suite_name, suite in cls.suites.items():
            cls.playground.create_asset_usage(
                resource=cls.playground.create_signing_key_asset(
                    purpose=KeyPurpose.OPENPGP_REPOSITORY,
                    fingerprint=cls.keys[suite_name].fingerprint,
                    public_key=str(cls.keys[suite_name]),
                    workspace=suite.workspace,
                ),
                workspace=suite.workspace,
            )
            suite.data["signing_keys"] = [cls.keys[suite_name].fingerprint]
            suite.save()

    def test_current(self) -> None:
        response = self.archive_file_get(path=self.example_path)

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            response.headers["Content-Type"], self.expect_content_type
        )
        self.assertEqual(
            PGPKeyring(response.content).fingerprints(),
            {key.fingerprint for key in self.keys.values()},
        )
        self.assert_has_caching_headers(response, mutable=True)


class SigningKeysViewSuiteTests(ArchiveViewTests):
    """Test retrieving signing keys for a suite."""

    keys: ClassVar[list[PGPKey]]

    example_path = "dists/bookworm/signing-keys.asc"
    expect_in_suite = True
    expect_mutable_no_archive = True
    expect_mutable = True
    expect_content_type = "application/pgp-keys; charset=utf-8"

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        suite = cls.suites["bookworm"]
        cls.keys = [
            PGPKey.new(PubKeyAlgorithm.EdDSA, EllipticCurveOID.Ed25519).pubkey
            for _ in range(2)
        ]
        for key in cls.keys:
            cls.playground.create_asset_usage(
                resource=cls.playground.create_signing_key_asset(
                    purpose=KeyPurpose.OPENPGP_REPOSITORY,
                    fingerprint=key.fingerprint,
                    public_key=str(key),
                    workspace=suite.workspace,
                ),
                workspace=suite.workspace,
            )
        suite.data["signing_keys"] = [key.fingerprint for key in cls.keys]
        suite.save()

    def test_current(self) -> None:
        response = self.archive_file_get(path=self.example_path)

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(
            response.headers["Content-Type"], self.expect_content_type
        )
        self.assertEqual(
            PGPKeyring(response.content).fingerprints(),
            {key.fingerprint for key in self.keys},
        )
        self.assert_has_caching_headers(response, mutable=True)


class DistsByHashFileViewTests(ArchiveFileViewTests):
    """Test retrieving ``by-hash`` files from ``dists/``."""

    contents: ClassVar[dict[str, bytes]]
    sha256: ClassVar[dict[str, str]]

    example_path_nonexistent = "dists/bookworm/main/source/by-hash/SHA256/00"
    expect_in_suite = True
    expect_mutable_no_archive = False
    expect_mutable = False
    expect_content_type = "application/x-xz"

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.contents = {
            "old": b"Old Sources file\n",
            "new": b"New Sources file\n",
        }
        cls.sha256 = {
            name: hashlib.sha256(c).hexdigest()
            for name, c in cls.contents.items()
        }
        cls.suites["bookworm"].manager.add_artifact(
            cls.playground.create_repository_index(
                "Sources.xz", contents=cls.contents["old"]
            ),
            user=cls.scenario.user,
            variables={"path": "main/source/Sources.xz"},
        )
        cls.suites["bookworm"].manager.add_artifact(
            cls.playground.create_repository_index(
                "Sources.xz", contents=cls.contents["new"]
            ),
            user=cls.scenario.user,
            variables={"path": "main/source/Sources.xz"},
            replace=True,
        )
        cls.suites["bookworm"].manager.add_artifact(
            cls.playground.create_repository_index("Packages.xz"),
            user=cls.scenario.user,
            variables={"path": "main/binary-amd64/Packages.xz"},
        )
        cls.suites["trixie"].manager.add_artifact(
            cls.playground.create_repository_index("Sources.xz"),
            user=cls.scenario.user,
            variables={"path": "main/source/Sources.xz"},
        )
        cls.example_path = (
            f"dists/bookworm/main/source/by-hash/SHA256/{cls.sha256['new']}"
        )

    def test_current(self) -> None:
        response = self.archive_file_get(
            path=(
                f"dists/bookworm/main/source/by-hash/SHA256/"
                f"{self.sha256['old']}"
            )
        )

        self.assert_response_ok(
            response, contents=self.contents["old"], mutable=False
        )

        response = self.archive_file_get(
            path=(
                f"dists/bookworm/main/source/by-hash/SHA256/"
                f"{self.sha256['new']}"
            )
        )

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=False
        )

    def test_snapshot(self) -> None:
        one_month_ago = (timezone.now() - dt.timedelta(days=30)).replace(
            microsecond=0
        )
        self.archive.child_items.filter(
            category=CollectionCategory.SUITE
        ).update(created_at=one_month_ago)
        old_item = (
            self.suites["bookworm"]
            .child_items.filter(name="index:main/source/Sources.xz")
            .order_by("id")[0]
        )
        old_item.created_at = one_month_ago
        old_item.save()
        new_item = (
            self.suites["bookworm"]
            .child_items.filter(name="index:main/source/Sources.xz")
            .order_by("id")[1]
        )

        for snapshot, sha256, expected_contents in (
            (
                old_item.created_at - dt.timedelta(seconds=1),
                self.sha256["old"],
                None,
            ),
            (
                old_item.created_at - dt.timedelta(seconds=1),
                self.sha256["new"],
                None,
            ),
            (old_item.created_at, self.sha256["old"], self.contents["old"]),
            (old_item.created_at, self.sha256["new"], None),
            (
                new_item.created_at - dt.timedelta(seconds=1),
                self.sha256["old"],
                self.contents["old"],
            ),
            (
                new_item.created_at - dt.timedelta(seconds=1),
                self.sha256["new"],
                None,
            ),
            # Snapshot URLs don't include a microsecond component, so add a
            # second to avoid rounding problems.
            (
                new_item.created_at + dt.timedelta(seconds=1),
                self.sha256["old"],
                self.contents["old"],
            ),
            (
                new_item.created_at + dt.timedelta(seconds=1),
                self.sha256["new"],
                self.contents["new"],
            ),
        ):
            with self.subTest(
                snapshot=self.format_snapshot(snapshot), sha256=sha256
            ):
                response = self.archive_file_get(
                    snapshot=snapshot,
                    path=f"dists/bookworm/main/source/by-hash/SHA256/{sha256}",
                )

                if expected_contents is None:
                    self.assertResponse404(
                        response, "No FileInArtifact matches the given query."
                    )
                else:
                    self.assert_response_ok(
                        response, contents=expected_contents, mutable=False
                    )

    def test_suite_not_exported(self) -> None:
        response = self.archive_file_get(
            path=(
                f"dists/bookworm/main/source/by-hash/SHA256/"
                f"{self.sha256['new']}"
            )
        )

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=False
        )

        self.suites["bookworm"].data["exported"] = False
        self.suites["bookworm"].save()

        response = self.archive_file_get(
            path=(
                f"dists/bookworm/main/source/by-hash/SHA256/"
                f"{self.sha256['new']}"
            )
        )

        self.assertResponse404(
            response, "No FileInArtifact matches the given query."
        )
        self.assert_has_caching_headers(response, mutable=False)

    def test_ignores_content_negotiation(self) -> None:
        response = self.archive_file_get(
            path=(
                f"dists/bookworm/main/source/by-hash/SHA256/"
                f"{self.sha256['new']}"
            ),
            headers={"Accept": "application/octet-stream"},
        )

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=False
        )

    def test_by_hash_matches_multiple_artifacts(self) -> None:
        contents = b"Sources file"
        sha256 = hashlib.sha256(contents).hexdigest()
        for _ in range(2):
            self.suites["bookworm"].manager.add_artifact(
                self.playground.create_repository_index(
                    "Sources.xz", contents=contents
                ),
                replace=True,
                user=self.scenario.user,
                variables={"path": "main/source/Sources.xz"},
            )

        response = self.archive_file_get(
            path=f"dists/bookworm/main/source/by-hash/SHA256/{sha256}",
        )

        self.assert_response_ok(
            response, contents=b"Sources file", mutable=False
        )


class DistsFileViewTests(ArchiveFileViewTests):
    """Test retrieving non-``by-hash`` files from ``dists/``."""

    contents: ClassVar[dict[str, bytes]]

    example_path = "dists/bookworm/Release"
    example_path_nonexistent = "dists/bookworm/Nonexistent"
    expect_in_suite = True
    expect_mutable_no_archive = True
    expect_mutable = True
    # This is really just the Content-Type of the Release file.
    expect_content_type = "application/octet-stream"

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.contents = {
            "old": b"Old Release file\n",
            "new": b"New Release file\n",
        }
        cls.suites["bookworm"].manager.add_artifact(
            cls.playground.create_repository_index(
                "Release", contents=b"Old Release file\n"
            ),
            user=cls.scenario.user,
            variables={"path": "Release"},
        )
        cls.suites["bookworm"].manager.add_artifact(
            cls.playground.create_repository_index(
                "Release", contents=b"New Release file\n"
            ),
            user=cls.scenario.user,
            variables={"path": "Release"},
            replace=True,
        )
        cls.suites["bookworm"].manager.add_artifact(
            cls.playground.create_repository_index("Sources.xz"),
            user=cls.scenario.user,
            variables={"path": "main/source/Sources.xz"},
        )
        cls.suites["trixie"].manager.add_artifact(
            cls.playground.create_repository_index("Release"),
            user=cls.scenario.user,
            variables={"path": "Release"},
        )

    def test_current(self) -> None:
        response = self.archive_file_get(path="dists/bookworm/Release")

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=True
        )

    def test_snapshot(self) -> None:
        one_month_ago = (timezone.now() - dt.timedelta(days=30)).replace(
            microsecond=0
        )
        self.archive.child_items.filter(
            category=CollectionCategory.SUITE
        ).update(created_at=one_month_ago)
        old_item = (
            self.suites["bookworm"]
            .child_items.filter(name="index:Release")
            .order_by("id")[0]
        )
        old_item.created_at = one_month_ago
        old_item.save()
        new_item = (
            self.suites["bookworm"]
            .child_items.filter(name="index:Release")
            .order_by("id")[1]
        )

        for snapshot, expected_contents in (
            (old_item.created_at - dt.timedelta(seconds=1), None),
            (old_item.created_at, self.contents["old"]),
            (
                new_item.created_at - dt.timedelta(seconds=1),
                self.contents["old"],
            ),
            # Snapshot URLs don't include a microsecond component, so add a
            # second to avoid rounding problems.
            (
                new_item.created_at + dt.timedelta(seconds=1),
                self.contents["new"],
            ),
        ):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    snapshot=snapshot, path="dists/bookworm/Release"
                )

                if expected_contents is None:
                    self.assertResponse404(
                        response, "No FileInArtifact matches the given query."
                    )
                else:
                    self.assert_response_ok(
                        response, contents=expected_contents, mutable=False
                    )

    def test_suite_not_exported(self) -> None:
        response = self.archive_file_get(path="dists/bookworm/Release")

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=True
        )

        self.suites["bookworm"].data["exported"] = False
        self.suites["bookworm"].save()

        response = self.archive_file_get(path="dists/bookworm/Release")

        self.assertResponse404(
            response, "No FileInArtifact matches the given query."
        )
        self.assert_has_caching_headers(response, mutable=True)

    def test_ignores_content_negotiation(self) -> None:
        response = self.archive_file_get(
            path="dists/bookworm/Release",
            headers={"Accept": "application/octet-stream"},
        )

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=True
        )


class PoolFileViewTests(ArchiveFileViewTests):
    """Test retrieving files from ``pool/``."""

    example_path = "pool/contrib/h/hello/hello_1.1-1.dsc"
    example_path_nonexistent = "pool/main/n/nonexistent/nonexistent_1.0.dsc"
    expect_in_suite = True
    expect_mutable_no_archive = True
    expect_mutable = False
    expect_content_type = "text/plain; charset=utf-8"

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        for suite_name, component, name, version, paths in (
            (
                "bookworm",
                "main",
                "hello",
                "1.0",
                ["hello_1.0.dsc", "hello_1.0.tar.xz"],
            ),
            (
                "bookworm",
                "contrib",
                "hello",
                "1.1-1",
                [
                    "hello_1.1-1.dsc",
                    "hello_1.1-1.debian.tar.xz",
                    "hello_1.1.orig.tar.xz",
                ],
            ),
            (
                "bookworm",
                "contrib",
                "hello",
                "1.1-2",
                [
                    "hello_1.1-2.dsc",
                    "hello_1.1-2.debian.tar.xz",
                    "hello_1.1.orig.tar.xz",
                ],
            ),
            ("bookworm", "main", "base-files", "1.0", ["base-files_1.0.dsc"]),
            (
                "trixie",
                "main",
                "hello",
                "1.0",
                ["hello_1.0.dsc", "hello_1.0.tar.xz"],
            ),
        ):
            cls.suites[suite_name].manager.add_artifact(
                cls.playground.create_minimal_source_package_artifact(
                    name,
                    version,
                    paths={
                        path: f"Contents of {path}".encode() for path in paths
                    },
                    create_files=True,
                ),
                user=cls.scenario.user,
                variables={"component": component, "section": "devel"},
            )

        for (
            suite_name,
            component,
            srcpkg_name,
            srcpkg_version,
            name,
            version,
            architecture,
        ) in (
            ("bookworm", "main", "hello", "1.0", "libhello1", "1.0", "amd64"),
            ("bookworm", "main", "hello", "1.0", "libhello1", "1.0", "s390x"),
            ("bookworm", "main", "hello", "1.0", "libhello-doc", "1.0", "all"),
            (
                "bookworm",
                "contrib",
                "hello",
                "1.1",
                "libhello1",
                "1:1.1",
                "amd64",
            ),
            (
                "bookworm",
                "main",
                "base-files",
                "1.0",
                "base-files",
                "1.0",
                "amd64",
            ),
            ("trixie", "main", "hello", "1.0", "libhello1", "1.0", "amd64"),
        ):
            deb_path = f"{name}_{version.split(':')[-1]}_{architecture}.deb"
            cls.suites[suite_name].manager.add_artifact(
                cls.playground.create_minimal_binary_package_artifact(
                    srcpkg_name,
                    srcpkg_version,
                    name,
                    version,
                    architecture,
                    paths={deb_path: f"Contents of {deb_path}".encode()},
                    create_files=True,
                ),
                user=cls.scenario.user,
                variables={
                    "component": component,
                    "section": "devel",
                    "priority": "optional",
                },
            )

    def test_current_source(self) -> None:
        for path in (
            "pool/main/h/hello/hello_1.0.dsc",
            "pool/main/h/hello/hello_1.0.tar.xz",
            "pool/contrib/h/hello/hello_1.1-1.dsc",
            "pool/contrib/h/hello/hello_1.1-1.debian.tar.xz",
            "pool/contrib/h/hello/hello_1.1-2.dsc",
            "pool/contrib/h/hello/hello_1.1-2.debian.tar.xz",
            "pool/contrib/h/hello/hello_1.1.orig.tar.xz",
        ):
            response = self.archive_file_get(path=path)

            self.assert_response_ok(
                response,
                contents=f"Contents of {PurePath(path).name}".encode(),
                mutable=False,
            )

    def test_current_binary(self) -> None:
        for path in (
            "pool/main/h/hello/libhello1_1.0_amd64.deb",
            "pool/main/h/hello/libhello1_1.0_s390x.deb",
            "pool/main/h/hello/libhello-doc_1.0_all.deb",
            "pool/contrib/h/hello/libhello1_1.1_amd64.deb",
        ):
            response = self.archive_file_get(path=path)

            self.assert_response_ok(
                response,
                contents=f"Contents of {PurePath(path).name}".encode(),
                mutable=False,
            )

    def test_snapshot_source(self) -> None:
        one_month_ago = (timezone.now() - dt.timedelta(days=30)).replace(
            microsecond=0
        )
        one_day_ago = (timezone.now() - dt.timedelta(days=1)).replace(
            microsecond=0
        )
        self.archive.child_items.filter(
            category=CollectionCategory.SUITE
        ).update(created_at=one_month_ago)
        items = []
        for version in ("1.1-1", "1.1-2", "1.1-3"):
            paths = {
                path: f"Contents of {path}".encode()
                for path in (
                    f"hello-snapshot_{version}.dsc",
                    f"hello-snapshot_{version.split('-')[0]}.orig.tar.xz",
                )
            }
            items.append(
                self.suites["bookworm"].manager.add_artifact(
                    self.playground.create_minimal_source_package_artifact(
                        "hello-snapshot",
                        version,
                        paths=paths,
                        create_files=True,
                    ),
                    user=self.scenario.user,
                    variables={"component": "main", "section": "devel"},
                )
            )
        items[0].created_at = one_month_ago
        items[0].removed_at = one_day_ago
        items[0].save()
        items[1].created_at = one_day_ago
        items[1].save()
        start = items[0].created_at - dt.timedelta(seconds=1)
        # Snapshot URLs don't include a microsecond component, so add a
        # second to avoid rounding problems.
        end = items[2].created_at + dt.timedelta(seconds=1)

        for snapshot, name, expected in (
            (start, "hello-snapshot_1.1-1.dsc", False),
            (start, "hello-snapshot_1.1-2.dsc", False),
            (start, "hello-snapshot_1.1-3.dsc", False),
            (start, "hello-snapshot_1.1.orig.tar.xz", False),
            (items[0].created_at, "hello-snapshot_1.1-1.dsc", True),
            (items[0].created_at, "hello-snapshot_1.1-2.dsc", False),
            (items[0].created_at, "hello-snapshot_1.1-3.dsc", False),
            (items[0].created_at, "hello-snapshot_1.1.orig.tar.xz", True),
            (items[1].created_at, "hello-snapshot_1.1-1.dsc", False),
            (items[1].created_at, "hello-snapshot_1.1-2.dsc", True),
            (items[1].created_at, "hello-snapshot_1.1-3.dsc", False),
            (items[1].created_at, "hello-snapshot_1.1.orig.tar.xz", True),
            (end, "hello-snapshot_1.1-1.dsc", False),
            (end, "hello-snapshot_1.1-2.dsc", True),
            (end, "hello-snapshot_1.1-3.dsc", True),
            (end, "hello-snapshot_1.1.orig.tar.xz", True),
        ):
            with self.subTest(
                snapshot=self.format_snapshot(snapshot), name=name
            ):
                response = self.archive_file_get(
                    snapshot=snapshot, path=f"pool/main/h/hello-snapshot/{name}"
                )

                if expected:
                    self.assert_response_ok(
                        response,
                        contents=f"Contents of {name}".encode(),
                        mutable=False,
                    )
                else:
                    self.assertResponse404(
                        response, "No FileInArtifact matches the given query."
                    )

    def test_suite_not_exported(self) -> None:
        response = self.archive_file_get(path=self.example_path)

        self.assert_response_ok(
            response,
            contents=f"Contents of {self.example_path.split('/')[-1]}".encode(),
            mutable=False,
        )

        self.suites["bookworm"].data["exported"] = False
        self.suites["bookworm"].save()

        response = self.archive_file_get(path=self.example_path)

        self.assertResponse404(
            response, "No FileInArtifact matches the given query."
        )
        self.assert_has_caching_headers(response, mutable=False)

    def test_wrong_component(self) -> None:
        for name in ("hello_1.0.dsc", "libhello1_1.0_amd64.deb"):
            response = self.archive_file_get(path=f"pool/main/h/hello/{name}")

            self.assertEqual(response.status_code, status.HTTP_200_OK)

            response = self.archive_file_get(
                path=f"pool/contrib/h/hello/{name}"
            )

            self.assertResponse404(
                response, "No FileInArtifact matches the given query."
            )

    def test_wrong_source_prefix(self) -> None:
        for name in ("hello_1.0.dsc", "libhello1_1.0_amd64.deb"):
            response = self.archive_file_get(path=f"pool/main/h/hello/{name}")

            self.assertEqual(response.status_code, status.HTTP_200_OK)

            response = self.archive_file_get(path=f"pool/main/a/hello/{name}")

            self.assertResponse404(
                response, "No FileInArtifact matches the given query."
            )

    def test_ignores_content_negotiation(self) -> None:
        response = self.archive_file_get(
            path="pool/main/h/hello/hello_1.0.dsc",
            headers={"Accept": "application/octet-stream"},
        )

        self.assert_response_ok(
            response, contents=b"Contents of hello_1.0.dsc", mutable=False
        )


class TopLevelFileViewTests(ArchiveFileViewTests):
    """Test retrieving top-level files, not in any suite."""

    contents: ClassVar[dict[str, bytes]]

    example_path = "doc/social-contract.txt"
    example_path_nonexistent = "NONEXISTENT"
    expect_in_suite = False
    expect_mutable_no_archive = True
    expect_mutable = True
    expect_content_type = "text/plain; charset=utf-8"

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.contents = {
            "old": b'"Social Contract" with the Free Software Community\n',
            "new": b'"Social Contract" with the Free Software Community (v2)\n',
        }
        old_item = cls.archive.manager.add_artifact(
            cls.playground.create_repository_index(
                "social-contract.txt", contents=cls.contents["old"]
            ),
            user=cls.scenario.user,
            variables={"path": "doc/social-contract.txt"},
        )
        old_item.created_at = (timezone.now() - dt.timedelta(days=30)).replace(
            microsecond=0
        )
        old_item.save()
        cls.archive.manager.add_artifact(
            cls.playground.create_repository_index(
                "social-contract.txt", contents=cls.contents["new"]
            ),
            user=cls.scenario.user,
            variables={"path": "doc/social-contract.txt"},
            replace=True,
        )

    def test_current(self) -> None:
        response = self.archive_file_get(path="doc/social-contract.txt")

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=True
        )

    def test_snapshot(self) -> None:
        old_item = self.archive.child_items.filter(
            name="index:doc/social-contract.txt"
        ).order_by("id")[0]
        new_item = self.archive.child_items.filter(
            name="index:doc/social-contract.txt"
        ).order_by("id")[1]

        for snapshot, expected_contents in (
            (old_item.created_at - dt.timedelta(seconds=1), None),
            (old_item.created_at, self.contents["old"]),
            (
                new_item.created_at - dt.timedelta(seconds=1),
                self.contents["old"],
            ),
            # Snapshot URLs don't include a microsecond component, so add a
            # second to avoid rounding problems.
            (
                new_item.created_at + dt.timedelta(seconds=1),
                self.contents["new"],
            ),
        ):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    snapshot=snapshot, path="doc/social-contract.txt"
                )

                if expected_contents is None:
                    self.assertResponse404(
                        response, "No FileInArtifact matches the given query."
                    )
                else:
                    self.assert_response_ok(
                        response, contents=expected_contents, mutable=False
                    )

    def test_ignores_content_negotiation(self) -> None:
        response = self.archive_file_get(
            path=self.example_path,
            headers={"Accept": "application/octet-stream"},
        )

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=True
        )


class RenderableSuiteTests(TestCase):
    """Tests for RenderableSuite."""

    @override
    def setUp(self) -> None:
        super().setUp()
        self.trixie = self.playground.create_collection(
            "trixie", CollectionCategory.SUITE
        )

    def test_as_deb822_source(self) -> None:
        renderable = RenderableSuite(
            request_scheme="http", collection=self.trixie
        )
        self.assertEqual(
            renderable.as_deb822_source(),
            "Types: deb\n"
            "URIs: http://127.0.0.1:8001/debusine/System\n"
            "Suites: trixie\n",
        )

    def test_required_suites_as_deb822_sources(self) -> None:
        personal = self.playground.create_collection(
            "personal", CollectionCategory.SUITE
        )
        trixie_backports = self.playground.create_collection(
            "trixie-backports", CollectionCategory.SUITE
        )
        trixie_security = self.playground.create_collection(
            "trixie-security", CollectionCategory.SUITE
        )
        CollectionRelation.objects.create(
            source=personal,
            target=trixie_backports,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            created_by_user=self.playground.get_default_user(),
            position=1,
        )
        CollectionRelation.objects.create(
            source=personal,
            target=trixie_security,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            created_by_user=self.playground.get_default_user(),
            position=2,
        )
        CollectionRelation.objects.create(
            source=trixie_backports,
            target=self.trixie,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            created_by_user=self.playground.get_default_user(),
            position=1,
        )
        CollectionRelation.objects.create(
            source=trixie_backports,
            target=trixie_security,
            type=CollectionRelation.Types.SUITE_REQUIRES,
            created_by_user=self.playground.get_default_user(),
            position=1,
        )

        trixie_backports_renderable = RenderableSuite(
            request_scheme="http", collection=trixie_backports
        )
        trixie_security_renderable = RenderableSuite(
            request_scheme="http", collection=trixie_security
        )
        trixie_renderable = RenderableSuite(
            request_scheme="http", collection=self.trixie
        )

        renderable = RenderableSuite(request_scheme="http", collection=personal)

        self.assertEqual(
            renderable.required_suites_as_deb822_sources(),
            f"{trixie_backports_renderable.as_deb822_source()}\n"
            f"{trixie_security_renderable.as_deb822_source()}\n"
            f"{trixie_renderable.as_deb822_source()}",
        )


class DebugInfoViewTests(ArchiveFileViewTests):
    """Test ``/buildid/<build-id>/debuginfo``."""

    BUILD_ID = "a" * 40
    BUILD_ID_OTHER = "b" * 40

    example_path = f"buildid/{BUILD_ID}/debuginfo"
    example_path_nonexistent = f"buildid/{BUILD_ID_OTHER}/debuginfo"
    expect_in_suite = True
    expect_mutable_no_archive = False
    expect_mutable = False
    expect_content_type = "application/octet-stream"

    contents: ClassVar[dict[str, bytes]]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.suites["bookworm"].data["may_reuse_versions"] = True
        cls.suites["bookworm"].save()
        cls.contents = {
            "old": b"debug-symbol-bytes-old",
            "new": b"debug-symbol-bytes-new",
        }
        binary = cls.playground.create_minimal_binary_package_artifact(
            "hello",
            "1.0-1",
            "hello",
            "1.0-1",
            "amd64",
        )
        for label in ("old", "new"):
            debug_artifact, _ = cls.playground.create_artifact(
                category=ArtifactCategory.DEBUG_SYMBOLS,
                data={"build_ids": [cls.BUILD_ID]},
                paths={cls.BUILD_ID: cls.contents[label]},
                create_files=True,
            )
            cls.playground.create_artifact_relation(
                debug_artifact, binary, ArtifactRelation.Relations.RELATES_TO
            )
            cls.suites["bookworm"].manager.add_artifact(
                debug_artifact,
                user=cls.scenario.user,
                variables={"build_id": cls.BUILD_ID},
                replace=label == "new",
            )

    def test_current(self) -> None:
        response = self.archive_file_get(path=self.example_path)

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=False
        )
        self.assertEqual(
            response.headers["X-DEBUGINFOD-FILE"],
            f"/usr/lib/debug/.build-id/{self.BUILD_ID[:2]}/"
            f"{self.BUILD_ID[2:]}.debug",
        )
        self.assertEqual(
            response.headers["X-DEBUGINFOD-SIZE"],
            str(len(self.contents["new"])),
        )

    def test_snapshot(self) -> None:
        one_month_ago = (timezone.now() - dt.timedelta(days=30)).replace(
            microsecond=0
        )
        self.archive.child_items.filter(
            category=CollectionCategory.SUITE
        ).update(created_at=one_month_ago)
        old_item = (
            self.suites["bookworm"]
            .child_items.filter(category=ArtifactCategory.DEBUG_SYMBOLS)
            .order_by("id")[0]
        )
        old_item.created_at = one_month_ago
        old_item.save()
        new_item = (
            self.suites["bookworm"]
            .child_items.filter(category=ArtifactCategory.DEBUG_SYMBOLS)
            .order_by("id")[1]
        )

        for snapshot, expected_contents in (
            (old_item.created_at - dt.timedelta(seconds=1), None),
            (old_item.created_at, self.contents["old"]),
            (
                new_item.created_at - dt.timedelta(seconds=1),
                self.contents["old"],
            ),
            # Snapshot URLs don't include a microsecond component, so add a
            # second to avoid rounding problems.
            (
                new_item.created_at + dt.timedelta(seconds=1),
                self.contents["new"],
            ),
        ):
            with self.subTest(snapshot=self.format_snapshot(snapshot)):
                response = self.archive_file_get(
                    snapshot=snapshot, path=self.example_path
                )

                if expected_contents is None:
                    self.assertResponse404(
                        response, "No FileInArtifact matches the given query."
                    )
                else:
                    self.assert_response_ok(
                        response, contents=expected_contents, mutable=False
                    )

    def test_build_id_case_insensitive(self) -> None:
        response = self.archive_file_get(
            path=f"buildid/{self.BUILD_ID.upper()}/debuginfo"
        )

        self.assert_response_ok(
            response, contents=self.contents["new"], mutable=False
        )

    def test_invalid_build_id(self) -> None:
        response = self.archive_file_get(
            path="buildid/not-a-build-id/debuginfo"
        )

        self.assertResponse404(response, "Invalid build-ID.")
        self.assert_has_caching_headers(response, mutable=False)

    def test_head(self) -> None:
        fqdn = settings.DEBUSINE_DEBIAN_ARCHIVE_PRIMARY_FQDN
        scope_name = self.scenario.scope.name
        workspace_name = self.scenario.workspace.name
        response = self.client.head(
            f"/{scope_name}/{workspace_name}/{self.example_path}",
            HTTP_HOST=fqdn,
        )

        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assert_has_caching_headers(response, mutable=False)
        self.assertEqual(
            response.headers["X-DEBUGINFOD-FILE"],
            f"/usr/lib/debug/.build-id/{self.BUILD_ID[:2]}/"
            f"{self.BUILD_ID[2:]}.debug",
        )
        self.assertEqual(
            response.headers["X-DEBUGINFOD-SIZE"],
            str(len(self.contents["new"])),
        )

    def test_range(self) -> None:
        response = self.archive_file_get(
            path=self.example_path,
            headers={"Range": "bytes=0-4"},
        )

        self.assertEqual(response.status_code, status.HTTP_206_PARTIAL_CONTENT)
        self.assert_has_caching_headers(response, mutable=False)
        self.assertEqual(
            b"".join(getattr(response, "streaming_content")),
            self.contents["new"][:5],
        )
        self.assertEqual(
            response.headers["Content-Range"],
            f"bytes 0-4/{len(self.contents['new'])}",
        )
        self.assertEqual(
            response.headers["X-DEBUGINFOD-FILE"],
            f"/usr/lib/debug/.build-id/{self.BUILD_ID[:2]}/"
            f"{self.BUILD_ID[2:]}.debug",
        )
        self.assertEqual(
            response.headers["X-DEBUGINFOD-SIZE"],
            str(len(self.contents["new"])),
        )


# Avoid running tests from common base classes.
del ArchiveViewTests
del ArchiveFileViewTests
