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

"""Common test-helper code involving django usage."""

import abc
import asyncio
import contextlib
import copy
import datetime as dt
import http.client
import inspect
import io
import itertools
import json
import traceback
import unittest
from collections.abc import Callable, Generator, Iterator, Sequence
from collections.abc import Collection as AbcCollection
from operator import itemgetter
from pathlib import Path
from typing import (
    Any,
    ClassVar,
    Protocol,
    TYPE_CHECKING,
    cast,
    override,
    runtime_checkable,
)
from unittest import mock
from unittest.util import safe_repr

import django.template
import django.test
import lxml
import requests
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.conf import settings
from django.core.signals import request_finished
from django.db import (
    OperationalError,
    close_old_connections,
    connection,
    connections,
    transaction,
)
from django.db.models import Model, QuerySet
from django.http import HttpRequest
from django.http.response import HttpResponseBase
from django.template import engines
from django.test.client import JSON_CONTENT_TYPE_RE
from django.urls import set_urlconf
from django.utils import timezone
from rest_framework import status

from debusine.db.context import ContextConsistencyError, context
from debusine.db.models import (
    Collection,
    CollectionItemMatchConstraint,
    Scope,
    Token,
    User,
    WorkRequest,
    Workspace,
)
from debusine.db.models.permissions import (
    Allow,
    PermissionCheckPredicate,
    get_resource_scope,
    get_resource_workspace,
    resolve_roles_list,
)
from debusine.db.models.work_requests import WorkRequestQuerySet
from debusine.db.permissioncontext import PermissionContext, PermissionUser
from debusine.db.playground.scenarios import Scenario
from debusine.db.tasks import DBTask, DBWorkerTask
from debusine.tasks.models import (
    BaseDynamicTaskData,
    BaseTaskData,
    BaseTaskDataModel,
)
from debusine.test.playground import Playground

# Relative import needed to break an import loop
from .base import TestCase as BaseTestCase  # noqa: ABS101, TID252

if TYPE_CHECKING:
    from django.test.client import _MonkeyPatchedWSGIResponse

    TestMixinBase = unittest.TestCase
else:
    TestMixinBase = object

#: Type alias for response objects returned by the Django test client
type TestResponseType = "_MonkeyPatchedWSGIResponse"
#: Type alias for any response type
type AnyResponseType = HttpResponseBase | "_MonkeyPatchedWSGIResponse"


@contextlib.contextmanager
def preserve_db_task_registry() -> Generator[None]:
    """Make task registry changes ephemeral."""
    db_orig = DBTask._sub_tasks
    DBTask._sub_tasks = copy.deepcopy(DBTask._sub_tasks)
    try:
        yield
    finally:
        DBTask._sub_tasks = db_orig


class BaseTestWorkerTask[TD: BaseTaskData](
    DBWorkerTask[TD, BaseDynamicTaskData], metaclass=abc.ABCMeta
):
    """Base class for simple worker tasks used in test code."""

    @override
    def get_task_configuration_subject_context(
        self,
    ) -> tuple[str | None, str | None]:
        raise NotImplementedError()

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


class PermissionOverride[M: Model](abc.ABC):
    """Base class for configurable permission overrides for tests."""

    @abc.abstractmethod
    def check(self, obj: M, user: PermissionUser | None) -> bool:
        """Test the predicate on an object."""

    @abc.abstractmethod
    def filter(
        self, queryset: QuerySet[M, M], user: PermissionUser | None
    ) -> QuerySet[M, M]:
        """Filter a QuerySet with the predicate."""


class AllowAll[M: Model](PermissionOverride[M]):
    """Permission override allowing all."""

    @override
    def check(self, obj: M, user: PermissionUser | None) -> bool:  # noqa: ARG002, U100
        return True

    @override
    def filter(
        self,
        queryset: QuerySet[M, M],
        user: PermissionUser | None,  # noqa: ARG002, U100
    ) -> QuerySet[M, M]:
        return queryset


class DenyAll[M: Model](PermissionOverride[M]):
    """Permission override denying all."""

    @override
    def check(self, obj: M, user: PermissionUser | None) -> bool:  # noqa: ARG002, U100
        return False

    @override
    def filter(
        self,
        queryset: QuerySet[M, M],
        user: PermissionUser | None,  # noqa: ARG002, U100
    ) -> QuerySet[M, M]:
        return queryset.none()


class ListFilter[M: Model](PermissionOverride[M]):
    """Permission override that includes or excludes specific items."""

    def __init__(
        self, include: list[M] | None = None, exclude: list[M] | None = None
    ) -> None:
        """Store filter arguments."""
        super().__init__()
        self._include = include
        self._exclude = exclude

    @override
    def check(self, obj: M, user: PermissionUser | None) -> bool:  # noqa: ARG002, U100
        if self._include is not None:
            return obj in self._include
        if self._exclude is not None:
            return obj not in self._exclude
        return True

    @override
    def filter(
        self,
        queryset: QuerySet[M, M],
        user: PermissionUser | None,  # noqa: ARG002, U100
    ) -> QuerySet[M, M]:
        res = queryset
        if self._include is not None:
            res = res.filter(pk__in=[x.pk for x in self._include])
        if self._exclude is not None:
            res = res.exclude(pk__in=[x.pk for x in self._exclude])
        return res


