# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Unit tests for model permissions."""

import re
from collections.abc import Collection, Generator
from typing import Any, ClassVar, Protocol, Self, override

from django.core.exceptions import PermissionDenied
from django.db import models

from debusine.db.context import context
from debusine.db.models import Group, Scope, Token, User, Workspace
from debusine.db.models.permissions import (
    Allow,
    enforce,
    format_permission_check_error,
    get_resource_scope,
    get_resource_workspace,
    permission_check,
    permission_filter,
    resolve_role,
    resolve_roles_list,
)
from debusine.db.permissioncontext import PermissionContext, PermissionUser
from debusine.test.django import (
    AllowAll,
    DenyAll,
    TestCase,
    override_permission,
)


class PredicateWrapper(Protocol):
    """Structural typing for check/filter predicate testers."""

    last_user: PermissionUser | None
    called: bool

    def test(self, pc: PermissionContext) -> bool: ...


class PredicateTester:
    """Encapsulate tests for check and filter predicates."""

    def __init__(
        self,
        test_case: TestCase,
        pc: PermissionContext,
        tester: PredicateWrapper,
    ) -> None:
        """Test calling tester with user."""
        self.test_case = test_case
        self.pc = pc
        self.tester = tester

    @override
    def __repr__(self) -> str:
        """Representation for debugging, shown by subTest."""
        return f"{self.pc=!r}, {self.tester=!r}"

    def test(self) -> bool:
        """Invoke the wrapper's test method."""
        return self.tester.test(self.pc)

    def assert_forced_false(self) -> None:
        """The predicate failed without calling its body."""
        self.test_case.assertFalse(self.test())
        self.test_case.assertFalse(self.tester.called)

    def assert_called(self) -> None:
        """The predicate body has been called with the user passed to it."""
        self.test_case.assertFalse(self.test())
        self.test_case.assertTrue(self.tester.called)
        self.test_case.assertEqual(self.tester.last_user, self.pc.user)

    def assert_passes_with_checks_disabled(self) -> None:
        """The predicate always succeeds with permission checks disabled."""
        orig = self.pc
        try:
            self.pc = self.pc.with_permission_checks_disabled()
            self.test_case.assertTrue(self.test())
            self.test_case.assertFalse(self.tester.called)
        finally:
            self.pc = orig


def mock_pred_check(
    test_case: TestCase,
    *,
    pc: PermissionContext,
    work_request: Allow,
    anonymous: Allow,
) -> PredicateTester:
    """Create a tester for check predicates."""

    class CheckContainer:
        """Mock resource used to test the permission_check decorator."""

        def __init__(self) -> None:
            """Make space to store a trace of checks."""
            self.last_user: PermissionUser | None = None
            self.called = False

        @override
        def __repr__(self) -> str:
            return f"permission_check({work_request=!r}, {anonymous=!r}, {pc=})"

        # permission_check wants a Model, but it's easier to have a mock class
        # that is not a model to avoid having to deal with Django's model
        # registry. Override the check instead
        @permission_check(  # type: ignore[type-var]
            "{user} cannot pred {resource}",
            work_request=work_request,
            anonymous=anonymous,
        )
        def _predicate(self, pc: PermissionContext) -> bool:
            self.last_user = pc.user
            self.called = True
            return False

        def test(self, pc: PermissionContext) -> bool:
            self.called = False
            return self._predicate(pc)

    return PredicateTester(test_case, pc, CheckContainer())


def mock_pred_filter(
    test_case: TestCase,
    *,
    pc: PermissionContext,
    work_request: Allow,
    anonymous: Allow,
) -> PredicateTester:
    """Create a tester for filter predicates."""

    class FilterContainer(models.QuerySet[User]):
        """Mock resource used to test the permission_filter decorator."""

        def __init__(self, *args: Any, **kwargs: Any) -> None:
            """Make space to store a trace of checks."""
            super().__init__(*args, **kwargs)
            self.last_user: PermissionUser | None = None
            self.called = False

        @override
        def __repr__(self) -> str:
            return (
                f"permission_filter({work_request=!r}, {anonymous=!r}, {pc=})"
            )

        @permission_filter(work_request=work_request, anonymous=anonymous)
        def _predicate(self, pc: PermissionContext) -> Self:
            self.last_user = pc.user
            self.called = True
            return self.none()

        def test(self, pc: PermissionContext) -> bool:
            self.called = False
            return self._predicate(pc).count() > 0

    return PredicateTester(test_case, pc, FilterContainer(User))


