# 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 application context."""

import asyncio
from collections.abc import Callable
from threading import Thread
from typing import Any, ClassVar, NamedTuple, Self, override

from django.db import connection, transaction

from debusine.client.models import LookupChildType
from debusine.db.context import ContextConsistencyError, context
from debusine.db.models import Scope, Token, Workspace
from debusine.db.models.auth import system_user
from debusine.db.models.lookups import (
    LookupCache,
    LookupCacheGroup,
    LookupResultBare,
)
from debusine.db.models.scopes import ScopeRoles
from debusine.db.models.workspaces import WorkspaceRoles
from debusine.db.permissioncontext import PermissionContext
from debusine.db.playground import scenarios
from debusine.test.django import (
    PlaygroundTestCase,
    TestCase,
    TransactionTestCase,
)


class ContextSnapshot(NamedTuple):
    """Comparable snapshot of the contents of a context."""

    scope: Scope | None
    pc: PermissionContext | None
    scope_roles: frozenset[ScopeRoles] | None
    workspace: Workspace | None
    workspace_roles: frozenset[WorkspaceRoles] | None
    lookup_cache_sizes: tuple[int, int] | None

    @classmethod
    def take(cls) -> Self:
        """Take a snapshot of the contents of a context."""
        lookup_cache = context._lookup_cache.get()
        return cls(
            scope=context._scope.get(),
            pc=context._permission_context.get(),
            scope_roles=context._scope_roles.get(),
            workspace=context._workspace.get(),
            workspace_roles=context._workspace_roles.get(),
            lookup_cache_sizes=(
                (
                    len(lookup_cache.single),
                    len(lookup_cache.dictionary),
                )
                if lookup_cache
                else None
            ),
        )


class TestContextBase(PlaygroundTestCase):
    """Common base for context tests."""

    scenario = scenarios.DefaultContext()

    @override
    def setUp(self) -> None:
        super().setUp()
        self.lookup_cache_single_key = (
            1,
            self.scenario.workspace,
            self.scenario.user,
            None,
            None,
            LookupChildType.ANY,
        )
        self.lookup_cache_single_value = LookupResultBare(lookup="")

    def assert_context_initial(self) -> None:
        """Check that the context has initial values."""
        with self.assertRaisesRegex(
            ContextConsistencyError, r"permission context has not yet been set"
        ):
            context.user_token
        self.assertIsNone(context.scope)
        self.assertIsNone(context._scope_roles.get())
        self.assertIsNone(context.workspace)
        self.assertIsNone(context._workspace_roles.get())
        self.assertIsNone(context._permission_context.get())
        self.assertFalse(context.permission_checks_disabled)
        self.assertEqual(context.lookup_cache, LookupCache())
        with self.assertRaisesRegex(
            ContextConsistencyError, r"scope is not set in context"
        ):
            context.require_scope()
        with self.assertRaisesRegex(
            ContextConsistencyError, r"permission context has not yet been set"
        ):
            context.user
        with self.assertRaisesRegex(
            ContextConsistencyError, r"workspace is not set in context"
        ):
            context.require_workspace()

    def assert_context_all_set(self) -> None:
        """Check that the context has been set to default test values."""
        self.assertEqual(
            context._permission_context.get(), self.scenario.context_ui()
        )
        self.assertEqual(context.user, self.scenario.user)
        self.assertEqual(context.scope, self.scenario.scope)
        self.assertEqual(context.require_scope(), self.scenario.scope)
        self.assertIsNotNone(context._scope_roles.get())
        self.assertEqual(context.workspace, self.scenario.workspace)
        self.assertIsNotNone(context._workspace_roles.get())
        self.assertEqual(
            context.require_workspace(),
            self.scenario.workspace,
        )
        self.assertEqual(
            context.lookup_cache,
            LookupCache(
                single=LookupCacheGroup(
                    {
                        self.lookup_cache_single_key: (
                            self.lookup_cache_single_value
                        )
                    }
                )
            ),
        )