def set_context_for_resource(resource: Model, user: PermissionUser) -> None:
    """Set the current context to fit the resource."""
    scope = get_resource_scope(resource)
    assert scope is not None
    context.set_permission_context_scoped(
        scope,
        (
            PermissionContext.for_ui_user(user)
            if user.is_authenticated
            else PermissionContext.for_ui_anonymous()
        ),
    )
    if workspace := get_resource_workspace(resource):
        workspace.set_current()


@contextlib.contextmanager
def override_permission[M: Model](
    model_cls: type[M],
    pred: str,
    override_cls: type[PermissionOverride[M]],
    **kwargs: Any,
) -> Generator[None]:
    """
    Override a permission.

    Example usage::

        with override_permission(
            Workspace, "can_display", AllowAll
        ):
            ...

    Or also::

        @override_permission(
            Workspace, "can_display", AllowAll
        )
        def test_method(self) -> None:
            ...
    """
    override = override_cls(**kwargs)

    # Model.objects is a manager instance generated by from_queryset, and we
    # need a bit of creativity to mock its methods
    def on_manager(user: PermissionUser | None) -> QuerySet[M, M]:
        return override.filter(getattr(model_cls, "objects").all(), user)

    manager = getattr(model_cls, "objects")
    queryset_cls = manager.get_queryset().__class__
    error_template = getattr(getattr(model_cls, pred), "error_template")
    with (
        mock.patch.object(
            model_cls, pred, autospec=True, side_effect=override.check
        ) as check_patch,
        mock.patch.object(
            manager,
            pred,
            autospec=True,
            side_effect=on_manager,
        ),
        mock.patch.object(
            queryset_cls,
            pred,
            autospec=True,
            side_effect=override.filter,
        ),
    ):
        setattr(check_patch, "error_template", error_template)
        yield


@runtime_checkable
class JSONResponseProtocol(Protocol):
    """A Django test client response with a monkey-patched json() method."""

    def json(self) -> Any:
        """Return the body of the response, parsed as JSON."""