class PermissionsTests(TestCase):
    """Tests for permission infrastructure."""

    scope: ClassVar[Scope]
    scope1: ClassVar[Scope]
    user: ClassVar[User]
    user1: ClassVar[User]
    group: ClassVar[Group]
    work_request_token: ClassVar[Token]
    work_request_pc: ClassVar[PermissionContext]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.scope = cls.playground.get_default_scope()
        cls.scope1 = cls.playground.get_or_create_scope("scope")
        cls.user = cls.playground.get_default_user()
        cls.user1 = cls.playground.create_user("user1")
        cls.group = cls.playground.create_group_role(
            cls.scope, Scope.Roles.OWNER
        )
        cls.work_request_token = cls.playground.create_work_request_token()
        cls.work_request_pc = PermissionContext.for_work_request(
            token=cls.work_request_token
        )

    def test_resolve_role(self) -> None:
        """Test passing a valid role type."""
        for val, result in (
            ("owner", Workspace.Roles.OWNER),
            (Workspace.Roles.OWNER, Workspace.Roles.OWNER),
        ):
            with self.subTest(val=repr(val)):
                self.assertEqual(resolve_role(Workspace, val), result)

    def test_resolve_role_invalid_type(self) -> None:
        """Test passing an invalid role type."""
        with self.assertRaisesRegex(
            ValueError, r"42 is not a valid ScopeRoles"
        ):
            resolve_role(Scope, 42)  # type: ignore[arg-type]

    def test_resolve_role_invalid_enum(self) -> None:
        """Test passing an invalid role enum."""
        with self.assertRaisesRegex(
            TypeError,
            re.escape(
                f"{Workspace.Roles.OWNER!r} cannot be converted to"
                " <enum 'ScopeRoles'>"
            ),
        ):
            resolve_role(Scope, Workspace.Roles.OWNER)

    def test_resolve_role_invalid_name(self) -> None:
        """Test passing an invalid role name."""
        with self.assertRaisesRegex(
            ValueError, r"'does_not_exist' is not a valid ScopeRoles"
        ):
            resolve_role(Scope, "does_not_exist")

    def test_resolve_roles_list(self) -> None:
        """Test resolve_roles_list function."""
        for roles_cls, arg, expected in (
            (Scope, "owner", [Scope.Roles.OWNER]),
            (Scope, Scope.Roles.OWNER, [Scope.Roles.OWNER]),
            (
                Workspace,
                ["owner", Workspace.Roles.OWNER],
                [Workspace.Roles.OWNER, Workspace.Roles.OWNER],
            ),
        ):
            with self.subTest(roles_cls=repr(roles_cls), arg=repr(arg)):
                self.assertEqual(resolve_roles_list(roles_cls, arg), expected)

    def test_permission_error_template(self) -> None:
        """Test accessing the permission error template."""
        res = mock_pred_check(
            self,
            pc=PermissionContext.for_ui_anonymous(),
            work_request=Allow.PASS,
            anonymous=Allow.PASS,
        )
        resource = res.tester
        resource_class = resource.__class__
        cls_predicate = resource_class._predicate  # type: ignore[attr-defined]
        self.assertEqual(
            cls_predicate.error_template,
            "{user} cannot pred {resource}",
        )

        self.assertEqual(
            resource._predicate.error_template,  # type: ignore[attr-defined]
            "{user} cannot pred {resource}",
        )

    def test_format_permission_check_error(self) -> None:
        """Test format_permission_check_error."""
        self.assertEqual(
            format_permission_check_error(
                self.scope.can_display, PermissionContext.for_ui_user(self.user)
            ),
            "playground cannot display scope debusine",
        )

    def test_enforce(self) -> None:
        """Test the enforce method."""
        context.set_permission_context_scoped(
            self.scope, PermissionContext.for_ui_anonymous()
        )

        with override_permission(Scope, "can_display", AllowAll):
            enforce(self.scope.can_display)
        with (
            override_permission(Scope, "can_display", DenyAll),
            self.assertRaisesRegex(
                PermissionDenied, "AnonymousUser cannot display scope debusine"
            ),
        ):
            enforce(self.scope.can_display)

    def predicate_testers(
        self,
        work_request: Collection[Allow],
        anonymous: Collection[Allow],
        pcs: Collection[PermissionContext],
    ) -> Generator[PredicateTester]:
        for w in work_request:
            for a in anonymous:
                for pc in pcs:
                    yield mock_pred_check(
                        self, work_request=w, anonymous=a, pc=pc
                    )
                    yield mock_pred_filter(
                        self, work_request=w, anonymous=a, pc=pc
                    )

    def test_predicates_worker_activation(self) -> None:
        token = self.playground.create_worker_activation_token()
        for tester in self.predicate_testers(
            work_request=(Allow.PASS, Allow.NEVER),
            anonymous=(Allow.PASS, Allow.NEVER),
            pcs=[PermissionContext.for_worker_activation(token)],
        ):
            with self.subTest(tester=repr(tester)):
                tester.assert_forced_false()
                tester.assert_passes_with_checks_disabled()

    def test_predicates_worker_token(self) -> None:
        token = self.playground.create_worker_token()
        for tester in self.predicate_testers(
            work_request=(Allow.PASS, Allow.NEVER),
            anonymous=(Allow.PASS, Allow.NEVER),
            pcs=[PermissionContext.for_worker(token)],
        ):
            with self.subTest(tester=repr(tester)):
                tester.assert_forced_false()
                tester.assert_passes_with_checks_disabled()

    def test_predicates_worker_pass(self) -> None:
        """Test predicates with worker, work_request=PASS."""
        for tester in self.predicate_testers(
            work_request=[Allow.PASS],
            anonymous=(Allow.PASS, Allow.NEVER),
            pcs=[self.work_request_pc],
        ):
            with self.subTest(tester=repr(tester)):
                tester.assert_called()
                tester.assert_passes_with_checks_disabled()

    def test_predicates_worker_never(self) -> None:
        """Test predicates with worker, work_request=NEVER."""
        for tester in self.predicate_testers(
            work_request=[Allow.NEVER],
            anonymous=(Allow.PASS, Allow.NEVER),
            pcs=[self.work_request_pc],
        ):
            with self.subTest(tester=repr(tester)):
                tester.assert_forced_false()
                tester.assert_passes_with_checks_disabled()

    def test_predicates_anonymous_never(self) -> None:
        """Test predicates with anonymous user, anonymous=NEVER."""
        for tester in self.predicate_testers(
            work_request=(Allow.PASS, Allow.NEVER),
            anonymous=[Allow.NEVER],
            pcs=[PermissionContext.for_ui_anonymous()],
        ):
            with self.subTest(tester=repr(tester)):
                tester.assert_forced_false()
                tester.assert_passes_with_checks_disabled()

    def test_predicates_anonymous_pass(self) -> None:
        """Test predicates with anonymous user, anonymous=PASS."""
        for tester in self.predicate_testers(
            work_request=[Allow.PASS, Allow.NEVER],
            anonymous=[Allow.PASS],
            pcs=[PermissionContext.for_ui_anonymous()],
        ):
            with self.subTest(tester=repr(tester)):
                tester.assert_called()
                tester.assert_passes_with_checks_disabled()

    def test_predicates_user(self) -> None:
        """Test predicates with explicitly given user."""
        for tester in self.predicate_testers(
            work_request=(Allow.PASS, Allow.NEVER),
            anonymous=(Allow.PASS, Allow.NEVER),
            pcs=[PermissionContext.for_ui_user(self.user)],
        ):
            with self.subTest(tester=repr(tester)):
                tester.assert_called()
                tester.assert_passes_with_checks_disabled()

    def test_get_resource_scope(self) -> None:
        """Test get_resource_scope."""
        workspace = self.playground.get_default_workspace()
        artifact, _ = self.playground.create_artifact(create_files=False)

        self.assertIsNone(get_resource_scope(self.user))
        self.assertEqual(get_resource_scope(self.group), self.scope)

        self.assertEqual(get_resource_scope(self.scope), self.scope)
        self.assertEqual(get_resource_scope(self.scope1), self.scope1)

        self.assertEqual(get_resource_scope(workspace), self.scope)
        self.assertEqual(get_resource_scope(artifact), self.scope)

    def test_get_resource_workspace(self) -> None:
        """Test get_resource_workspace."""
        workspace = self.playground.get_default_workspace()
        artifact, _ = self.playground.create_artifact(create_files=False)

        self.assertIsNone(get_resource_workspace(self.user))
        self.assertIsNone(get_resource_workspace(self.group))
        self.assertIsNone(get_resource_workspace(self.scope))

        self.assertEqual(get_resource_workspace(workspace), workspace)
        self.assertEqual(get_resource_workspace(artifact), workspace)
