# 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 auth-related models."""

from typing import ClassVar

from asgiref.sync import sync_to_async
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.db import IntegrityError
from django.utils import timezone

from debusine.db.models import (
    Group,
    Identity,
    Scope,
    Token,
    User,
    Worker,
    Workspace,
)
from debusine.db.models.auth import system_user
from debusine.db.playground import scenarios
from debusine.test.django import ChannelsHelpersMixin, TestCase


class TokenTests(ChannelsHelpersMixin, TestCase):
    """Unit tests of the ``Token`` model."""

    scenario = scenarios.DefaultScopeUser()

    def test_save(self) -> None:
        """The model creates a Token.key on save or keeps it if it existed."""
        token = Token.objects.create(user=self.scenario.user)

        self.assertIsNotNone(token.id)
        self.assertEqual(len(token.key), 64)

        key = token.key
        token.save()
        self.assertEqual(token.key, key)

    def test_str(self) -> None:
        """Test Token.__str__."""
        token = Token.objects.create(user=self.scenario.user)
        self.assertEqual(token.__str__(), token.hash)

    def test_user_field(self) -> None:
        """Test User field is None by default."""
        token = Token.objects.create()
        self.assertIsNone(token.user)

    def test_get_token_or_none_found(self) -> None:
        """get_token_or_none looks up a token and returns it."""
        token_hash = Token._generate_hash('some_key')
        token = Token.objects.create(hash=token_hash)

        self.assertEqual(token, Token.objects.get_token_or_none('some_key'))

        # Worker is supposed to be prefetched, but it doesn't work in this
        # case: django bug? This is important because hasattr looks like a
        # harmless thing to call from async code
        with self.assertNumQueries(1):
            self.assertFalse(hasattr(token, "worker"))

    def test_get_token_or_none_not_found(self) -> None:
        """get_token_or_none cannot find a token and returns None."""
        self.assertIsNone(Token.objects.get_token_or_none('a_non_existing_key'))

    def test_get_token_worker_refetch(self) -> None:
        """get_token_or_none prefetches the worker attribute if present."""
        token_hash = Token._generate_hash('some_key')
        token = Token.objects.create(hash=token_hash)
        worker = self.playground.create_worker()
        worker.token = token

        self.assertEqual(token, Token.objects.get_token_or_none('some_key'))

        # worker is prefetched
        with self.assertNumQueries(0):
            self.assertEqual(token.worker, worker)

    def test_enable(self) -> None:
        """enable() enables the token."""
        token = Token.objects.create()

        # Assert the default is disabled tokens
        self.assertFalse(token.enabled)

        token.enable()
        token.refresh_from_db()

        self.assertTrue(token.enabled)

    async def test_disable(self) -> None:
        """disable() disables the token."""
        token = await Token.objects.acreate(enabled=True)
        await Worker.objects.acreate(token=token, registered_at=timezone.now())

        channel = await self.create_channel(token.hash)

        await sync_to_async(token.disable)()

        await self.assert_channel_received(channel, {"type": "worker.disabled"})
        await token.arefresh_from_db()

        self.assertFalse(token.enabled)


class TokenManagerTests(TestCase):
    """Unit tests for the ``TokenManager`` class."""

    user_john: ClassVar[User]
    user_bev: ClassVar[User]
    token_john: ClassVar[Token]
    token_bev: ClassVar[Token]

    @classmethod
    def setUpTestData(cls) -> None:
        """Test data used by all the tests."""
        super().setUpTestData()
        cls.user_john = get_user_model().objects.create_user(
            username="John", email="john@example.com"
        )
        cls.user_bev = get_user_model().objects.create_user(
            username="Bev", email="bev@example.com"
        )
        cls.token_john = Token.objects.create(user=cls.user_john)
        cls.token_bev = Token.objects.create(user=cls.user_bev)

    def test_get_tokens_all(self) -> None:
        """get_tokens returns all the tokens if no filter is applied."""
        self.assertQuerySetEqual(
            Token.objects.get_tokens(),
            {self.token_bev, self.token_john},
            ordered=False,
        )

    def test_get_tokens_by_owner(self) -> None:
        """get_tokens returns the correct tokens when filtering by owner."""
        self.assertQuerySetEqual(
            Token.objects.get_tokens(username='John'), [self.token_john]
        )
        self.assertQuerySetEqual(
            Token.objects.get_tokens(username='Bev'), [self.token_bev]
        )
        self.assertQuerySetEqual(
            Token.objects.get_tokens(username='Someone'), []
        )

    def test_get_tokens_by_key(self) -> None:
        """get_tokens returns the correct tokens when filtering by key."""
        self.assertQuerySetEqual(
            Token.objects.get_tokens(key=self.token_john.key),
            [self.token_john],
        )
        self.assertQuerySetEqual(
            Token.objects.get_tokens(key='non-existing-key'), []
        )

    def test_get_tokens_by_key_owner_empty(self) -> None:
        """
        get_tokens returns nothing if using a key and username without matches.

        Key for the key parameter or username for the user parameter exist
        but are for different tokens.
        """
        self.assertQuerySetEqual(
            Token.objects.get_tokens(
                key=self.token_john.key, username=self.user_bev.username
            ),
            [],
        )