class BaseDjangoTestCase(django.test.TransactionTestCase, BaseTestCase):
    """
    Django-specific Debusine test methods.

    This augments debusine.test.TestCase with django-specific assert statements
    and factory functions.

    This is the common base class for TestCase and TransactionTestCase, for
    tests that do depend on Django code.
    """

    @override
    @classmethod
    def get_precious_globals(cls) -> dict[str, Any]:
        """Collect global values that tests should not change."""
        res = super().get_precious_globals()
        res["templates_0_dirs"] = settings.TEMPLATES[0]["DIRS"]
        return res

    @override
    @classmethod
    def setUpClass(cls) -> None:
        """Run class-wide test setup in a temporary context."""
        with context.local():
            context.reset()
            super().setUpClass()

    @override
    def setUp(self) -> None:
        """Ensure test methods do not change the global scope."""
        super().setUp()
        self.enterContext(context.local())
        context.reset()
        set_urlconf(None)

    @override
    def tearDown(self) -> None:
        # Some tests change Django's URL configuration.  Django resets this
        # at the start and end of each request, but that doesn't necessarily
        # happen in tests.  If a test that uses reverse() runs with a
        # non-default URL configuration, then it may spuriously fail.  Reset
        # the URL configuration at the end of each test to make sure that
        # doesn't happen.
        set_urlconf(None)
        super().tearDown()

    @contextlib.contextmanager
    def custom_template(self, path: str | Path) -> Generator[Path]:
        """
        Install a custom template, removing it at context manager exit.

        It generates a path pointing to the custom template file, but nothing
        is written: you can use the path you get to write out the template
        contents.
        """
        # Clear template caches before writing the new template and after
        # removing it: if we are changing an included template, Django doesn't
        # seem able to notice that the cache needs invalidating
        path = Path(path)
        template_path = Path(settings.DEBUSINE_TEMPLATE_DIRECTORY) / path
        template_path.parent.mkdir(exist_ok=True, parents=True)
        try:
            self.clear_template_cache()
            yield template_path
        finally:
            template_path.unlink(missing_ok=True)
            self.clear_template_cache()

    def clear_template_cache(self) -> None:
        """Clear the Django template cache."""
        from django.template.backends.django import DjangoTemplates
        from django.template.loaders.base import Loader

        for engine in django.template.engines.all():
            if not isinstance(engine, DjangoTemplates):
                continue
            for loader in engine.engine.template_loaders:
                if not isinstance(loader, Loader):
                    continue
                loader.reset()

    @contextlib.contextmanager
    def ephemeral_savepoint(self) -> Generator[Any]:
        """
        Run the code inside a savepoint, rolled back at the end.

        Generates the transaction identifier.

        Rollback only happens if the code succeeded: a runaway exception is
        left to the care of the transaction inside which we are running.
        """
        sid = transaction.savepoint()
        yield sid
        transaction.savepoint_rollback(sid)

    @contextlib.contextmanager
    def separate_connection(self, alias: str) -> Generator[None]:
        """Temporarily open a separate database connection."""
        connections[alias] = temp_connection = connection.copy(alias)
        try:
            yield
        finally:
            del connections[alias]
            temp_connection.close()

    def assertResponseHTML(
        self,
        response: TestResponseType,
        status_code: int = status.HTTP_200_OK,
        content_type: str = "text/html; charset=utf-8",
        dump_on_error: bool = False,
    ) -> lxml.objectify.ObjectifiedElement:
        """
        Check that the response is valid HTML.

        :param response: the Django response to check
        :param status_code: expected status code
        :param content_type: expected content type
        :returns: the parsed HTML contents
        """
        self.assertEqual(response.status_code, status_code)
        self.assertEqual(response.headers["Content-Type"], content_type)
        return self.assertHTMLValid(
            response.content, dump_on_error=dump_on_error
        )

    def assertResponse400(
        self, response: "_MonkeyPatchedWSGIResponse", error: str
    ) -> None:
        """Assert that response is Http400 and contents is in response."""
        self.assertEqual(response.status_code, 400)
        self.assertEqual(response.context["error"], error)

    def format_api_response(self, response: AnyResponseType) -> str:
        """
        Format an API response.

        The result is intended to be compact and useful in error messages and
        exception annotations.
        """
        status_description = http.client.responses.get(
            response.status_code, 'unknown'
        )
        content_type = response.get("content-type")
        res = f"{response.status_code} {status_description} ({content_type}): "

        if hasattr(response, "content"):
            assert content_type is not None
            content = response.content.decode(response.charset)
            if content_type == "application/problem+json":
                # Format a ProblemResponse
                decoded = json.loads(content)
                title = decoded.get("title", "(no title)")
                res += f"{title=!r}"
                if detail := decoded.get("detail"):
                    res += f" {detail=!r}"
                if validation_errors := decoded.get("validation_errors"):
                    res += f" {validation_errors=!r}"
                if original_exception := getattr(
                    response, "_original_exception", None
                ):
                    res += " original exception follows:\n"
                    exception_formatted = traceback.format_exception(
                        original_exception
                    )
                    res += "".join(exception_formatted)
            elif JSON_CONTENT_TYPE_RE.match(content_type):
                content = repr(json.loads(content))
                if len(content) > 200:
                    content = content[:200] + "…"
                res += f"content={content}"
            else:
                if len(content) > 200:
                    content = content[:200] + "…"
                res += f"{content=!r}"
        else:
            res += "content not available"
        return res

    def assertAPIResponseOk(
        self, response: AnyResponseType, status_code: int = status.HTTP_200_OK
    ) -> Any:
        """
        Ensure an API response succeeded with the given status code.

        :returns: the decoded JSON contents.
        """
        try:
            self.assertEqual(response.status_code, status_code)

            if status_code == status.HTTP_204_NO_CONTENT:
                assert (
                    not hasattr(response, "content") or response.content == b""
                )
                return None
            content_type = response.get("Content-Type")
            if content_type is None:
                self.fail("response has no content type")
            if not JSON_CONTENT_TYPE_RE.match(content_type):
                self.fail(
                    f"content type {content_type!r} is not a JSON response"
                )

            assert hasattr(response, "content")
            return json.loads(response.content.decode(response.charset))
        except Exception as exc:
            exc.add_note(self.format_api_response(response))
            raise

    def assertResponseProblem(
        self,
        response: HttpResponseBase,
        title: str,
        detail_pattern: str | None = None,
        validation_errors_pattern: str | None = None,
        status_code: int = requests.codes.bad_request,
    ) -> None:
        """
        Assert that response is a valid application/problem+json.

        Assert that the content_type is application/problem+json and the
        title exist and matches title.

        :param response: response that it is asserting
        :param status_code: assert response.status_code == status_code
        :param title: exact match with response.data["title"]
        :param detail_pattern: if not None: assertRegex with
           response.data["detail"]. If None checks that response.data does not
           contain "detail".
        :param validation_errors_pattern: if not None: assertRegex with
           response.data["validation_errors"]. If None checks that
           response.data does not contain "validation_errors".
        """
        self.assertEqual(
            response.status_code,
            status_code,
            f"response status {response.status_code} != {status_code}",
        )

        content_type = response.headers["Content-Type"]
        self.assertEqual(
            content_type,
            "application/problem+json",
            f'content_type "{content_type}" != "application/problem+json"',
        )

        assert isinstance(response, JSONResponseProtocol)
        data = response.json()

        self.assertIn("title", data, '"title" not found in response')

        response_title = data["title"]
        self.assertEqual(
            response_title, title, f'title "{response_title}" != "{title}"'
        )

        if detail_pattern is not None:
            self.assertIn("detail", data, '"detail" not found in response')

            response_detail = str(data["detail"])
            self.assertRegex(
                response_detail,
                detail_pattern,
                f'Detail regexp "{detail_pattern}" did not '
                f'match "{response_detail}"',
            )
        else:
            self.assertNotIn("detail", data, '"detail" is in the response')

        if validation_errors_pattern is not None:
            self.assertIn(
                "validation_errors",
                data,
                '"validation_errors" not found in response',
            )
            self.assertIsInstance(data["validation_errors"], dict)

            response_validation_errors = str(data["validation_errors"])
            self.assertRegex(
                response_validation_errors,
                validation_errors_pattern,
                f'Validation errors regexp "{validation_errors_pattern}" did '
                f'not match "{response_validation_errors}"',
            )
        else:
            self.assertNotIn(
                "validation_errors",
                data,
                '"validation_errors" is in the response',
            )

    def assertRolePredicateResult(
        self,
        obj: Model,
        name: str,
        role: str,
        *,
        pc: PermissionContext,
        expected_when_has_role: bool = True,
    ) -> None:
        """
        Check the result of a predicate that only depends on the user's role.

        :param obj: an instance of the model. The actual value does not matter,
          as testing is done via mocking.
        :param name: predicate name
        :param role: role that the predicate should check
        :param pc: the PermissionContext to use for checking
        :param expected_when_has_role: the expected result of the predicate
          when the user has the role
        """
        model = obj.__class__
        model_name = model.__name__

        # Checks that model.name and model.objects.name exist
        perm_check = getattr(obj, name, None)
        if perm_check is None:
            self.fail(f"{model_name} has no @permission_check {name}")
        if not hasattr(perm_check, "work_request") or not hasattr(
            perm_check, "anonymous"
        ):
            self.fail(f"{model_name}.{name} is not marked as @permission_check")

        # Test the check predicate
        with mock.patch.object(
            obj, "has_role", autospec=True, return_value=False
        ) as has_role:
            self.assertFalse(perm_check(pc))
        has_role.assert_called_with(pc, role)

        with mock.patch.object(
            obj, "has_role", autospec=True, return_value=True
        ) as has_role:
            self.assertEqual(perm_check(pc), expected_when_has_role)
        has_role.assert_called_with(pc, role)

        # Test the filter predicate on manager
        manager = getattr(model, "objects")
        manager_name = f"{model_name}.objects"
        perm_filter_manager = getattr(manager, name, None)
        if perm_filter_manager is None:
            self.fail(f"{manager_name} has no @permission_filter {name}")
        if not hasattr(perm_filter_manager, "work_request") or not hasattr(
            perm_filter_manager, "anonymous"
        ):
            self.fail(
                f"{manager_name}.{name} is not marked as @permission_filter"
            )

        queryset_class = manager.get_queryset().__class__

        with mock.patch.object(
            queryset_class, "with_role", return_value=manager.filter(pk=obj.pk)
        ) as with_role:
            filtered = perm_filter_manager(pc)
            self.assertIsInstance(filtered, queryset_class)
            if expected_when_has_role:
                self.assertQuerySetEqual(filtered, [obj])
            else:
                self.assertQuerySetEqual(filtered, [])
        with_role.assert_called_with(pc, role)

        with mock.patch.object(
            queryset_class, "with_role", return_value=manager.none()
        ) as with_role:
            filtered = perm_filter_manager(pc)
            self.assertIsInstance(filtered, queryset_class)
            self.assertQuerySetEqual(filtered, [])
        with_role.assert_called_with(pc, role)

        # Test the filter predicate on QuerySet
        queryset = getattr(model, "objects").all()
        queryset_name = f"QuerySet[{model_name}]"

        perm_filter_queryset = getattr(queryset, name, None)
        if perm_filter_queryset is None:
            self.fail(f"{queryset_name} has no @permission_filter {name}")
        if not hasattr(perm_filter_queryset, "work_request") or not hasattr(
            perm_filter_queryset, "anonymous"
        ):
            self.fail(
                f"{queryset_name}.{name} is not marked as @permission_filter"
            )

        with mock.patch.object(
            queryset, "with_role", return_value=queryset.filter(pk=obj.pk)
        ) as with_role:
            filtered = perm_filter_queryset(pc)
            self.assertIsInstance(filtered, queryset_class)
            if expected_when_has_role:
                self.assertQuerySetEqual(filtered, [obj])
            else:
                self.assertQuerySetEqual(filtered, [])
        with_role.assert_called_with(pc, role)

        with mock.patch.object(
            queryset, "with_role", return_value=queryset.none()
        ) as with_role:
            filtered = perm_filter_queryset(pc)
            self.assertIsInstance(filtered, queryset_class)
            self.assertQuerySetEqual(filtered, [])
        with_role.assert_called_with(pc, role)

    def assertRolePredicate(
        self,
        obj: Model,
        name: str,
        role: str,
        *,
        work_request: Allow = Allow.NEVER,
        anonymous: Allow = Allow.NEVER,
    ) -> None:
        """
        Test a predicate only checks if the user has a role.

        :param obj: an instance of the model. The actual value does not matter,
                    as testing is done via mocking.
        :param name: predicate name
        :param role: role that the predicate should check
        """
        user = User()
        pc = PermissionContext.for_ui_user(user)
        self.assertRolePredicateResult(
            obj, name, role, pc=pc, expected_when_has_role=True
        )

        model = obj.__class__

        # Check that the work_request= and anonymous= arguments match

        perm_check = getattr(obj, name)
        self.assertEqual(perm_check.work_request, work_request)
        self.assertEqual(perm_check.anonymous, anonymous)

        manager = getattr(model, "objects")
        perm_filter_manager = getattr(manager, name)
        self.assertEqual(perm_filter_manager.work_request, work_request)
        self.assertEqual(perm_filter_manager.anonymous, anonymous)

        queryset = getattr(model, "objects").all()
        perm_filter_queryset = getattr(queryset, name)
        self.assertEqual(perm_filter_queryset.work_request, work_request)
        self.assertEqual(perm_filter_queryset.anonymous, anonymous)

    def assertPermissionPredicate(
        self,
        pred: Callable[[PermissionContext], bool],
        pc: PermissionContext,
        expected_pass: bool,
    ) -> None:
        """
        Check the result of a permission predicate.

        :param pred: the permission predicate, as the check method on a
                     resource object
        :param user: the user to check
        :param expected_pass: True if the predicate is expected to pass
        """
        resource: Model = pred.__self__  # type: ignore[attr-defined]
        self.assertEqual(pred(pc), expected_pass)
        filtered = getattr(
            resource.__class__.objects,  # type: ignore[attr-defined]
            pred.__name__,
        )(pc)
        if expected_pass:
            self.assertIn(resource, list(filtered))
        else:
            self.assertNotIn(resource, list(filtered))

    def _assertPermission_context(
        self, user: PermissionUser | None = None, token: Token | None = None
    ) -> PermissionContext:
        if token is None:
            assert user is not None
            if user.is_authenticated:
                return PermissionContext.for_ui_user(user)
            else:
                return PermissionContext.for_ui_anonymous()

        match token.token_type:
            case Token.Types.WORKER_ACTIVATION:
                return PermissionContext.for_worker_activation(token)
            case Token.Types.WORKER:
                return PermissionContext.for_worker(token)
            case Token.Types.WORK_REQUEST:
                return PermissionContext.for_work_request(token=token)
            case Token.Types.USER:
                return PermissionContext.for_api_user(token)
            case Token.Types.EXTERNAL_USER:
                return PermissionContext.for_api_external_user(token)
            case _ as unreachable:
                # Enforced by database constraint.
                raise AssertionError(f"Unexpected token type: {unreachable}")

    @context.local()
    def _test_check_predicate[M: Model](
        self,
        pc: PermissionContext,
        obj: M,
        name: str,
        expected: bool,
    ) -> list[str]:
        """Test a single check predicate on an object."""
        errors: list[str] = []
        predicate = getattr(obj, name)

        # Test on an empty context
        context.reset()

        if predicate(pc) is not expected:
            errors.append(
                f"{name} is {not expected} for {pc} on {obj} with empty context"
            )

        # Try with scope set in context, to check shortcut test code
        if (scope := get_resource_scope(obj)) is None:
            return errors

        try:
            context.set_permission_context_scoped(scope, pc)
            if predicate(context.pc) is not expected:
                errors.append(
                    f"{name} is {not expected} for {pc} on {obj}"
                    " with scope set in context"
                )

            if (workspace := get_resource_workspace(obj)) is None:
                return errors

            # Try with also workspace set in context, to check shortcut test
            # code
            workspace.set_current()
            if predicate(context.pc) is not expected:
                errors.append(
                    f"{name} is {not expected} for {pc} on {obj}"
                    " with scope and workspace set in context"
                )
        except ContextConsistencyError:
            pass

        return errors

    def _check_predicate_filter_output(
        self,
        prefix: str,
        filtered: list[str],
        expected: list[str],
        errors: list[str],
    ) -> None:
        if filtered == expected:
            return
        error = prefix
        if filtered:
            error += ", ".join(filtered)
        else:
            error += "nothing"
        error += " instead of "
        if expected:
            error += ", ".join(expected)
        else:
            error += "nothing"
        errors.append(error)

    @context.local()
    def assertPermission[M: Model](
        self,
        name: str,
        users: PermissionUser | None | AbcCollection[PermissionUser | None],
        allowed: M | AbcCollection[M] = (),
        denied: M | AbcCollection[M] = (),
        token: Token | None = None,
    ) -> None:
        """
        Test a permission predicate.

        For each user given, test that the model objects in ``allowed`` allow
        the named permission, and those in ``denied`` do not.

        Both check and filter predicates are tested. Note that, because the
        filter predicate is tested, ``allowed`` must list all objects in the
        database that test true for the permission.
        """
        if not isinstance(users, AbcCollection):
            users = (users,)
        if not isinstance(allowed, AbcCollection):
            allowed = (allowed,)
        if not isinstance(denied, AbcCollection):
            denied = (denied,)

        errors: list[str] = []

        # Ensure permission is not granted when using worker activation tokens
        worker_activation_token = Token(
            token_type=Token.Types.WORKER_ACTIVATION, enabled=True
        )
        worker_activation_pc = self._assertPermission_context(
            None, worker_activation_token
        )
        for obj in itertools.chain(allowed, denied):
            errors += self._test_check_predicate(
                worker_activation_pc, obj, name, False
            )

        # Ensure permission is not granted when using worker tokens
        worker_token = Token(token_type=Token.Types.WORKER, enabled=True)
        worker_pc = self._assertPermission_context(None, worker_token)
        for obj in itertools.chain(allowed, denied):
            errors += self._test_check_predicate(worker_pc, obj, name, False)

        # We cannot use subTest because the reported stack trace would stop at
        # subTest, and we would lose information on where assertPermission is
        # called
        for user in users:
            pc = self._assertPermission_context(user, token)
            for obj in allowed:
                errors += self._test_check_predicate(pc, obj, name, True)

            for obj in denied:
                errors += self._test_check_predicate(pc, obj, name, False)

            # Pick the model class from the first element of allow or
            # denied
            model_cls = next(itertools.chain(allowed, denied)).__class__
            queryset = getattr(model_cls, "objects")
            filter_predicate = getattr(queryset, name)

            # Check that there are no allowed resources that weren't in the
            # allowed list
            filtered = sorted(str(obj) for obj in filter_predicate(pc))
            expected = sorted(str(obj) for obj in allowed)
            self._check_predicate_filter_output(
                f"{name} for {user} selects ", filtered, expected, errors
            )

            # Check that there are no allowed resources that weren't in the
            # allowed list
            filtered = sorted(
                str(obj)
                for obj in queryset.exclude(pk__in=filter_predicate(pc))
            )
            expected = sorted(str(obj) for obj in denied)
            self._check_predicate_filter_output(
                f"{name} for {user} denies ", filtered, expected, errors
            )

        if errors:
            with io.StringIO() as buf:
                print("Predicate permission mismatch:", file=buf)
                for error in errors:
                    print(f"* {error}", file=buf)
                self.fail(buf.getvalue())

    @contextlib.contextmanager
    def assert_model_count_unchanged(
        self, model_cls: type[Model]
    ) -> Generator[None]:
        """Context manager to assert count of rows in a model is unchanged."""
        initial_count = getattr(model_cls, "objects").count()
        try:
            yield
        finally:
            self.assertEqual(
                getattr(model_cls, "objects").count(), initial_count
            )

    @contextlib.contextmanager
    def template_dir(self) -> Generator[Path]:
        """Prepend a temporary directory to the template search paths."""
        template_dir = self.create_temporary_directory()
        template_settings = copy.deepcopy(settings.TEMPLATES)
        cast(list[str], template_settings[0]["DIRS"]).insert(
            0, template_dir.as_posix()
        )
        with django.test.override_settings(TEMPLATES=template_settings):
            yield template_dir

    def assert_collection_has_constraints(
        self,
        collection: Collection,
        expected_constraints: list[dict[str, Any]],
    ) -> None:
        """Assert that a collection has these constraints."""
        constraints = [
            dict(constraint)
            for constraint in CollectionItemMatchConstraint.objects.filter(
                collection=collection
            ).values()
        ]
        for constraint in constraints:
            del constraint["id"]
        self.assertEqual(
            sorted(constraints, key=itemgetter("collection_item_id", "key")),
            sorted(
                [
                    {"collection_id": collection.id, **constraint}
                    for constraint in expected_constraints
                ],
                key=itemgetter("collection_item_id", "key"),
            ),
        )

    def close_streaming_response(self, response: HttpResponseBase) -> None:
        """Close a streaming response without reading the rest of its data."""
        # See https://code.djangoproject.com/ticket/35618
        #
        # django.test.client.ClientHandler does this when closing a response in
        # the non-streaming case, but I cannot find a standard way to close a
        # streaming response without invalidating the database connection held
        # by the tests, which will cause further tests to fail.
        #
        # This is a copy of what django.test.client.ClientHandler does in the
        # non-streaming case
        request_finished.disconnect(close_old_connections)
        response.close()  # will fire request_finished
        request_finished.connect(close_old_connections)

    def render_string(
        self,
        template_code: str,
        request: HttpRequest,
        **context: Any,
    ) -> str:
        """Render a template from a string."""
        template = engines["django"].from_string(template_code)
        return template.render(context, request=request)

    @contextlib.contextmanager
    def freeze_time(self) -> Iterator[dt.datetime]:
        """Freeze django.utils.timezone.now() timestamps."""
        frozen_time = timezone.now()
        with mock.patch("django.utils.timezone.now", return_value=frozen_time):
            yield frozen_time


