govoplan-core/tests/test_reference_options.py

from __future__ import annotations

import unittest
from types import SimpleNamespace

from govoplan_core.core.access import (
    CAPABILITY_ACCESS_DIRECTORY,
    AccountRef,
    GroupRef,
    UserRef,
)
from govoplan_core.core.references import (
    CAPABILITY_ACCESS_REFERENCE_OPTIONS,
    ReferenceOption,
    ReferenceSearchPage,
    access_scope_reference_page,
    access_scope_reference_options,
    validate_access_scope_reference,
)


class _Directory:
    users = (
        UserRef(
            id="membership-1",
            account_id="account-1",
            tenant_id="tenant-1",
            email="ada@example.test",
            display_name="Ada",
        ),
        UserRef(
            id="membership-2",
            account_id="account-2",
            tenant_id="tenant-1",
            email="inactive@example.test",
            display_name="Inactive",
            status="inactive",
        ),
    )
    groups = (
        GroupRef(id="group-1", tenant_id="tenant-1", name="Finance"),
        GroupRef(
            id="group-2",
            tenant_id="tenant-1",
            name="Former team",
            status="inactive",
        ),
    )

    def get_account(self, account_id):
        if account_id not in {"account-1", "account-2"}:
            return None
        return AccountRef(id=account_id, email=f"{account_id}@example.test")

    def get_user(self, user_id):
        return next((item for item in self.users if item.id == user_id), None)

    def get_users(self, user_ids):
        return {item.id: item for item in self.users if item.id in set(user_ids)}

    def users_for_tenant(self, tenant_id):
        return tuple(item for item in self.users if item.tenant_id == tenant_id)

    def get_group(self, group_id):
        return next((item for item in self.groups if item.id == group_id), None)

    def get_groups(self, group_ids):
        return {item.id: item for item in self.groups if item.id in set(group_ids)}

    def groups_for_tenant(self, tenant_id):
        return tuple(item for item in self.groups if item.tenant_id == tenant_id)

    def groups_for_user(self, user_id, *, tenant_id):
        del user_id
        return self.groups_for_tenant(tenant_id)

    def display_label(self, subject):
        return subject.label


class _Registry:
    directory = _Directory()

    def has_capability(self, name):
        return name == CAPABILITY_ACCESS_DIRECTORY

    def capability(self, name):
        return self.directory if self.has_capability(name) else None


class _ReferenceProvider:
    def __init__(self) -> None:
        self.requests = []

    def search_reference_options(self, session, principal, *, request):
        del session, principal
        self.requests.append(request)
        return ReferenceSearchPage(
            options=(
                ReferenceOption(
                    value="account-1",
                    label="Ada",
                    kind="user",
                    source_module="access",
                ),
            ),
            next_cursor="offset:1",
            has_more=True,
        )


class _ProviderRegistry:
    def __init__(self) -> None:
        self.provider = _ReferenceProvider()

    def has_capability(self, name):
        return name == CAPABILITY_ACCESS_REFERENCE_OPTIONS

    def capability(self, name):
        return self.provider if self.has_capability(name) else None


class ReferenceOptionTests(unittest.TestCase):
    def setUp(self) -> None:
        self.principal = SimpleNamespace(
            tenant_id="tenant-1",
            account_id="account-1",
            display_name="Ada",
            email="ada@example.test",
            group_ids=frozenset({"group-1"}),
        )

    def test_user_values_are_canonical_account_ids(self) -> None:
        options = access_scope_reference_options(
            _Registry(),
            self.principal,
            scope_type="user",
            administrative=True,
        )

        self.assertEqual(["account-1", "account-2"], [item.value for item in options])
        self.assertEqual("membership-1", options[0].provenance["membership_id"])
        self.assertTrue(options[1].disabled)
        self.assertEqual("inactive", options[1].availability)

    def test_unavailable_selected_values_remain_readable(self) -> None:
        options = access_scope_reference_options(
            _Registry(),
            self.principal,
            scope_type="group",
            query="missing",
            selected_values=("removed-group",),
            administrative=True,
        )

        self.assertEqual(1, len(options))
        self.assertEqual("removed-group", options[0].value)
        self.assertEqual("unavailable", options[0].availability)
        self.assertTrue(options[0].disabled)

    def test_non_administrators_only_receive_permitted_scope_targets(self) -> None:
        users = access_scope_reference_options(
            _Registry(),
            self.principal,
            scope_type="user",
            administrative=False,
        )
        groups = access_scope_reference_options(
            _Registry(),
            self.principal,
            scope_type="group",
            administrative=False,
        )

        self.assertEqual(["account-1"], [item.value for item in users])
        self.assertEqual(["group-1"], [item.value for item in groups])

    def test_inactive_targets_are_rejected_but_existing_values_survive(self) -> None:
        with self.assertRaisesRegex(ValueError, "Inactive groups"):
            validate_access_scope_reference(
                _Registry(),
                tenant_id="tenant-1",
                scope_type="group",
                scope_id="group-2",
            )
        self.assertEqual(
            "group-2",
            validate_access_scope_reference(
                _Registry(),
                tenant_id="tenant-1",
                scope_type="group",
                scope_id="group-2",
                preserve_existing="group-2",
            ),
        )

    def test_missing_optional_access_provider_keeps_reduced_mode(self) -> None:
        options = access_scope_reference_options(
            None,
            self.principal,
            scope_type="user",
        )

        self.assertEqual("account-1", options[0].value)
        self.assertEqual("core", options[0].source_module)

    def test_bounded_provider_receives_normalized_search_and_retains_stale_selection(self) -> None:
        registry = _ProviderRegistry()

        page = access_scope_reference_page(
            registry,
            self.principal,
            scope_type="user",
            query="  Ada  ",
            selected_values=("removed-account",),
            limit=500,
            cursor="offset:20",
            administrative=True,
            session=object(),
        )

        request = registry.provider.requests[0]
        self.assertEqual("ada", request.query)
        self.assertEqual(200, request.limit)
        self.assertEqual("offset:20", request.cursor)
        self.assertTrue(request.context["administrative"])
        self.assertEqual(
            ["account-1", "removed-account"],
            [option.value for option in page.options],
        )
        self.assertEqual("unavailable", page.options[1].availability)
        self.assertTrue(page.has_more)
        self.assertEqual("offset:1", page.next_cursor)


if __name__ == "__main__":
    unittest.main()