class UserTests(TestCase):
    """Tests for the User class."""

    scenario = scenarios.DefaultScopeUser()
    user: ClassVar[User]
    system_user: ClassVar[User]

    @classmethod
    def setUpTestData(cls) -> None:
        """Set up common test data."""
        super().setUpTestData()
        cls.user = cls.playground.create_user("test")
        cls.system_user = system_user()

    def test_can_display_anonymous(self) -> None:
        """Test can_display with anonymous user."""
        self.assertPermission(
            "can_display",
            AnonymousUser(),
            denied=[self.scenario.user, self.user, self.system_user],
        )

    def test_can_display(self) -> None:
        """Test the can_display predicate."""
        self.assertPermission(
            "can_display",
            [self.scenario.user, self.user, self.system_user],
            allowed=[self.scenario.user, self.user, self.system_user],
        )

    def test_can_manage_anonymous(self) -> None:
        """Test can_manage with anonymous user."""
        self.assertPermission(
            "can_manage",
            AnonymousUser(),
            denied=[self.scenario.user, self.user, self.system_user],
        )

    def test_can_manage(self) -> None:
        """Test can_manage."""
        self.assertPermission(
            "can_manage",
            self.scenario.user,
            allowed=self.scenario.user,
            denied=self.user,
        )
        self.assertPermission(
            "can_manage",
            self.user,
            allowed=self.user,
            denied=self.scenario.user,
        )

    def test_get_absolute_url(self) -> None:
        """Test the get_absolute_url method."""
        self.assertEqual(
            self.scenario.user.get_absolute_url(),
            f"/-/user/{self.scenario.user.username}/",
        )


class IdentityTests(TestCase):
    """Tests for the Identity class."""

    def test_str(self) -> None:
        """Stringification should show the unique key."""
        ident = Identity(issuer="salsa", subject="test@debian.org")
        self.assertEqual(str(ident), "salsa:test@debian.org")


class GroupManagerTests(TestCase):
    """Test for GroupManager class."""

    scope1: ClassVar[Scope]
    scope2: ClassVar[Scope]
    group1: ClassVar[Group]
    group2: ClassVar[Group]

    @classmethod
    def setUpTestData(cls) -> None:
        """Set up common test data."""
        super().setUpTestData()
        cls.scope1 = cls.playground.get_or_create_scope("scope1")
        cls.group1 = Group.objects.create(scope=cls.scope1, name="group")
        cls.scope2 = cls.playground.get_or_create_scope("scope2")
        cls.group2 = Group.objects.create(scope=cls.scope2, name="group")

    def test_from_scoped_name(self) -> None:
        """Test from_scoped_name."""
        for arg, result in (
            ("scope1/group", self.group1),
            ("scope2/group", self.group2),
        ):
            with self.subTest(arg=arg):
                self.assertEqual(Group.objects.from_scoped_name(arg), result)

    def test_from_scoped_name_fail_lookup(self) -> None:
        """Test from_scoped_name."""
        for arg, exc_class in (
            ("scope1/fail", Group.DoesNotExist),
            ("scope2:group", ValueError),
        ):
            with self.subTest(arg=arg), self.assertRaises(exc_class):
                Group.objects.from_scoped_name(arg)

    def test_create_ephemeral(self) -> None:
        """Test create_ephemeral."""
        user = self.playground.get_default_user()
        group = Group.objects.create_ephemeral(
            scope=self.scope1, name="test", owner=user
        )
        self.assertEqual(group.scope, self.scope1)
        self.assertEqual(group.name, "test")
        self.assertQuerySetEqual(group.users.all(), [user])

    def test_unused_ephemeral_unused(self) -> None:
        """Test unused_ephemeral."""
        self.assertQuerySetEqual(Group.objects.unused_ephemeral(), [])

        self.group1.ephemeral = True
        self.group1.save()
        self.assertQuerySetEqual(
            Group.objects.unused_ephemeral(), [self.group1]
        )

    def test_unused_ephemeral_used_by_scope(self) -> None:
        """Test unused_ephemeral for a group used by a Scope."""
        self.group1.ephemeral = True
        self.group1.save()
        self.assertQuerySetEqual(
            Group.objects.unused_ephemeral(), [self.group1]
        )

        self.group1.assign_role(self.scope1, Scope.Roles.OWNER)
        self.assertQuerySetEqual(Group.objects.unused_ephemeral(), [])

    def test_unused_ephemeral_used_by_workspace(self) -> None:
        """Test unused_ephemeral for a group used by a Workspace."""
        self.group1.ephemeral = True
        self.group1.save()
        self.assertQuerySetEqual(
            Group.objects.unused_ephemeral(), [self.group1]
        )

        workspace = self.playground.create_workspace(scope=self.scope1)
        self.group1.assign_role(workspace, workspace.Roles.OWNER)
        self.assertQuerySetEqual(Group.objects.unused_ephemeral(), [])

    def test_unused_ephemeral_used_by_asset(self) -> None:
        """Test unused_ephemeral for a group used by an Asset."""
        self.group1.ephemeral = True
        self.group1.save()
        self.assertQuerySetEqual(
            Group.objects.unused_ephemeral(), [self.group1]
        )

        workspace = self.playground.create_workspace(scope=self.scope1)
        asset = self.playground.create_signing_key_asset(workspace=workspace)
        self.group1.assign_role(asset, asset.Roles.OWNER)
        self.assertQuerySetEqual(Group.objects.unused_ephemeral(), [])

    def test_unused_ephemeral_used_by_asset_usage(self) -> None:
        """Test unused_ephemeral for a group used by an AssetUsage."""
        self.group1.ephemeral = True
        self.group1.save()
        self.assertQuerySetEqual(
            Group.objects.unused_ephemeral(), [self.group1]
        )

        workspace = self.playground.create_workspace(scope=self.scope1)
        asset = self.playground.create_signing_key_asset(workspace=workspace)
        asset_usage = self.playground.create_asset_usage(
            asset, workspace=workspace
        )
        self.group1.assign_role(asset_usage, asset_usage.Roles.SIGNER)
        self.assertQuerySetEqual(Group.objects.unused_ephemeral(), [])