class PlaygroundTestCase(BaseDjangoTestCase):
    """
    Playground-specific test case setup.

    Class annotations are scanned in test case setup for Scenario instances,
    which are automatically instantiated and set in the class.

    Scenario instances can be annotated with a dict, which will provide extra
    arguments passed to playground.scenario(). This can be used, for example,
    to call ``set_current``::

      scenario: Annotated[ClassVar[DefaultContext], {"set_current": True}]
    """

    #: set to False if you do not need a playground in your test case
    playground_needed: bool = True

    #: set to False if you need a playground with a LocalFileStore instead of
    #: an in-memory version
    playground_memory_file_store: bool = True

    @classmethod
    def create_playground(cls) -> Playground:
        """
        Create the playground object.

        This method can be overridden by subclasses to create a playground with
        a different configuration
        """
        return Playground(memory_file_store=cls.playground_memory_file_store)

    @classmethod
    def list_playground_scenarios(
        cls,
    ) -> Generator[tuple[str, Scenario]]:
        """
        List the scenarios configured in the test case.

        Generate tuples like ``(name, scenario)``.
        """
        for name, value in inspect.getmembers(cls):
            if not isinstance(value, Scenario):
                continue
            yield name, value

    def _build_permission_when_role_testlist(
        self,
        predicate: Callable[[PermissionContext], bool],
        roles: str | AbcCollection[str] = (),
        *,
        scope_roles: str | AbcCollection[str] = (),
        workspace_roles: str | AbcCollection[str] = (),
    ) -> list[tuple[Model, str]]:
        """Build a TODO list for testing the predicate with the given roles."""
        assert isinstance(predicate, PermissionCheckPredicate)
        resource = predicate.__self__
        roles = resolve_roles_list(resource, roles)
        scope_roles = resolve_roles_list(Scope, scope_roles)
        workspace_roles = resolve_roles_list(Workspace, workspace_roles)

        tests: list[tuple[Model, str]] = []
        if scope_roles:
            scope = get_resource_scope(resource)
            assert scope is not None
            tests.extend((scope, role) for role in scope_roles)
        if workspace_roles:
            workspace = get_resource_workspace(resource)
            assert workspace is not None
            tests.extend((workspace, role) for role in workspace_roles)
        tests.extend((resource, role) for role in roles)

        return tests

    def assertNoPermissionWhenRole(  # noqa: C901
        self,
        predicate: Callable[[PermissionContext], bool],
        pc: PermissionContext,
        roles: str | AbcCollection[str] = (),
        *,
        scope_roles: str | AbcCollection[str] = (),
        workspace_roles: str | AbcCollection[str] = (),
    ) -> None:
        """
        Check that assigning roles does not make the predicate come true.

        Each role is tested in isolation: the test is repeated for each given
        role.
        """
        assert isinstance(predicate, PermissionCheckPredicate)
        errors: list[str] = []
        resource: Model = predicate.__self__
        pred_name = f"{predicate.__func__.__name__!r} on {str(resource)!r}"
        filter_predicate = getattr(
            getattr(resource.__class__, "objects"), predicate.__func__.__name__
        )
        for assignee, role in self._build_permission_when_role_testlist(
            predicate,
            roles,
            scope_roles=scope_roles,
            workspace_roles=workspace_roles,
        ):
            with context.local(), self.ephemeral_savepoint():
                context.reset()

                if predicate(pc):
                    errors.append(f"{pred_name} is true even without {role}")
                if filter_predicate(pc).filter(pk=resource.pk).exists():
                    errors.append(
                        f"{pred_name} passes filtering even without {role}"
                    )

                # Ignoring type until we find a way to type self.playground to
                # accept either a member or a classvar, depending on subclasses
                self.playground.create_group_role(  # type: ignore[attr-defined]
                    assignee,
                    role,
                    users=[pc.user],
                    name="assertNoPermissionWhenRole",
                )

                if predicate(pc):
                    errors.append(f"{pred_name} is true when given {role}")
                if filter_predicate(pc).filter(pk=resource.pk).exists():
                    errors.append(
                        f"{pred_name} passes filtering when given {role}"
                    )

                # Once more with context set, to test shortcutting
                set_context_for_resource(assignee, pc.user)

                if predicate(pc):
                    errors.append(
                        f"{pred_name} is true when given {role} in context"
                    )
                if filter_predicate(pc).filter(pk=resource.pk).exists():
                    errors.append(
                        f"{pred_name} passes filtering when given {role}"
                        " in context"
                    )

        self.assertEqual(errors, [])

    def assertPermissionWhenRole(  # noqa: C901
        self,
        predicate: Callable[[PermissionContext], bool],
        pc: PermissionContext,
        roles: str | AbcCollection[str] = (),
        scope_roles: str | AbcCollection[str] = (),
        workspace_roles: str | AbcCollection[str] = (),
    ) -> None:
        """
        Check that assigning roles makes the predicate come true.

        Each role is tested in isolation: the test is repeated for each given
        role.
        """
        assert isinstance(predicate, PermissionCheckPredicate)
        errors: list[str] = []
        resource = predicate.__self__
        pred_name = f"{predicate.__func__.__name__!r} on {str(resource)!r}"
        filter_predicate = getattr(
            resource.__class__.objects, predicate.__func__.__name__
        )
        for assignee, role in self._build_permission_when_role_testlist(
            predicate,
            roles,
            scope_roles=scope_roles,
            workspace_roles=workspace_roles,
        ):
            with context.local(), self.ephemeral_savepoint():
                context.reset()

                if predicate(pc):
                    errors.append(f"{pred_name} is true even without {role}")
                if filter_predicate(pc).filter(pk=resource.pk).exists():
                    errors.append(
                        f"{pred_name} passes filtering even without {role}"
                    )

                # Ignoring type until we find a way to type self.playground to
                # accept either a member or a classvar, depending on subclasses
                self.playground.create_group_role(  # type: ignore[attr-defined]
                    assignee,
                    role,
                    users=[pc.user],
                    name="assertPermissionWhenRole",
                )

                if not predicate(pc):
                    errors.append(f"{pred_name} is false even with {role}")
                if not filter_predicate(pc).filter(pk=resource.pk).exists():
                    errors.append(
                        f"{pred_name} does not pass filtering even with {role}"
                    )

                # Once more with context set, to test shortcutting
                set_context_for_resource(assignee, pc.user)

                if not predicate(pc):
                    errors.append(
                        f"{pred_name} is false even with {role} in context"
                    )
                if not filter_predicate(pc).filter(pk=resource.pk).exists():
                    errors.append(
                        f"{pred_name} does not pass filtering even with {role}"
                        " in context"
                    )

        self.assertEqual(errors, [])


