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

"""Tests for the collections views."""

import datetime as dt
from typing import Any, ClassVar, assert_never, override
from unittest import mock

from django.db import IntegrityError
from django.db.models import Max
from django.urls import reverse
from rest_framework import status

from debusine.artifacts.models import (
    BareDataCategory,
    CollectionCategory,
    DebusineTaskConfiguration,
    TaskTypes,
)
from debusine.client.exceptions import DebusineError
from debusine.client.models import CollectionData, CollectionDataNew
from debusine.db.models import (
    Collection,
    CollectionItem,
    CollectionTokenGrant,
    Token,
    Workspace,
)
from debusine.db.playground import scenarios
from debusine.server.collections import DebusineTaskConfigurationManager
from debusine.server.views.collections import TaskConfigurationCollectionView
from debusine.server.views.tests.base import TestCase
from debusine.test.django import (
    AllowAll,
    DenyAll,
    TestResponseType,
    override_permission,
)


class TaskConfigurationCollectionViewTests(TestCase):
    """Tests for TaskConfigurationCollectionView."""

    scenario = scenarios.DefaultContextAPI()
    collection: ClassVar[Collection]
    manager: ClassVar[DebusineTaskConfigurationManager]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.collection = cls.scenario.default_task_configuration_collection
        cls.manager = DebusineTaskConfigurationManager(
            collection=cls.collection
        )

    @classmethod
    def add_config(cls, entry: DebusineTaskConfiguration) -> CollectionItem:
        """Add a config entry to config_collection."""
        return cls.manager.add_bare_data(
            BareDataCategory.TASK_CONFIGURATION,
            user=cls.scenario.user,
            data=entry,
        )

    def get_items(self) -> list[DebusineTaskConfiguration]:
        collection = self.collection

        return [
            DebusineTaskConfiguration(**child.data)
            for child in collection.child_items.active().filter(
                child_type=CollectionItem.Types.BARE
            )
        ]

    def get(
        self,
        name: str = "default",
        workspace: Workspace | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for a named collection."""
        match workspace:
            case None:
                workspace_name = self.scenario.workspace.name
            case str():
                workspace_name = workspace
            case _:
                workspace_name = workspace.name

        match user_token:
            case True:
                headers = {"token": self.scenario.user_token.key}
            case False:
                headers = {}
            case _ as unreachable:
                assert_never(unreachable)

        return self.client.get(
            reverse(
                "api:task-configuration-collection",
                kwargs={"workspace": workspace_name, "name": name},
            ),
            content_type="application/json",
            headers=headers,
        )

    def post(
        self,
        name: str = "default",
        workspace: Workspace | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | None = None,
    ) -> TestResponseType:
        """POST request for a named collection."""
        match workspace:
            case None:
                workspace_name = self.scenario.workspace.name
            case str():
                workspace_name = workspace
            case _:
                workspace_name = workspace.name

        match user_token:
            case True:
                headers = {"token": self.scenario.user_token.key}
            case False:
                headers = {}
            case _ as unreachable:
                assert_never(unreachable)

        if data is None:
            data = {}

        return self.client.post(
            reverse(
                "api:task-configuration-collection",
                kwargs={"workspace": workspace_name, "name": name},
            ),
            data=data,
            content_type="application/json",
            headers=headers,
        )

    def test_unauthenticated(self) -> None:
        """Authentication is required."""
        for method in "get", "post":
            with self.subTest(method=method):
                response = getattr(self, method)(user_token=False)
                self.assertResponseProblem(
                    response,
                    "Error",
                    detail_pattern=(
                        "Authentication credentials were not provided."
                    ),
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_workspace_not_found(self) -> None:
        """Workspace must exist."""
        for method in "get", "post":
            with self.subTest(method=method):
                response = getattr(self, method)(workspace="does-not-exist")
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace does-not-exist not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_workspace_not_accessible(self) -> None:
        """User must be able to display the workspace."""
        workspace = self.playground.create_workspace(name="private")
        for method in "get", "post":
            with self.subTest(method=method):
                response = getattr(self, method)(workspace=workspace)
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace private not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_not_found(self) -> None:
        """The collection must exist."""
        for method in "get", "post":
            with self.subTest(method=method):
                response = getattr(self, method)(name="does-not-exist")
                self.assertResponseProblem(
                    response,
                    "Collection not found",
                    detail_pattern=(
                        "Task configuration collection 'does-not-exist'"
                        " does not exist in workspace 'System'"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_get_unauthorized(self) -> None:
        """The user must be able to display the collection."""
        with override_permission(Collection, "can_display", DenyAll):
            response = self.get()
            self.assertResponseProblem(
                response,
                (
                    "playground cannot display collection"
                    " default@debusine:task-configuration"
                ),
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_post_unauthorized(self) -> None:
        """The user must be able to display the collection."""
        with override_permission(
            Workspace, "can_edit_task_configuration", DenyAll
        ):
            response = self.post()
            self.assertResponseProblem(
                response,
                (
                    "playground cannot edit task configuration"
                    " in debusine/System"
                ),
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_get(self) -> None:
        response = self.get()
        data = self.assertAPIResponseOk(response)

        collection = data["collection"]
        self.assertEqual(collection["id"], self.collection.pk)
        self.assertEqual(collection["name"], "default")
        self.assertEqual(collection["data"], self.collection.data)

        items = data["items"]
        self.assertEqual(items, [])

    def test_get_items(self) -> None:
        item = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop"
        )
        self.add_config(item)
        response = self.get()
        data = self.assertAPIResponseOk(response)
        items = data["items"]
        self.assertEqual(items, [item.model_dump()])

    def test_post_matches_collection(self) -> None:
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )
        for data, detail_pattern in (
            (
                {"id": 0, "name": self.collection.name, "data": {}},
                r"Data posted for collection 0"
                rf" to update collection {self.collection.pk}",
            ),
            (
                {
                    "id": self.collection.pk,
                    "name": "does-not-exist",
                    "data": {},
                },
                r"Data posted for collection 'does-not-exist'"
                r" to update collection 'default'",
            ),
        ):
            with self.subTest(data=data):
                response = self.post(data={"collection": data, "items": []})
                self.assertResponseProblem(
                    response,
                    "Posted collection mismatch",
                    detail_pattern=detail_pattern,
                    status_code=status.HTTP_400_BAD_REQUEST,
                )

    def test_post_add(self) -> None:
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )

        item = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop"
        )
        response = self.post(
            data={
                "collection": {
                    "id": self.collection.pk,
                    "name": self.collection.name,
                    "data": self.collection.data,
                },
                "items": [item.model_dump()],
            }
        )
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data, {"added": 1, "removed": 0, "updated": 0, "unchanged": 0}
        )

        self.assertEqual(self.get_items(), [item])

    def test_post_remove(self) -> None:
        self.add_config(
            DebusineTaskConfiguration(
                task_type=TaskTypes.WORKER, task_name="noop"
            )
        )
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )

        response = self.post(
            data={
                "collection": {
                    "id": self.collection.pk,
                    "name": self.collection.name,
                    "data": self.collection.data,
                },
                "items": [],
            }
        )
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data, {"added": 0, "removed": 1, "updated": 0, "unchanged": 0}
        )

        self.assertEqual(self.get_items(), [])

    def test_post_update(self) -> None:
        old = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop"
        )
        new = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER,
            task_name="noop",
            comment="updated",
        )
        self.add_config(old)
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )

        response = self.post(
            data={
                "collection": {
                    "id": self.collection.pk,
                    "name": self.collection.name,
                    "data": self.collection.data,
                },
                "items": [new.model_dump()],
            }
        )
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data, {"added": 0, "removed": 0, "updated": 1, "unchanged": 0}
        )
        self.assertEqual(self.get_items(), [new])

    def test_post_dry_run(self) -> None:
        old = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop"
        )
        new = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER,
            task_name="noop",
            comment="updated",
        )
        self.add_config(old)
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )

        response = self.post(
            data={
                "collection": {
                    "id": self.collection.pk,
                    "name": self.collection.name,
                    "data": self.collection.data,
                },
                "items": [new.model_dump()],
                "dry_run": True,
            }
        )
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data, {"added": 0, "removed": 0, "updated": 1, "unchanged": 0}
        )
        self.assertEqual(self.get_items(), [old])

    def test_post_git_commit_invalid_type(self) -> None:
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )
        value: Any
        for value in (False, 42, [], {}):
            with self.subTest(value=value):
                response = self.post(
                    data={
                        "collection": {
                            "id": self.collection.pk,
                            "name": self.collection.name,
                            "data": {"git_commit": value},
                        },
                        "items": [],
                        "dry_run": True,
                    }
                )
                self.assertResponseProblem(
                    response,
                    "Invalid git commit",
                    detail_pattern="git_commit is not a string",
                    status_code=status.HTTP_400_BAD_REQUEST,
                )

    def test_post_git_commit_invalid_string(self) -> None:
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )
        for value in ("f00", "foo", "", "a" * 100):
            with self.subTest(value=value):
                response = self.post(
                    data={
                        "collection": {
                            "id": self.collection.pk,
                            "name": self.collection.name,
                            "data": {"git_commit": value},
                        },
                        "items": [],
                        "dry_run": True,
                    }
                )
                self.assertResponseProblem(
                    response,
                    "Invalid git commit",
                    detail_pattern="git_commit is malformed",
                    status_code=status.HTTP_400_BAD_REQUEST,
                )

    def test_post_valid_git_commit(self) -> None:
        self.playground.create_group_role(
            self.scenario.workspace, Workspace.Roles.OWNER, [self.scenario.user]
        )
        commit = "bb07924dd93295740939e66db5a23777439c7c51"
        self.collection.data["git_commit"] = "0" * 40
        self.collection.save()

        data = self.collection.data.copy()
        data["git_commit"] = commit
        response = self.post(
            data={
                "collection": {
                    "id": self.collection.pk,
                    "name": self.collection.name,
                    "data": data,
                },
                "items": [],
                "dry_run": True,
            }
        )
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data, {"added": 0, "removed": 0, "updated": 0, "unchanged": 0}
        )

        self.collection.refresh_from_db()
        self.assertEqual(self.collection.data["git_commit"], commit)

    def test_diff_items(self) -> None:
        item1 = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop", subject="item1"
        )
        item2 = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop", subject="item2"
        )
        item3 = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop", subject="item3"
        )
        item4 = DebusineTaskConfiguration(
            task_type=TaskTypes.WORKER, task_name="noop", subject="item4"
        )

        for old_items, new_items, removed, added, unchanged in (
            ([], [], [], [], 0),
            ([item1], [], [item1], [], 0),
            ([item1], [item2], [item1], [item2], 0),
            (
                [item1, item2, item3, item4],
                [item2, item4],
                [item1, item3],
                [],
                2,
            ),
            ([item1, item3], [item2, item4], [item1, item3], [item2, item4], 0),
            ([item1, item3], [item3, item4], [item1], [item4], 1),
        ):
            with self.subTest(
                old=[x.subject for x in old_items],
                new=[x.subject for x in new_items],
            ):
                # Populate the collection with old_items
                self.collection.child_items.all().delete()
                for item in old_items:
                    self.add_config(item)

                actual_removed, actual_added, actual_unchanged = (
                    TaskConfigurationCollectionView._diff_items(
                        self.collection, new_items
                    )
                )
                self.assertEqual(
                    (
                        sorted(x.name for x in actual_removed),
                        sorted(x.name() for x in actual_added),
                        actual_unchanged,
                    ),
                    (
                        sorted(x.name() for x in removed),
                        sorted(x.name() for x in added),
                        unchanged,
                    ),
                )


class CollectionViewSetTests(TestCase):
    """Tests for :py:class:`CollectionViewSet`."""

    scenario = scenarios.DefaultContextAPI()
    sid: ClassVar[Collection]
    sid_links: ClassVar[dict[str, str | dict[str, str]]]
    trixie: ClassVar[Collection]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.sid = cls.playground.create_collection(
            "sid", CollectionCategory.SUITE, data={"test": 1}
        )
        cls.trixie = cls.playground.create_collection(
            "trixie", CollectionCategory.SUITE, data={"test": 1}
        )
        cls.sid_links = cls.get_links(cls.sid)

    @classmethod
    def get_links(
        cls, collection: Collection
    ) -> dict[str, str | dict[str, str]]:
        return {
            "self": reverse(
                "api:collection-detail",
                kwargs={
                    "workspace": collection.workspace.name,
                    "pk": collection.pk,
                },
            ),
            "webui_self": collection.get_absolute_url(),
            "webui_category": {
                "href": reverse(
                    "workspaces:collections:category_list",
                    kwargs={
                        "wname": collection.workspace.name,
                        "ccat": collection.category,
                    },
                ),
                "type": "category",
            },
        }

    def _workspace_url_kwarg(self, workspace: Workspace | str | None) -> str:
        match workspace:
            case None:
                return self.scenario.workspace.name
            case str():
                return workspace
            case _:
                return workspace.name

    def _reverse_list(
        self,
        workspace: Workspace | str | None,
        **kwargs: str,
    ) -> str:
        """Build the URL for a test request."""
        kwargs["workspace"] = self._workspace_url_kwarg(workspace)
        return reverse("api:collection-list", kwargs=kwargs)

    def _reverse_lookup(
        self,
        workspace: Workspace | str | None,
        **kwargs: str,
    ) -> str:
        """Build the URL for a test request."""
        kwargs["workspace"] = self._workspace_url_kwarg(workspace)
        return reverse("api:collection-lookup", kwargs=kwargs)

    def _reverse_detail(
        self,
        collection: int | Collection | None,
        workspace: Workspace | str | None,
        **kwargs: str,
    ) -> str:
        """Build the URL for a test request."""
        kwargs["workspace"] = self._workspace_url_kwarg(workspace)

        match collection:
            case None:
                kwargs["pk"] = str(self.sid.pk)
            case int():
                kwargs["pk"] = str(collection)
            case Collection():
                kwargs["pk"] = str(collection.pk)
            case _ as unreachable:
                assert_never(unreachable)

        return reverse("api:collection-detail", kwargs=kwargs)

    def _headers(self, user_token: bool) -> dict[str, str]:
        """Build headers for a test request."""
        match user_token:
            case True:
                return {"token": self.scenario.user_token.key}
            case False:
                return {}
            case _ as unreachable:
                assert_never(unreachable)

    def lookup(
        self,
        category: str = CollectionCategory.SUITE,
        name: str = "sid",
        workspace: Workspace | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for the lookup endpoint."""
        return self.client.get(
            self._reverse_lookup(workspace, category=category, name=name),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def list(
        self,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for a named collection."""
        return self.client.get(
            self._reverse_list(workspace),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def get(
        self,
        collection: int | Collection | None = None,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for a named collection."""
        return self.client.get(
            self._reverse_detail(collection, workspace),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def post(
        self,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | None = None,
    ) -> TestResponseType:
        """POST request."""
        return self.client.post(
            self._reverse_list(workspace),
            data=data or {},
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def put(
        self,
        collection: int | Collection | None = None,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | None = None,
    ) -> TestResponseType:
        """PUT request for a named collection."""
        return self.client.put(
            self._reverse_detail(collection, workspace),
            data=data or {},
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def patch(
        self,
        collection: int | Collection | None = None,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | None = None,
    ) -> TestResponseType:
        """PATCH request for a named collection."""
        return self.client.patch(
            self._reverse_detail(collection, workspace),
            data=data or {},
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def delete(
        self,
        collection: int | Collection | None = None,
        workspace: Workspace | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """DELETE request for a named collection."""
        return self.client.delete(
            self._reverse_detail(collection, workspace),
            headers=self._headers(user_token),
        )

    def assertSidUnchanged(self) -> None:
        """Ensure self.sid is unchanged."""
        self.sid.refresh_from_db()
        self.assertEqual(self.sid.name, "sid")
        self.assertEqual(self.sid.category, CollectionCategory.SUITE)
        self.assertIsNone(self.sid.full_history_retention_period)
        self.assertIsNone(self.sid.metadata_only_retention_period)
        self.assertEqual(self.sid.data, {"test": 1})

    def test_unauthenticated(self) -> None:
        """Authentication is required."""
        for method in ("lookup", "list", "post", "put", "patch", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(user_token=False)
                self.assertResponseProblem(
                    response,
                    "Error",
                    detail_pattern=(
                        "Authentication credentials were not provided."
                    ),
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_workspace_not_found(self) -> None:
        """Workspace must exist."""
        for method in ("lookup", "list", "post", "put", "patch", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(workspace="does-not-exist")
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace does-not-exist not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_workspace_not_accessible(self) -> None:
        """User must be able to display the workspace."""
        workspace = self.playground.create_workspace(name="private")
        for method in ("lookup", "list", "post", "put", "patch", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(workspace=workspace)
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace private not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_not_found(self) -> None:
        """The collection must exist."""
        max_id = Collection.objects.aggregate(Max('id'))['id__max']
        for method in ("get", "put", "patch", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(collection=max_id + 1)
                self.assertResponseProblem(
                    response,
                    "Object not found",
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_detail_unauthorized(self) -> None:
        """The user must be able to display the collection."""
        for method in ("get", "put", "patch", "delete"):
            with (
                self.subTest(method=method),
                override_permission(Collection, "can_display", DenyAll),
            ):
                response = getattr(self, method)()
                self.assertResponseProblem(
                    response,
                    "Object not found",
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_create_unauthorized(self) -> None:
        """The user must be able to create the collection."""
        with override_permission(Workspace, "can_create_collections", DenyAll):
            response = getattr(self, "post")()
            self.assertResponseProblem(
                response,
                "playground cannot create collections"
                f" in {self.scenario.workspace}",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_update_unauthorized(self) -> None:
        """The user must be able to update the collection."""
        for method in ("put", "patch"):
            with (
                self.subTest(method=method),
                override_permission(Collection, "can_configure", DenyAll),
            ):
                response = getattr(self, method)()
                self.assertResponseProblem(
                    response,
                    f"playground cannot configure collection {self.sid}",
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_delete_unauthorized(self) -> None:
        """The user must be able to delete the collection."""
        with override_permission(Collection, "can_delete", DenyAll):
            response = self.delete()
            self.assertResponseProblem(
                response,
                f"playground cannot delete collection {self.sid}",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_lookup(self) -> None:
        response = self.lookup()
        data = self.assertAPIResponseOk(response)
        assert data == {
            "id": self.sid.pk,
            "name": self.sid.name,
            "category": self.sid.category,
            "full_history_retention_period": None,
            "metadata_only_retention_period": None,
            "data": {"test": 1},
            "links": self.sid_links,
            "stats": None,
        }

    def test_lookup_not_found(self) -> None:
        response = self.lookup(category=CollectionCategory.SUITE, name="buster")
        self.assertResponseProblem(
            response,
            "Object not found",
            status_code=status.HTTP_404_NOT_FOUND,
        )

    def test_list(self) -> None:
        response = self.list()
        data = self.assertAPIResponseOk(response)
        names = [f"{x["name"]}@{x["category"]}" for x in data["results"]]
        assert names == [
            '_@debian:archive',
            "_@debian:package-build-logs",
            "sid@debian:suite",
            'trixie@debian:suite',
            'default@debusine:task-configuration',
            '_@debusine:task-history',
        ]

    def test_list_omits_workflow_internal(self) -> None:
        self.playground.create_collection(
            name="1234", category=CollectionCategory.WORKFLOW_INTERNAL
        )
        response = self.list()
        data = self.assertAPIResponseOk(response)
        names = [f"{x["name"]}@{x["category"]}" for x in data["results"]]
        assert names == [
            '_@debian:archive',
            "_@debian:package-build-logs",
            "sid@debian:suite",
            'trixie@debian:suite',
            'default@debusine:task-configuration',
            '_@debusine:task-history',
        ]

    def test_detail(self) -> None:
        response = self.get()
        data = self.assertAPIResponseOk(response)
        assert data == {
            "id": self.sid.pk,
            "name": self.sid.name,
            "category": self.sid.category,
            "full_history_retention_period": None,
            "metadata_only_retention_period": None,
            "data": {"test": 1},
            "links": self.sid_links,
            "stats": None,
        }

    @override_permission(Workspace, "can_create_collections", AllowAll)
    def test_post(self) -> None:
        response = self.post(
            data={
                "category": CollectionCategory.SUITE,
                "name": "new",
                "full_history_retention_period": None,
                "metadata_only_retention_period": None,
                "data": {"created": True},
            }
        )
        new_data = self.assertAPIResponseOk(
            response, status_code=status.HTTP_201_CREATED
        )
        new = Collection.objects.get(
            workspace=self.scenario.workspace,
            category=CollectionCategory.SUITE,
            name="new",
        )
        self.assertIsNone(new.full_history_retention_period)
        self.assertIsNone(new.metadata_only_retention_period)
        self.assertEqual(new.data, {"created": True})
        assert new_data == {
            "id": new.pk,
            "category": CollectionCategory.SUITE,
            "name": "new",
            "full_history_retention_period": None,
            "metadata_only_retention_period": None,
            "data": {"created": True},
            "links": self.get_links(new),
            "stats": None,
        }

    @override_permission(Workspace, "can_create_collections", AllowAll)
    def test_post_name_conflict(self) -> None:
        response = self.post(
            data={
                "category": CollectionCategory.SUITE,
                "name": "sid",
                "full_history_retention_period": None,
                "metadata_only_retention_period": None,
                "data": {},
            }
        )
        self.assertResponseProblem(
            response,
            "Cannot create collection",
            status_code=status.HTTP_400_BAD_REQUEST,
            detail_pattern="A collection with the same name already exists",
        )
        self.assertSidUnchanged()

    @override_permission(Workspace, "can_create_collections", AllowAll)
    def test_post_unexpected_error(self) -> None:
        with mock.patch(
            "debusine.server.serializers.CollectionSerializer.save",
            side_effect=IntegrityError("unexpected message"),
        ):
            response = self.post(
                data={
                    "category": CollectionCategory.SUITE,
                    "name": "sid",
                    "full_history_retention_period": None,
                    "metadata_only_retention_period": None,
                    "data": {},
                }
            )
        self.assertResponseProblem(
            response,
            "Cannot create collection",
            status_code=status.HTTP_400_BAD_REQUEST,
        )
        self.assertSidUnchanged()

    @override_permission(Workspace, "can_create_collections", AllowAll)
    def test_post_invalid_name(self) -> None:
        response = self.post(
            data={
                "category": CollectionCategory.SUITE,
                "name": "sid/test",
                "full_history_retention_period": None,
                "metadata_only_retention_period": None,
                "data": {},
            }
        )
        self.assertResponseProblem(
            response,
            "Cannot deserialize collection",
            validation_errors_pattern=(
                "'sid/test' is not a valid collection name"
            ),
            status_code=status.HTTP_400_BAD_REQUEST,
        )
        self.assertFalse(Collection.objects.filter(name="sid/test").exists())

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_noop(self) -> None:
        response = self.patch(data={})
        new_data = self.assertAPIResponseOk(response)
        assert new_data == {
            "id": self.sid.pk,
            "category": self.sid.category,
            "name": self.sid.name,
            "full_history_retention_period": None,
            "metadata_only_retention_period": None,
            "data": self.sid.data,
            "links": self.sid_links,
            "stats": None,
        }
        self.assertSidUnchanged()

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_noop_all_set(self) -> None:
        response = self.patch(
            data={
                "name": "sid",
                "full_history_retention_period": None,
                "metadata_only_retention_period": None,
                "data": {"test": 1},
            },
        )
        new_data = self.assertAPIResponseOk(response)
        assert new_data == {
            "id": self.sid.pk,
            "category": self.sid.category,
            "name": self.sid.name,
            "full_history_retention_period": None,
            "metadata_only_retention_period": None,
            "data": self.sid.data,
            "links": self.sid_links,
            "stats": None,
        }
        self.assertSidUnchanged()

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_rename(self) -> None:
        response = self.patch(data={"name": "renamed"})
        new_data = self.assertAPIResponseOk(response)
        self.sid.refresh_from_db()
        assert new_data == {
            "id": self.sid.pk,
            "category": self.sid.category,
            "name": "renamed",
            "full_history_retention_period": None,
            "metadata_only_retention_period": None,
            "data": self.sid.data,
            "links": self.get_links(self.sid),
            "stats": None,
        }
        self.assertEqual(self.sid.name, "renamed")
        self.assertEqual(self.sid.category, CollectionCategory.SUITE)
        self.assertIsNone(self.sid.full_history_retention_period)
        self.assertIsNone(self.sid.metadata_only_retention_period)
        self.assertEqual(self.sid.data, {"test": 1})

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_change_retention_periods(self) -> None:
        response = self.patch(
            data={
                "full_history_retention_period": 123,
                "metadata_only_retention_period": 456,
            }
        )
        new_data = self.assertAPIResponseOk(response)
        assert new_data == {
            "id": self.sid.pk,
            "category": self.sid.category,
            "name": self.sid.name,
            "full_history_retention_period": 123,
            "metadata_only_retention_period": 456,
            "data": self.sid.data,
            "links": self.sid_links,
            "stats": None,
        }
        self.sid.refresh_from_db()
        self.assertEqual(self.sid.name, "sid")
        self.assertEqual(self.sid.category, CollectionCategory.SUITE)
        self.assertEqual(
            self.sid.full_history_retention_period,
            dt.timedelta(days=123),
        )
        self.assertEqual(
            self.sid.metadata_only_retention_period,
            dt.timedelta(days=456),
        )
        self.assertEqual(self.sid.data, {"test": 1})

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_change_data(self) -> None:
        response = self.patch(data={"data": {"changed": True}})
        new_data = self.assertAPIResponseOk(response)
        assert new_data == {
            "id": self.sid.pk,
            "category": self.sid.category,
            "name": self.sid.name,
            "full_history_retention_period": None,
            "metadata_only_retention_period": None,
            "data": {"changed": True},
            "links": self.sid_links,
            "stats": None,
        }
        self.sid.refresh_from_db()
        self.assertEqual(self.sid.name, "sid")
        self.assertEqual(self.sid.category, CollectionCategory.SUITE)
        self.assertIsNone(self.sid.full_history_retention_period)
        self.assertIsNone(self.sid.metadata_only_retention_period)
        self.assertEqual(self.sid.data, {"changed": True})

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_change_all(self) -> None:
        response = self.patch(
            data={
                "name": "renamed",
                "full_history_retention_period": 123,
                "metadata_only_retention_period": 456,
                "data": {"changed": True},
            }
        )
        new_data = self.assertAPIResponseOk(response)
        self.sid.refresh_from_db()
        assert new_data == {
            "id": self.sid.pk,
            "name": "renamed",
            "category": self.sid.category,
            "full_history_retention_period": 123,
            "metadata_only_retention_period": 456,
            "data": {"changed": True},
            "links": self.get_links(self.sid),
            "stats": None,
        }
        self.assertEqual(self.sid.name, "renamed")
        self.assertEqual(
            self.sid.full_history_retention_period,
            dt.timedelta(days=123),
        )
        self.assertEqual(
            self.sid.metadata_only_retention_period,
            dt.timedelta(days=456),
        )
        self.assertEqual(self.sid.data, {"changed": True})

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_cannot_change_category(self) -> None:
        response = self.patch(
            data={
                "name": "renamed",
                "category": "debusine:test",
                "full_history_retention_period": 123,
                "metadata_only_retention_period": 456,
                "data": {"changed": True},
            }
        )
        self.assertResponseProblem(
            response,
            "The category of a collection cannot be changed",
            status_code=status.HTTP_400_BAD_REQUEST,
        )
        self.assertSidUnchanged()

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_rename_conflict(self) -> None:
        self.playground.create_collection("renamed", CollectionCategory.SUITE)

        response = self.patch(data={"name": "renamed"})
        self.assertResponseProblem(
            response,
            "New name conflicts with an existing collection",
            detail_pattern=(
                "A collection with category 'debian:suite'"
                " called 'renamed' already exists"
            ),
            status_code=status.HTTP_400_BAD_REQUEST,
        )
        self.assertSidUnchanged()

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_rename_no_conflict_in_other_workspaces(self) -> None:
        other = self.playground.create_workspace(name="other")
        self.playground.create_collection(
            "renamed", CollectionCategory.SUITE, workspace=other
        )

        response = self.patch(data={"name": "renamed"})
        new_data = self.assertAPIResponseOk(response)
        self.sid.refresh_from_db()
        assert new_data == {
            "id": self.sid.pk,
            "category": self.sid.category,
            "name": "renamed",
            "full_history_retention_period": (
                self.sid.full_history_retention_period
            ),
            "metadata_only_retention_period": (
                self.sid.metadata_only_retention_period
            ),
            "data": self.sid.data,
            "links": self.get_links(self.sid),
            "stats": None,
        }

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_rename_singleton_ok(self) -> None:
        collection = self.scenario.workspace.get_singleton_collection(
            pc=self.scenario.context_ui(), category=CollectionCategory.ARCHIVE
        )
        response = self.patch(
            collection=collection,
            data={"name": collection.name},
        )
        new_data = self.assertAPIResponseOk(response)
        assert new_data == {
            "id": collection.pk,
            "category": collection.category,
            "name": collection.name,
            "full_history_retention_period": (
                collection.full_history_retention_period
            ),
            "metadata_only_retention_period": (
                collection.metadata_only_retention_period
            ),
            "data": collection.data,
            "links": self.get_links(collection),
            "stats": None,
        }

    @override_permission(Collection, "can_configure", AllowAll)
    def test_patch_rename_singleton_invalid_name(self) -> None:
        collection = self.scenario.workspace.get_singleton_collection(
            pc=self.scenario.context_ui(), category=CollectionCategory.ARCHIVE
        )
        response = self.patch(
            collection=collection,
            data={"name": "renamed"},
        )
        self.assertResponseProblem(
            response,
            "Invalid name for singleton collection",
            detail_pattern=(
                "'debian:archive' singleton collections must be named '_'"
            ),
            status_code=status.HTTP_400_BAD_REQUEST,
        )
        collection.refresh_from_db()
        self.assertEqual(collection.name, "_")

    @override_permission(Collection, "can_delete", AllowAll)
    def test_delete(self) -> None:
        response = self.delete(collection=self.sid)
        new_data = self.assertAPIResponseOk(
            response, status_code=status.HTTP_204_NO_CONTENT
        )
        self.assertIsNone(new_data)
        self.assertFalse(Collection.objects.filter(pk=self.sid.pk).exists())
        self.assertTrue(Collection.objects.filter(pk=self.trixie.pk).exists())

    def test_debusine_collection_show(self) -> None:
        debusine = self.get_debusine()
        result = debusine.collection_get(
            self.scenario.workspace.name, CollectionCategory.SUITE, "sid"
        )
        self.assertIsNone(result.stats)
        result = debusine.collection_get(
            self.scenario.workspace.name,
            CollectionCategory.SUITE,
            "sid",
            with_stats=True,
        )
        self.assertEqual(result.stats, [])

    def test_debusine_collection_iter(self) -> None:
        debusine = self.get_debusine()
        result = debusine.collection_iter(self.scenario.workspace.name)
        self.assertCountEqual(
            [(c.category, c.name) for c in result],
            [
                (CollectionCategory.PACKAGE_BUILD_LOGS, "_"),
                (CollectionCategory.TASK_HISTORY, "_"),
                (CollectionCategory.TASK_CONFIGURATION, "default"),
                (CollectionCategory.ARCHIVE, "_"),
                (CollectionCategory.SUITE, "sid"),
                (CollectionCategory.SUITE, "trixie"),
            ],
        )

    def test_debusine_collection_get(self) -> None:
        debusine = self.get_debusine()
        result = debusine.collection_get(
            self.scenario.workspace.name, self.sid.category, self.sid.name
        )
        self.assertEqual(result.id, self.sid.id)
        self.assertEqual(result.name, self.sid.name)
        self.assertEqual(result.category, self.sid.category)
        self.assertEqual(result.data, self.sid.data)

    def test_debusine_collection_get_not_found(self) -> None:
        debusine = self.get_debusine()
        with self.assertRaisesRegex(DebusineError, r"Object not found"):
            debusine.collection_get(
                self.scenario.workspace.name,
                self.sid.category,
                "does-not-exist",
            )

    @override_permission(Workspace, "can_create_collections", AllowAll)
    def test_debusine_create_collection(self) -> None:
        debusine = self.get_debusine()
        result = debusine.collection_create(
            self.scenario.workspace.name,
            CollectionDataNew(
                name="new",
                category=CollectionCategory.SUITE,
                full_history_retention_period=12,
                data={"created": True},
            ),
        )
        new = Collection.objects.get(
            workspace=self.scenario.workspace,
            category=CollectionCategory.SUITE,
            name="new",
        )
        self.assertEqual(result.id, new.pk)
        self.assertEqual(result.name, "new")
        self.assertEqual(result.category, CollectionCategory.SUITE)
        self.assertEqual(result.full_history_retention_period, 12)
        self.assertIsNone(result.metadata_only_retention_period)
        self.assertEqual(result.data, {"created": True})

        self.assertEqual(
            new.full_history_retention_period, dt.timedelta(days=12)
        )
        self.assertIsNone(new.metadata_only_retention_period)
        self.assertEqual(new.data, {"created": True})

    @override_permission(Collection, "can_configure", AllowAll)
    def test_debusine_collection_update(self) -> None:
        debusine = self.get_debusine()
        result = debusine.collection_update(
            self.scenario.workspace.name,
            CollectionData(
                id=self.sid.id,
                name=self.sid.name,
                category=self.sid.category,
                full_history_retention_period=3,
                data=self.sid.data,
            ),
        )
        self.assertEqual(result.id, self.sid.id)
        self.assertEqual(result.name, self.sid.name)
        self.assertEqual(result.category, self.sid.category)
        self.assertEqual(result.full_history_retention_period, 3)
        self.assertEqual(result.data, self.sid.data)
        self.sid.refresh_from_db()
        self.assertEqual(
            self.sid.full_history_retention_period, dt.timedelta(days=3)
        )

    @override_permission(Collection, "can_delete", AllowAll)
    def test_debusine_collection_delete(self) -> None:
        debusine = self.get_debusine()
        debusine.collection_delete(
            self.scenario.workspace.name,
            CollectionData(
                id=self.sid.id,
                name=self.sid.name,
                category=self.sid.category,
                full_history_retention_period=3,
                data=self.sid.data,
            ),
        )
        self.assertFalse(Collection.objects.filter(pk=self.sid.pk).exists())

    @override_permission(Collection, "can_delete", DenyAll)
    def test_debusine_collection_delete_not_allowed(self) -> None:
        debusine = self.get_debusine()
        with self.assertRaisesRegex(
            DebusineError,
            "playground cannot delete collection sid@debian:suite",
        ):
            debusine.collection_delete(
                self.scenario.workspace.name,
                CollectionData(
                    id=self.sid.id,
                    name=self.sid.name,
                    category=self.sid.category,
                    full_history_retention_period=3,
                    data=self.sid.data,
                ),
            )
        self.assertTrue(Collection.objects.filter(pk=self.sid.pk).exists())


class CollectionTokenGrantViewSetTests(TestCase):
    """Tests for :py:class:`CollectionTokenGrantViewSet`."""

    scenario = scenarios.DefaultContextAPI()
    sid: ClassVar[Collection]
    tokens: ClassVar[list[Token]]
    grants: ClassVar[list[CollectionTokenGrant]]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.sid = cls.playground.create_collection(
            "sid", CollectionCategory.SUITE
        )
        cls.tokens = [
            cls.playground.create_external_user_token(
                username=username, comment=comment
            )
            for username, comment in (("test1", ""), ("test2", "Test comment"))
        ]
        cls.grants = [
            cls.sid.grant_role(
                token, Collection.Roles.VIEWER, created_by=cls.scenario.user
            )
            for token in cls.tokens
        ]

    def _collection_url_kwargs(
        self, collection: Collection | str | None
    ) -> dict[str, str]:
        match collection:
            case None:
                workspace_name = self.sid.workspace.name
                collection_name = self.sid.name
                collection_category = self.sid.category
            case str():
                workspace_name, collection_lookup = collection.split("/", 1)
                assert "@" in collection_lookup
                collection_name, collection_category = collection_lookup.rsplit(
                    "@", 1
                )
            case _:
                workspace_name = collection.workspace.name
                collection_name = collection.name
                collection_category = collection.category
        return {
            "workspace": workspace_name,
            "ccat": collection_category,
            "cname": collection_name,
        }

    def _reverse_list(
        self, collection: Collection | str | None, **kwargs: str
    ) -> str:
        """Build the URL for a test request."""
        kwargs.update(self._collection_url_kwargs(collection))
        return reverse("api:collection-token-grant-list", kwargs=kwargs)

    def _reverse_detail(
        self,
        grant: (
            CollectionTokenGrant | tuple[str, Collection.Roles] | None
        ) = None,
        collection: Collection | str | None = None,
        **kwargs: str,
    ) -> str:
        """Build the URL for a test request."""
        match grant:
            case None:
                kwargs.update(
                    self._collection_url_kwargs(collection or self.sid)
                )
                kwargs["token"] = self.grants[0].token.hash
                kwargs["role"] = self.grants[0].role
            case (str() as token_hash, Collection.Roles() as role):
                kwargs.update(
                    self._collection_url_kwargs(collection or self.sid)
                )
                kwargs["token"] = token_hash
                kwargs["role"] = role
            case CollectionTokenGrant():
                kwargs.update(
                    self._collection_url_kwargs(collection or grant.collection)
                )
                kwargs["token"] = grant.token.hash
                kwargs["role"] = grant.role
            case _ as unreachable:
                raise AssertionError(f"Unexpected grant: {unreachable}")
        return reverse("api:collection-token-grant-detail", kwargs=kwargs)

    def _headers(self, user_token: bool) -> dict[str, str]:
        """Build headers for a test request."""
        match user_token:
            case True:
                return {"token": self.scenario.user_token.key}
            case False:
                return {}
            case _ as unreachable:
                assert_never(unreachable)

    def list(
        self,
        collection: Collection | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for all token grants."""
        return self.client.get(
            self._reverse_list(collection=collection),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def get(
        self,
        grant: (
            CollectionTokenGrant | tuple[str, Collection.Roles] | None
        ) = None,
        collection: Collection | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """GET request for a particular token grant."""
        return self.client.get(
            self._reverse_detail(grant=grant, collection=collection),
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def post(
        self,
        collection: Collection | str | None = None,
        user_token: bool = True,
        data: dict[str, Any] | None = None,
    ) -> TestResponseType:
        """POST request."""
        return self.client.post(
            self._reverse_list(collection=collection),
            data=data or {},
            content_type="application/json",
            headers=self._headers(user_token),
        )

    def delete(
        self,
        grant: (
            CollectionTokenGrant | tuple[str, Collection.Roles] | None
        ) = None,
        collection: Collection | str | None = None,
        user_token: bool = True,
    ) -> TestResponseType:
        """DELETE request for a named collection."""
        return self.client.delete(
            self._reverse_detail(grant=grant, collection=collection),
            headers=self._headers(user_token),
        )

    def test_unauthenticated(self) -> None:
        """Authentication is required."""
        for method in ("list", "get", "post", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(user_token=False)
                self.assertResponseProblem(
                    response,
                    "Error",
                    detail_pattern=(
                        "Authentication credentials were not provided."
                    ),
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_workspace_not_found(self) -> None:
        """Workspace must exist."""
        for method in ("list", "get", "post", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(
                    collection=f"does-not-exist/sid@{CollectionCategory.SUITE}"
                )
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace does-not-exist not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_workspace_not_visible(self) -> None:
        """User must be able to view the workspace."""
        workspace = self.playground.create_workspace(name="private")
        collection = self.playground.create_collection(
            name="test", category=CollectionCategory.TEST, workspace=workspace
        )
        for method in ("list", "get", "post", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(collection=collection)
                self.assertResponseProblem(
                    response,
                    "Workspace not found",
                    detail_pattern=(
                        "Workspace private not found in scope debusine"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_collection_not_found(self) -> None:
        """Collection must exist."""
        collection_name = f"nonexistent@{CollectionCategory.SUITE}"
        for method in ("list", "get", "post", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(
                    collection=(
                        f"{self.scenario.workspace.name}/{collection_name}"
                    )
                )
                self.assertResponseProblem(
                    response,
                    "Collection not found",
                    detail_pattern=(
                        f"Collection {collection_name!r} does not exist in"
                        f" workspace {self.scenario.workspace}"
                    ),
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_collection_not_visible(self) -> None:
        """User must be able to view the collection."""
        for method in ("list", "get", "post", "delete"):
            with (
                self.subTest(method=method),
                override_permission(Collection, "can_display", DenyAll),
            ):
                response = getattr(self, method)()
                self.assertResponseProblem(
                    response,
                    f"{self.scenario.user} cannot display collection"
                    f" {self.sid.name}@{self.sid.category}",
                    status_code=status.HTTP_403_FORBIDDEN,
                )

    def test_grant_nonexistent_hash(self) -> None:
        """The grant's token hash must exist."""
        for method in ("get", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(
                    grant=("nonexistent", Collection.Roles.VIEWER)
                )
                self.assertResponseProblem(
                    response,
                    "Object not found",
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_grant_nonexistent_role(self) -> None:
        """The grant's role must exist."""
        for method in ("get", "delete"):
            with self.subTest(method=method):
                response = getattr(self, method)(
                    grant=(self.grants[0].token.hash, Collection.Roles.OWNER)
                )
                self.assertResponseProblem(
                    response,
                    "Object not found",
                    status_code=status.HTTP_404_NOT_FOUND,
                )

    def test_create_unauthorized(self) -> None:
        """The user must be able to create the grant."""
        with override_permission(Collection, "can_configure", DenyAll):
            response = self.post(
                data={"username": "test", "role": Collection.Roles.VIEWER}
            )
            self.assertResponseProblem(
                response,
                f"{self.scenario.user} cannot configure collection"
                f" {self.sid.name}@{self.sid.category}",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_delete_unauthorized(self) -> None:
        """The user must be able to delete the grant."""
        with override_permission(Collection, "can_configure", DenyAll):
            response = self.delete()
            self.assertResponseProblem(
                response,
                f"{self.scenario.user} cannot configure collection"
                f" {self.sid.name}@{self.sid.category}",
                status_code=status.HTTP_403_FORBIDDEN,
            )

    def test_list(self) -> None:
        response = self.list()
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data["results"],
            [
                {
                    "token_type": Token.Types.EXTERNAL_USER,
                    "token_hash": self.tokens[0].hash,
                    "username": "test1",
                    "role": Collection.Roles.VIEWER,
                    "comment": "",
                },
                {
                    "token_type": Token.Types.EXTERNAL_USER,
                    "token_hash": self.tokens[1].hash,
                    "username": "test2",
                    "role": Collection.Roles.VIEWER,
                    "comment": "Test comment",
                },
            ],
        )

    def test_detail(self) -> None:
        response = self.get()
        data = self.assertAPIResponseOk(response)
        self.assertEqual(
            data,
            {
                "token_type": Token.Types.EXTERNAL_USER,
                "token_hash": self.tokens[0].hash,
                "username": "test1",
                "role": Collection.Roles.VIEWER,
                "comment": "",
            },
        )

    @override_permission(Collection, "can_configure", AllowAll)
    def test_post_token_hash(self) -> None:
        token = self.playground.create_external_user_token(username="test-new")
        response = self.post(
            data={"token_hash": token.hash, "role": Collection.Roles.VIEWER}
        )
        new_data = self.assertAPIResponseOk(
            response, status_code=status.HTTP_201_CREATED
        )
        self.assertEqual(
            new_data,
            {
                "token_type": Token.Types.EXTERNAL_USER,
                "token_hash": token.hash,
                "username": "test-new",
                "role": Collection.Roles.VIEWER,
                "comment": "",
            },
        )
        self.assertTrue(
            token.collection_grants.filter(
                collection=self.sid, role=Collection.Roles.VIEWER
            ).exists()
        )

    @override_permission(Collection, "can_configure", AllowAll)
    def test_post_username(self) -> None:
        response = self.post(
            data={
                "username": "test-new",
                "role": Collection.Roles.VIEWER,
                "comment": "Very important user",
            }
        )
        new_data = self.assertAPIResponseOk(
            response, status_code=status.HTTP_201_CREATED
        )
        token_key = new_data.pop("token_key")
        token = Token.objects.get_token_or_none(token_key)
        assert token is not None
        self.assertEqual(
            new_data,
            {
                "token_type": Token.Types.EXTERNAL_USER,
                "token_hash": token.hash,
                "username": "test-new",
                "role": Collection.Roles.VIEWER,
                "comment": "Very important user",
            },
        )
        self.assertTrue(token.enabled)
        self.assertTrue(
            token.collection_grants.filter(
                collection=self.sid, role=Collection.Roles.VIEWER
            ).exists()
        )

    @override_permission(Collection, "can_configure", AllowAll)
    def test_post_invalid_username(self) -> None:
        response = self.post(
            data={"username": "foo bar&baz", "role": Collection.Roles.VIEWER}
        )
        self.assertResponseProblem(
            response,
            "Cannot grant collection token",
            status_code=status.HTTP_400_BAD_REQUEST,
            validation_errors_pattern="Enter a valid username",
        )

    @override_permission(Collection, "can_configure", AllowAll)
    def test_delete(self) -> None:
        response = self.delete(grant=self.grants[0])
        new_data = self.assertAPIResponseOk(
            response, status_code=status.HTTP_204_NO_CONTENT
        )
        self.assertIsNone(new_data)
        self.assertFalse(
            self.sid.token_grants.filter(pk=self.grants[0].pk).exists()
        )
        self.assertTrue(
            self.sid.token_grants.filter(pk=self.grants[1].pk).exists()
        )