class TestAppContext(TestContextBase, TestCase):
    """Test application context variables."""

    user_token: ClassVar[Token]
    worker_token: ClassVar[Token]

    @override
    @classmethod
    def setUpTestData(cls) -> None:
        super().setUpTestData()
        cls.user_token = cls.playground.create_user_token()
        cls.worker_token = cls.playground.create_worker_token()

    @override
    def setUp(self) -> None:
        """Ensure context is not left dirty by a bug in a previous test."""
        super().setUp()
        context.reset()
        self.assertIsNone(context.scope)
        self.assertIsNone(context.workspace)
        self.assertIsNone(context._permission_context.get())

    def run_in_task(
        self, func: Callable[..., Any], *args: Any, **kwargs: Any
    ) -> Any:
        """Run a callable in an asyncio task."""

        async def _task_main() -> Any:
            return func(*args, **kwargs)

        async def _async_main() -> asyncio.Task[Any]:
            return await asyncio.create_task(_task_main())

        return asyncio.run(_async_main())

    def test_defaults(self) -> None:
        """Context variables are None by default."""
        self.assert_context_initial()

    def test_local_previously_none(self) -> None:
        """Using local() restores previously unset vars."""
        with context.local():
            self.scenario.set_current()
            context.lookup_cache.single[self.lookup_cache_single_key] = (
                self.lookup_cache_single_value
            )
            self.assert_context_all_set()

        self.assert_context_initial()

    def test_local_previously_set(self) -> None:
        """Using local() restores previously set vars."""
        try:
            self.scenario.set_current()
            context.lookup_cache.single[self.lookup_cache_single_key] = (
                self.lookup_cache_single_value
            )

            with context.local():
                context.reset()
                self.assert_context_initial()

            self.assert_context_all_set()
        finally:
            context.reset()

    def test_reset(self) -> None:
        """Calling reset() sets context to its initial values."""
        self.scenario.set_current()
        context.lookup_cache.single[self.lookup_cache_single_key] = (
            self.lookup_cache_single_value
        )
        self.assert_context_all_set()

        context.reset()
        self.assert_context_initial()

    def test_visibility_task(self) -> None:
        """Check visibility with asyncio tasks."""
        self.scenario.set_current()
        context.lookup_cache.single[self.lookup_cache_single_key] = (
            self.lookup_cache_single_value
        )

        def _test() -> None:
            # Application context is preserved when entering a Task
            self.assert_context_all_set()
            context.reset()
            self.assert_context_initial()

        self.run_in_task(_test)

        # Changes made in another Task are isolated
        self.assert_context_all_set()

    def test_set_permission_context_scoped(self) -> None:
        self.playground.create_group_role(
            self.scenario.scope, Scope.Roles.OWNER, users=[self.scenario.user]
        )

        self.assert_context_initial()
        context.set_permission_context_scoped(
            self.scenario.scope, self.scenario.context_ui()
        )
        self.assertEqual(context.scope, self.scenario.scope)
        self.assertEqual(context.pc, self.scenario.context_ui())
        self.assertIsNone(context.workspace)
        self.assertIsNone(context._workspace_roles.get())
        self.assertFalse(context.permission_checks_disabled)

        self.assertEqual(context.scope_roles, frozenset((Scope.Roles.OWNER,)))

    def test_set_permission_context_scoped_extra_groups(self) -> None:
        pc = self.playground.create_pc_with_extra_groups(
            self.scenario.scope_owners
        )
        self.assert_context_initial()
        context.set_permission_context_scoped(self.scenario.scope, pc)
        self.assertEqual(context.scope, self.scenario.scope)
        self.assertEqual(context.pc, pc)
        self.assertIsNone(context.workspace)
        self.assertIsNone(context._workspace_roles.get())
        self.assertFalse(context.permission_checks_disabled)

        self.assertEqual(context.scope_roles, frozenset((Scope.Roles.OWNER,)))

    def test_set_permission_context_scoped_twice(self) -> None:
        context.set_permission_context_scoped(
            self.scenario.scope, self.scenario.context_ui()
        )
        expected = ContextSnapshot.take()

        scope1 = self.playground.get_or_create_scope(name="scope1")
        pc1 = PermissionContext.for_ui_anonymous()
        with self.assertRaisesRegex(
            ContextConsistencyError, "Scope was already set to debusine"
        ):
            context.set_permission_context_scoped(scope1, pc1)

        self.assertEqual(ContextSnapshot.take(), expected)

    def test_set_permission_context_scopeless(self) -> None:
        self.assert_context_initial()
        context.set_permission_context_scopeless(self.scenario.context_ui())
        self.assertIsNone(context.scope)
        self.assertEqual(context.pc, self.scenario.context_ui())
        with self.assertRaisesRegex(
            ContextConsistencyError, "scope has not been set"
        ):
            self.assertIsNone(context.scope_roles)
        self.assertIsNone(context.workspace)
        self.assertIsNone(context._workspace_roles.get())
        self.assertFalse(context.permission_checks_disabled)

    def test_set_permission_context_scopeless_twice(self) -> None:
        context.set_permission_context_scopeless(self.scenario.context_ui())
        expected = ContextSnapshot.take()

        pc1 = PermissionContext.for_ui_anonymous()
        with self.assertRaisesRegex(
            ContextConsistencyError, "Permission context was already set to "
        ):
            context.set_permission_context_scopeless(pc1)

        self.assertEqual(ContextSnapshot.take(), expected)

    def test_set_permission_context_scopeless_then_scoped(self) -> None:
        context.set_permission_context_scopeless(self.scenario.context_ui())
        expected = ContextSnapshot.take()

        with self.assertRaisesRegex(
            ContextConsistencyError, "Permission context was already set to "
        ):
            context.set_permission_context_scoped(
                self.scenario.scope, self.scenario.context_ui()
            )

        self.assertEqual(ContextSnapshot.take(), expected)

    def test_set_permission_context_scoped_then_scopeless(self) -> None:
        context.set_permission_context_scoped(
            self.scenario.scope, self.scenario.context_ui()
        )
        expected = ContextSnapshot.take()

        with self.assertRaisesRegex(
            ContextConsistencyError,
            f"Scope was already set to {self.scenario.scope}",
        ):
            context.set_permission_context_scopeless(self.scenario.context_ui())

        self.assertEqual(ContextSnapshot.take(), expected)

    def test_scope_roles_unset(self) -> None:
        """Test scope_roles accessor."""
        with self.assertRaisesRegex(
            ContextConsistencyError, "scope has not been set"
        ):
            context.scope_roles

    def test_scope_roles_empty(self) -> None:
        """Test scope_roles accessor with empty roleset."""
        self.scenario.set_current()
        self.assertEqual(context.scope_roles, frozenset())

    def test_scope_roles(self) -> None:
        """Test scope_roles accessor."""
        self.playground.create_group_role(
            self.scenario.scope, Scope.Roles.OWNER, users=[self.scenario.user]
        )
        self.scenario.set_current()
        self.assertEqual(context.scope_roles, frozenset((Scope.Roles.OWNER,)))

    def test_workspace_roles_unset(self) -> None:
        """Test workspace_roles accessor."""
        with self.assertRaisesRegex(
            ContextConsistencyError, "workspace is not set"
        ):
            context.workspace_roles

    def test_workspace_roles_empty(self) -> None:
        """Test workspace_roles accessor with empty roleset."""
        self.scenario.set_current()
        self.assertEqual(
            context.workspace_roles, frozenset((Workspace.Roles.VIEWER,))
        )

    def test_workspace_roles(self) -> None:
        """Test workspace_roles accessor."""
        self.playground.create_group_role(
            self.scenario.workspace,
            Workspace.Roles.OWNER,
            users=[self.scenario.user],
        )

        self.scenario.set_current()
        self.assertEqual(
            context.workspace_roles, frozenset((Scope.Roles.OWNER,))
        )

    def test_management_command(self) -> None:
        self.scenario.set_current()

        with context.management_command():
            self.assertIsNone(context.scope)
            self.assertIsNone(context._scope_roles.get())
            self.assertIsNone(context.workspace)
            self.assertIsNone(context._workspace_roles.get())
            self.assertFalse(context.permission_checks_disabled)
            self.assertEqual(context.lookup_cache, LookupCache())
            self.assertEqual(context.user, system_user())

        self.assertIsNone(context.user_token)
        self.assertEqual(context.user, self.scenario.user)
        self.assertEqual(context.scope, self.scenario.scope)
        self.assertEqual(context.require_scope(), self.scenario.scope)
        self.assertIsNotNone(context._scope_roles.get())
        self.assertEqual(context.workspace, self.scenario.workspace)
        self.assertIsNotNone(context._workspace_roles.get())
        self.assertEqual(
            context.require_workspace(),
            self.scenario.workspace,
        )
        self.assertFalse(context.permission_checks_disabled)
        self.assertEqual(context.lookup_cache, LookupCache())