class TransactionTestCase(
    PlaygroundTestCase,
    django.test.TransactionTestCase,
):
    """Debusine-specific extensions to django's TransactionTestCase."""

    playground: Playground

    @override
    def setUp(self) -> None:
        """Create a playground for TransactionTestCase."""
        super().setUp()
        if self.playground_needed:
            # TransactionTestCase does not support setUpTestData, so we need a
            # playground instance per test method
            self.playground = self.enterContext(self.create_playground())
            # Instantiate playground scenarios
            for name, scenario in self.__class__.list_playground_scenarios():
                # Make a copy of the blank scenario in the class
                method_scenario = copy.deepcopy(scenario)
                # Instantiate a different one for each test method
                self.playground.build_scenario(
                    method_scenario, scenario_name=name, set_current=True
                )
                setattr(self, name, method_scenario)

    def assert_work_requests_locked(
        self, *, locked: Sequence[WorkRequest], unlocked: Sequence[WorkRequest]
    ) -> None:
        """Check that the given work requests are locked/unlocked for update."""
        with self.separate_connection(alias := "test-locks"):
            qs: QuerySet[WorkRequest] = WorkRequestQuerySet(
                WorkRequest, using=alias
            )
            for wr in locked:
                with (
                    transaction.atomic(using=alias),
                    self.assertRaisesRegex(
                        OperationalError, "could not obtain lock"
                    ),
                ):
                    # .exists() is enough here, since the query set contains
                    # at most one row.
                    qs.select_for_update(nowait=True, no_key=True).filter(
                        id=wr.id
                    ).exists()
            for wr in unlocked:
                with transaction.atomic(using=alias):
                    # .exists() is enough here, since the query set contains
                    # at most one row.
                    qs.select_for_update(nowait=True, no_key=True).filter(
                        id=wr.id
                    ).exists()