class GroupTests(TestCase):
    """Test for Group class."""

    scenario = scenarios.DefaultContext()

    group: ClassVar[Group]

    @classmethod
    def setUpTestData(cls) -> None:
        """Set up common test data."""
        super().setUpTestData()
        cls.group = Group.objects.create(name="group", scope=cls.scenario.scope)

    def test_str(self) -> None:
        """Test stringification."""
        self.assertEqual(str(self.group), f"{self.scenario.scope.name}/group")

    def test_defaults(self) -> None:
        """Test default values."""
        self.assertFalse(self.group.ephemeral)

    def test_unique(self) -> None:
        """Test scoped uniqueness."""
        scope2 = self.playground.get_or_create_scope("scope2")
        group2 = Group.objects.create(name=self.group.name, scope=scope2)
        self.assertNotEqual(self.group.pk, group2.pk)

        with self.assertRaisesRegex(
            IntegrityError,
            # Only match constraint name to support non-english locales
            # r"duplicate key value violates unique constraint"
            r"db_group_unique_name_scope",
        ):
            Group.objects.create(name=self.group.name, scope=self.group.scope)

    def test_assign_user(self) -> None:
        """Test assigning users to groups."""
        user1 = User.objects.create_user(
            username="user1",
            email="user1@debian.org",
        )
        user2 = User.objects.create_user(
            username="user2",
            email="user2@debian.org",
        )

        self.group.users.add(user1)

        self.assertQuerySetEqual(self.group.users.all(), [user1])
        self.assertQuerySetEqual(user1.groups.all(), [])
        self.assertQuerySetEqual(user1.debusine_groups.all(), [self.group])

        self.group.users.add(user2)

        self.assertQuerySetEqual(
            self.group.users.all(), [user1, user2], ordered=False
        )
        self.assertQuerySetEqual(user2.groups.all(), [])
        self.assertQuerySetEqual(user2.debusine_groups.all(), [self.group])

    def test_assign_role(self) -> None:
        """Test assign_role."""
        for resource, role in (
            (self.scenario.scope, Scope.Roles.OWNER),
            (self.scenario.scope, "owner"),
            (self.scenario.workspace, Workspace.Roles.OWNER),
            (self.scenario.workspace, "owner"),
        ):
            with self.subTest(resource=resource, role=role):
                assigned = self.group.assign_role(resource, role)
                # TODO: we can remove the type ignores once we find a way to
                # generically type the role assignment models
                self.assertEqual(
                    assigned.resource, resource  # type: ignore[attr-defined]
                )
                self.assertEqual(
                    assigned.group, self.group  # type: ignore[attr-defined]
                )
                self.assertEqual(
                    assigned.role, role  # type: ignore[attr-defined]
                )

    def test_assign_role_not_a_resource(self) -> None:
        """Try assigning role to something which is not a resource."""
        token = self.playground.create_bare_token()

        with self.assertRaisesRegex(
            NotImplementedError, r"Cannot get scope for Token object"
        ):
            self.group.assign_role(token, "owner")

    def test_assign_role_wrong_scope(self) -> None:
        """Try assigning role to a resource in a different scope."""
        scope2 = self.playground.get_or_create_scope("scope2")
        with self.assertRaisesRegex(
            ValueError, r"Scope 'scope2' is not in scope debusine"
        ):
            self.group.assign_role(scope2, "owner")

    def test_assign_role_wrong_role_enum(self) -> None:
        """Try assigning a role using the wrong enum."""
        with self.assertRaisesRegex(
            TypeError,
            r"ScopeRoles.OWNER cannot be converted to <enum 'WorkspaceRoles'>",
        ):
            self.group.assign_role(self.scenario.workspace, Scope.Roles.OWNER)

    def test_assign_role_wrong_role_name(self) -> None:
        """Try assigning a role using the wrong enum."""
        with self.assertRaisesRegex(
            ValueError, r"'does-not-exist' is not a valid WorkspaceRoles"
        ):
            self.group.assign_role(self.scenario.workspace, "does-not-exist")
