govoplan-core/tests/test_ownership_api.py

from __future__ import annotations

from datetime import datetime, timezone
from types import SimpleNamespace
import unittest

from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from govoplan_core.auth import ApiPrincipal, get_api_principal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.modules import ModuleManifest
from govoplan_core.core.ownership import (
    OwnershipActionDecision,
    OwnershipProviderRegistration,
    OwnershipSubjectRef,
    OwnershipTransfer,
    OwnershipTransferError,
)
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.core.runtime import (
    clear_runtime,
    configure_runtime,
    get_runtime_context,
)
from govoplan_core.db.base import Base
from govoplan_core.db.session import get_session
from govoplan_core.server.ownership import router


class _Provider:
    def __init__(self) -> None:
        self.owner = OwnershipSubjectRef(type="user", id="owner-1")

    def current_owner(self, session, *, tenant_id: str, resource_id: str):
        if tenant_id == "tenant-1" and resource_id == "resource-1":
            return self.owner
        return None

    def authorize_ownership_action(
        self,
        session,
        *,
        tenant_id: str,
        resource_id: str,
        action: str,
        actor: OwnershipSubjectRef,
        current_owner: OwnershipSubjectRef,
        target_owner: OwnershipSubjectRef,
    ) -> OwnershipActionDecision:
        del session, tenant_id, resource_id
        if action in {
            "propose_transfer",
            "approve_requested_transfer",
            "cancel_transfer",
        }:
            allowed = actor.id == current_owner.id
        elif action in {
            "request_ownership",
            "accept_transfer",
            "decline_transfer",
        }:
            allowed = actor.id == target_owner.id
        elif action == "view_transfer":
            allowed = actor.id in {current_owner.id, target_owner.id}
        else:
            allowed = False
        return OwnershipActionDecision(allowed=allowed)

    def apply_owner(
        self,
        session,
        *,
        tenant_id: str,
        resource_id: str,
        expected_owner: OwnershipSubjectRef,
        target_owner: OwnershipSubjectRef,
        actor: OwnershipSubjectRef,
        reason: str | None,
    ) -> None:
        del session, tenant_id, resource_id, actor, reason
        if (
            self.owner.type != expected_owner.type
            or self.owner.id != expected_owner.id
        ):
            raise OwnershipTransferError("Owner changed")
        self.owner = target_owner


class OwnershipApiTests(unittest.TestCase):
    def setUp(self) -> None:
        self.runtime_context = get_runtime_context()
        clear_runtime()
        self.engine = create_engine(
            "sqlite:///:memory:",
            connect_args={"check_same_thread": False},
            poolclass=StaticPool,
        )
        Base.metadata.create_all(
            self.engine,
            tables=[OwnershipTransfer.__table__],
        )
        self.session = sessionmaker(
            bind=self.engine,
            expire_on_commit=False,
        )()
        self.provider = _Provider()
        registry = PlatformRegistry()
        registry.register(
            ModuleManifest(
                id="example",
                name="Example",
                version="1.0.0",
                ownership_providers=(
                    OwnershipProviderRegistration(
                        resource_type="example",
                        provider=self.provider,
                    ),
                ),
            )
        )
        self.principal = _principal("owner-1")
        app = FastAPI()
        app.state.govoplan_registry = registry
        app.include_router(router, prefix="/api/v1")
        app.dependency_overrides[get_session] = lambda: self.session
        app.dependency_overrides[get_api_principal] = lambda: self.principal
        self.client = TestClient(app)

    def tearDown(self) -> None:
        self.client.close()
        self.session.close()
        Base.metadata.drop_all(
            self.engine,
            tables=[OwnershipTransfer.__table__],
        )
        self.engine.dispose()
        if self.runtime_context is not None:
            configure_runtime(self.runtime_context)

    def test_owner_transfer_requires_target_acceptance_and_is_listable(self) -> None:
        created = self.client.post(
            "/api/v1/ownership/transfers",
            json={
                "resource": {
                    "module_id": "example",
                    "resource_type": "example",
                    "resource_id": "resource-1",
                },
                "target_owner": {"type": "user", "id": "target-1"},
                "idempotency_key": "api-transfer-1",
                "reason": "Handover",
            },
        )
        self.assertEqual(201, created.status_code, created.text)
        transfer_id = created.json()["id"]
        self.assertEqual(
            "awaiting_target_acceptance",
            created.json()["status"],
        )
        self.assertEqual(
            ["started"],
            [item["action"] for item in created.json()["decisions"]],
        )

        visible = self.client.get("/api/v1/ownership/transfers")
        self.assertEqual(200, visible.status_code, visible.text)
        self.assertEqual([transfer_id], [item["id"] for item in visible.json()["transfers"]])

        self.principal = _principal("target-1")
        accepted = self.client.post(
            f"/api/v1/ownership/transfers/{transfer_id}/acceptance"
        )
        self.assertEqual(200, accepted.status_code, accepted.text)
        self.assertEqual("completed", accepted.json()["status"])
        self.assertEqual("target-1", self.provider.owner.id)
        self.assertEqual(
            ["started", "accepted"],
            [item["action"] for item in accepted.json()["decisions"]],
        )

    def test_resource_module_must_match_registered_provider(self) -> None:
        response = self.client.post(
            "/api/v1/ownership/transfers",
            json={
                "resource": {
                    "module_id": "wrong-module",
                    "resource_type": "example",
                    "resource_id": "resource-1",
                },
                "target_owner": {"type": "user", "id": "target-1"},
                "idempotency_key": "wrong-provider-module",
            },
        )
        self.assertEqual(404, response.status_code, response.text)


def _principal(user_id: str) -> ApiPrincipal:
    return ApiPrincipal(
        principal=PrincipalRef(
            account_id=f"account-{user_id}",
            membership_id=user_id,
            tenant_id="tenant-1",
            scopes=frozenset({"campaigns:campaign:share"}),
        ),
        account=SimpleNamespace(id=f"account-{user_id}"),
        user=SimpleNamespace(id=user_id),
        auth_session=SimpleNamespace(created_at=datetime.now(timezone.utc)),
    )


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