class TestCase(
    PlaygroundTestCase,
    django.test.TestCase,
):
    """Debusine-specific extensions to django's TestCase."""

    playground: ClassVar[Playground]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        """Create a playground to setup test data."""
        super().setUpTestData()
        if cls.playground_needed:
            cls.playground = cls.enterClassContext(cls.create_playground())
            for name, scenario in cls.list_playground_scenarios():
                # set_current is delayed to the setUp method
                cls.playground.build_scenario(
                    scenario, scenario_name=name, set_current=False
                )

    @override
    def setUp(self) -> None:
        """Create a playground for TransactionTestCase."""
        super().setUp()
        if self.playground_needed:
            for name, scenario in self.playground.scenarios.items():
                # Replace the class version of the scenario with its deepcopied
                # version
                setattr(self, name, scenario)
                # BaseDjangoTestCase.setUp calls context.reset, so we need to
                # call set_current here instead of in setUpTestData
                if scenario.needs_set_current:
                    scenario.set_current()

    def assert_work_request_event_reactions(
        self,
        work_request: WorkRequest,
        *,
        on_creation: Sequence[BaseTaskDataModel] | None = None,
        on_unblock: Sequence[BaseTaskDataModel] | None = None,
        on_assignment: Sequence[BaseTaskDataModel] | None = None,
        on_success: Sequence[BaseTaskDataModel] | None = None,
        on_failure: Sequence[BaseTaskDataModel] | None = None,
    ) -> None:
        """
        Assert that a work request has the given event reactions.

        The order of the items in each list of event reactions does not
        matter.  Leaving parameters unset or None is equivalent to passing
        the empty list, meaning that they are required to be empty.
        """
        self.assertCountEqual(
            work_request.event_reactions.on_creation or [],
            on_creation or [],
        )
        self.assertCountEqual(
            work_request.event_reactions.on_unblock or [],
            on_unblock or [],
        )
        self.assertCountEqual(
            work_request.event_reactions.on_assignment or [],
            on_assignment or [],
        )
        self.assertCountEqual(
            work_request.event_reactions.on_success or [],
            on_success or [],
        )
        self.assertCountEqual(
            work_request.event_reactions.on_failure or [],
            on_failure or [],
        )