class TestThread(TestContextBase, TransactionTestCase):
    """Test thread isolation for application context."""

    worker_token: Token

    @override
    def setUp(self) -> None:
        super().setUp()
        with transaction.atomic():
            self.worker_token = self.playground.create_worker_token()

    def run_in_thread(
        self, func: Callable[..., Any], *args: Any, **kwargs: Any
    ) -> Any:
        """Run a callable in a thread."""
        result: Any = None

        def _thread_main() -> None:
            nonlocal result
            try:
                result = func(*args, **kwargs)
            finally:
                connection.close()

        thread = Thread(target=_thread_main)
        thread.start()
        thread.join()

        return result

    def test_visibility_thread(self) -> None:
        """Check visibility with subthreads."""
        scope1 = self.playground.get_or_create_scope(name="scope1")
        workspace1 = self.playground.create_workspace(
            scope=scope1, name="test1", public=True
        )
        user1 = self.playground.create_user("test1")

        self.assert_context_initial()
        self.scenario.set_current()
        context.lookup_cache.single[self.lookup_cache_single_key] = (
            self.lookup_cache_single_value
        )

        def _test() -> None:
            # Application context is cleared when changing Thread
            self.assert_context_initial()

            context.set_permission_context_scoped(
                scope1, PermissionContext.for_ui_user(user1)
            )
            workspace1.set_current()
            context.lookup_cache.single[self.lookup_cache_single_key] = (
                self.lookup_cache_single_value
            )

            self.assertEqual(context.scope, scope1)
            self.assertIsNotNone(context._scope_roles.get())
            self.assertEqual(context.user, user1)
            self.assertEqual(context.workspace, workspace1)
            self.assertIsNotNone(context._workspace_roles.get())

        self.run_in_thread(_test)

        self.assert_context_all_set()
