RAGcore-Public/tests/unit/application/control/test_spaces.py

"""Unit tests for control-plane space and grant use cases."""

from __future__ import annotations

from uuid import uuid4

import pytest

from ragcore.application.control import (
    EffectiveSpaceResolver,
    GrantService,
    KnowledgeSpaceLifecycle,
)
from ragcore.domain.control import (
    DataPolicy,
    ProviderKind,
    Role,
    SpaceOperation,
    Workspace,
)


def setup_control_plane() -> tuple[Workspace, object, object, object]:
    workspace = Workspace(uuid4(), "Platform", "platform")
    application = workspace.create_application(
        id=uuid4(),
        name="Client",
        scopes=(),
        default_knowledge_space_ids=frozenset(),
    )
    private = workspace.create_knowledge_space(
        id=uuid4(), name="Private", slug="private", data_policy=DataPolicy.LOCAL_ONLY
    )
    shared = workspace.create_knowledge_space(
        id=uuid4(), name="Shared", slug="shared", data_policy=DataPolicy.LOCAL_PREFERRED
    )
    return workspace, application, private, shared


@pytest.mark.parametrize(
    ("policy", "provider", "operation", "allowed"),
    [
        (DataPolicy.LOCAL_ONLY, ProviderKind.LOCAL, SpaceOperation.ANSWER, True),
        (DataPolicy.LOCAL_ONLY, ProviderKind.CLOUD, SpaceOperation.SEARCH, False),
        (DataPolicy.LOCAL_PREFERRED, ProviderKind.CLOUD, SpaceOperation.ANSWER, True),
        (DataPolicy.RETRIEVAL_ONLY, ProviderKind.LOCAL, SpaceOperation.SEARCH, True),
        (DataPolicy.NO_GENERATION, ProviderKind.CLOUD, SpaceOperation.CONTEXT, True),
        (DataPolicy.NO_GENERATION, ProviderKind.LOCAL, SpaceOperation.ANSWER, False),
    ],
)
def test_data_policy_excludes_incompatible_provider_operations(
    policy: DataPolicy, provider: ProviderKind, operation: SpaceOperation, allowed: bool
) -> None:
    workspace = Workspace(uuid4(), "Platform", "platform")
    space = workspace.create_knowledge_space(
        id=uuid4(), name="Policy", slug="policy", data_policy=policy
    )

    assert space.allows(operation=operation, provider=provider) is allowed


def test_effective_spaces_are_requested_grants_and_policy_intersection() -> None:
    workspace, application, private, shared = setup_control_plane()
    user_id = uuid4()
    grants = GrantService()
    app_grant = grants.application_grant(
        workspace, id=uuid4(), application=application, space=private, role=Role.VIEWER
    )
    user_grant = grants.user_grant(
        workspace, id=uuid4(), user_id=user_id, space=shared, role=Role.VIEWER
    )

    effective = EffectiveSpaceResolver().resolve(
        workspace_id=workspace.id,
        spaces=(private, shared),
        grants=(app_grant,),
        user_grants=(user_grant,),
        application=application,
        user_id=user_id,
        requested_space_ids=(private.id, shared.id),
    )

    assert effective == frozenset({private.id, shared.id})


@pytest.mark.parametrize(
    ("role", "operation"),
    [
        (Role.VIEWER, SpaceOperation.CONTEXT),
        (Role.VIEWER, SpaceOperation.ANSWER),
        (Role.AUDITOR, SpaceOperation.SEARCH),
    ],
)
def test_grant_role_limits_effective_space_operations(
    role: Role, operation: SpaceOperation
) -> None:
    workspace, application, private, _ = setup_control_plane()
    grant = GrantService().application_grant(
        workspace, id=uuid4(), application=application, space=private, role=role
    )

    effective = EffectiveSpaceResolver().resolve(
        workspace_id=workspace.id,
        spaces=(private,),
        grants=(grant,),
        application=application,
        requested_space_ids=(private.id,),
        operation=operation,
    )

    assert effective == frozenset()
def test_default_spaces_never_expand_grants() -> None:
    workspace, application, private, shared = setup_control_plane()
    application = workspace.create_application(
        id=application.id,
        name=application.name,
        scopes=(),
        default_knowledge_space_ids=frozenset({private.id, shared.id}),
    )
    grant = GrantService().application_grant(
        workspace, id=uuid4(), application=application, space=private, role=Role.VIEWER
    )

    effective = EffectiveSpaceResolver().resolve(
        workspace_id=workspace.id,
        spaces=(private, shared),
        grants=(grant,),
        application=application,
    )

    assert effective == frozenset({private.id})


def test_inactive_spaces_and_grants_are_excluded() -> None:
    workspace, application, private, shared = setup_control_plane()
    inactive_space = workspace.create_knowledge_space(
        id=uuid4(), name="Inactive", slug="inactive", active=False
    )
    grants = GrantService()
    private_grant = grants.application_grant(
        workspace,
        id=uuid4(),
        application=application,
        space=private,
        role=Role.VIEWER,
        active=False,
    )
    inactive_grant = grants.application_grant(
        workspace, id=uuid4(), application=application, space=inactive_space, role=Role.VIEWER
    )

    effective = EffectiveSpaceResolver().resolve(
        workspace_id=workspace.id,
        spaces=(private, shared, inactive_space),
        grants=(private_grant, inactive_grant),
        application=application,
        requested_space_ids=(private.id, inactive_space.id),
    )

    assert effective == frozenset()


def test_cross_workspace_spaces_and_grants_are_excluded() -> None:
    workspace, application, private, _ = setup_control_plane()
    other = Workspace(uuid4(), "Other", "other")
    other_space = other.create_knowledge_space(id=uuid4(), name="Other", slug="other")
    cross_grant = other.grant_user(
        id=uuid4(), user_id=uuid4(), knowledge_space=other_space, role=Role.VIEWER
    )

    effective = EffectiveSpaceResolver().resolve(
        workspace_id=workspace.id,
        spaces=(private, other_space),
        user_grants=(cross_grant,),
        application=application,
        user_id=cross_grant.user_id,
        requested_space_ids=(private.id, other_space.id),
    )

    assert effective == frozenset()


def test_lifecycle_transitions_preserve_identity_without_mutating_source() -> None:
    _, _, private, _ = setup_control_plane()
    lifecycle = KnowledgeSpaceLifecycle()

    inactive = lifecycle.deactivate(private)
    cloud = lifecycle.set_data_policy(inactive, DataPolicy.CLOUD_ALLOWED)
    active = lifecycle.activate(cloud)

    assert private.active is True
    assert inactive.active is False
    assert cloud.data_policy is DataPolicy.CLOUD_ALLOWED
    assert active.active is True
    assert active.id == private.id