class ChannelsHelpersMixin(TestMixinBase):
    """
    Channel-related methods to help writing unit tests.

    Provides methods to setup a channel and assert messages or lack of messages.
    """

    @override
    def tearDown(self) -> None:
        """Flush the channel layer."""
        super().tearDown()
        channel_layer = get_channel_layer()
        assert channel_layer is not None
        async_to_sync(channel_layer.flush)()

    async def create_channel(self, group_name: str) -> dict[str, Any]:
        """
        Create a channel and add it to the group named ``group_name``.

        Return dict with layer and name.
        """
        channel_layer = get_channel_layer()
        assert channel_layer is not None
        channel_name = await channel_layer.new_channel()
        await channel_layer.group_add(group_name, channel_name)

        return {"layer": channel_layer, "name": channel_name}

    async def assert_channel_nothing_received(
        self, channel: dict[str, Any]
    ) -> None:
        """Assert that nothing is received in channel."""
        try:
            received = await asyncio.wait_for(
                channel["layer"].receive(channel["name"]), timeout=0.1
            )
        except asyncio.exceptions.TimeoutError:
            pass
        else:
            cast(unittest.TestCase, self).fail(
                f"Expected nothing. Received: '{safe_repr(received)}'"
            )

    async def assert_channel_received(
        self, channel: dict[str, Any], data: dict[str, Any]
    ) -> None:
        """Assert that data is received in channel_layer, channel_name."""
        try:
            received = await asyncio.wait_for(
                channel["layer"].receive(channel["name"]), timeout=0.1
            )
            cast(unittest.TestCase, self).assertEqual(received, data)
        except asyncio.exceptions.TimeoutError:
            cast(unittest.TestCase, self).fail(
                f"Expected '{safe_repr(data)}' received nothing"
            )
