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

from unittest import mock

from debusine.db.models import Group
from debusine.db.permissioncontext import PermissionContext
from debusine.db.playground import scenarios
from debusine.test.django import TestCase


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

    scenario = scenarios.DefaultContextAPI(set_current=True)

    def test_str(self) -> None:
        external_user_token = self.playground.create_external_user_token()
        worker_activation_token = (
            self.playground.create_worker_activation_token()
        )
        worker_token = self.playground.create_worker_token()
        work_request_token = self.playground.create_work_request_token()
        assert work_request_token.work_request is not None
        pc_with_extra_groups = self.playground.create_pc_with_extra_groups(
            self.scenario.scope_owners
        )
        for value, expected in (
            (PermissionContext.for_ui_anonymous(), "AnonymousUser"),
            (
                PermissionContext.for_ui_user(self.scenario.user),
                str(self.scenario.user),
            ),
            (
                PermissionContext.for_api_user(self.scenario.user_token),
                f"{self.scenario.user}[user token]",
            ),
            (
                PermissionContext.for_api_external_user(external_user_token),
                "AnonymousUser[external user token]",
            ),
            (
                PermissionContext.for_worker(worker_token),
                "AnonymousUser[worker token]",
            ),
            (
                PermissionContext.for_worker_activation(
                    worker_activation_token
                ),
                "AnonymousUser[worker-activation token]",
            ),
            (
                PermissionContext.for_work_request(token=work_request_token),
                f"{self.scenario.user}[work-request token]"
                f"[work request {work_request_token.work_request}]",
            ),
            (
                work_request_token.work_request.pc,
                f"{self.scenario.user}"
                f"[work request {work_request_token.work_request}]",
            ),
            (
                pc_with_extra_groups,
                f"{self.scenario.user}"
                f"[work request {pc_with_extra_groups.work_request}]"
                f"[extra groups {self.scenario.scope_owners}]",
            ),
        ):
            with self.subTest(value=repr(value)):
                self.assertEqual(str(value), expected)

    def test_compare(self) -> None:
        for first, second, expected in (
            (
                PermissionContext.for_ui_anonymous(),
                PermissionContext.for_ui_anonymous(),
                True,
            ),
            (self.scenario.context_ui(), self.scenario.context_ui(), True),
            (
                self.scenario.context_ui(),
                PermissionContext.for_ui_anonymous(),
                False,
            ),
            (
                self.scenario.context_ui(),
                PermissionContext.for_api_user(self.scenario.user_token),
                False,
            ),
            (
                self.scenario.context_ui(),
                self.scenario.context_ui().with_permission_checks_disabled(),
                False,
            ),
            (
                self.scenario.context_ui(),
                42,
                False,
            ),
        ):
            with self.subTest(first=repr(first), second=repr(second)):
                self.assertEqual(first == second, expected)

    def test_groups(self) -> None:
        self.scenario.workspace_owners.users.add(self.scenario.user)
        for pc, groups in (
            (PermissionContext.for_ui_anonymous(), []),
            (self.scenario.context_ui(), [self.scenario.workspace_owners]),
            (
                self.playground.create_pc_with_extra_groups(
                    self.scenario.scope_owners
                ),
                [self.scenario.workspace_owners, self.scenario.scope_owners],
            ),
        ):
            with self.subTest(pc=str(pc)):
                self.assertQuerySetEqual(pc.groups(), groups, ordered=False)

    def test_explicit_grants_only(self) -> None:
        external_user_token = self.playground.create_external_user_token()
        worker_activation_token = (
            self.playground.create_worker_activation_token()
        )
        worker_token = self.playground.create_worker_token()
        work_request_token = self.playground.create_work_request_token()
        work_request = self.playground.create_worker_task()
        for pc in (
            PermissionContext.for_api_user(self.scenario.user_token),
            PermissionContext.for_maintenance(),
            PermissionContext.for_ui_anonymous(),
            PermissionContext.for_ui_user(self.scenario.user),
            PermissionContext.for_work_request(token=work_request_token),
            PermissionContext.for_work_request(work_request=work_request),
            PermissionContext.for_worker(worker_token),
            PermissionContext.for_worker_activation(worker_activation_token),
        ):
            with self.subTest(pc=repr(pc)):
                self.assertFalse(pc.explicit_grants_only)

        for pc in (
            PermissionContext.for_api_external_user(external_user_token),
        ):
            with self.subTest(pc=repr(pc)):
                self.assertTrue(pc.explicit_grants_only)

    def test_is_non_permission_bearing_token(self) -> None:
        external_user_token = self.playground.create_external_user_token()
        worker_activation_token = (
            self.playground.create_worker_activation_token()
        )
        worker_token = self.playground.create_worker_token()
        work_request_token = self.playground.create_work_request_token()
        work_request = self.playground.create_worker_task()
        for pc in (
            PermissionContext.for_api_external_user(external_user_token),
            PermissionContext.for_api_user(self.scenario.user_token),
            PermissionContext.for_maintenance(),
            PermissionContext.for_ui_anonymous(),
            PermissionContext.for_ui_user(self.scenario.user),
            PermissionContext.for_work_request(token=work_request_token),
            PermissionContext.for_work_request(work_request=work_request),
        ):
            with self.subTest(pc=repr(pc)):
                self.assertFalse(pc.is_non_permission_bearing_token)

        for pc in (
            PermissionContext.for_worker(worker_token),
            PermissionContext.for_worker_activation(worker_activation_token),
        ):
            with self.subTest(pc=repr(pc)):
                self.assertTrue(pc.is_non_permission_bearing_token)

    def test_for_work_request_extra_groups(self) -> None:
        group1 = self.playground.create_group("group1")
        self.enterContext(
            mock.patch(
                "debusine.db.models.WorkRequest.get_extra_groups",
                return_value=Group.objects.filter(pk=group1.pk),
            )
        )
        token = self.playground.create_work_request_token()
        assert token.work_request is not None
        self.assertQuerySetEqual(
            PermissionContext.for_work_request(token=token).extra_groups,
            (group1,),
        )
        self.assertQuerySetEqual(
            PermissionContext.for_work_request(
                work_request=token.work_request
            ).extra_groups,
            (group1,),
        )
