tenant-engine/tests/test_api_reads.py

from datetime import UTC, datetime

from fastapi.testclient import TestClient
from helpers import AllowAllAuthorizer

from tenant_engine.app import create_app
from tenant_engine.domain import CapabilityRole, Tenant, create_role_grant
from tenant_engine.store import InMemoryTenantStore, TenantStore


def _client(store: TenantStore) -> TestClient:
    return TestClient(create_app(store=store, authorizer=AllowAllAuthorizer()))


class _BrokenStore:
    """Test double: every active_roles() call raises, simulating an outage."""

    def __init__(self, delegate: TenantStore) -> None:
        self._delegate = delegate

    def create_tenant(self, tenant):
        return self._delegate.create_tenant(tenant)

    def get_tenant(self, tenant_id):
        return self._delegate.get_tenant(tenant_id)

    def grant_role(self, grant):
        return self._delegate.grant_role(grant)

    def revoke_role(self, **kwargs):
        return self._delegate.revoke_role(**kwargs)

    def active_roles(self, tenant_id):
        from tenant_engine.store import StoreUnavailableError

        raise StoreUnavailableError("simulated outage")

    def assign_plan(self, assignment):
        return self._delegate.assign_plan(assignment)

    def events_for(self, tenant_id):
        return self._delegate.events_for(tenant_id)

    def record_authorization(self, record):
        return None


def _seeded_store() -> InMemoryTenantStore:
    store = InMemoryTenantStore()
    tenant = Tenant.create(tenant_id="t-binky", identifier="tenant:friendly:binky")
    store.create_tenant(tenant)
    store.grant_role(
        create_role_grant(
            tenant=tenant,
            grant_id="g-1",
            role=CapabilityRole.CUS,
            grant_reason="manual_grant",
            plan_id=None,
            granted_by="ops",
            correlation_id="corr-1",
            granted_at=datetime.now(UTC),
        )
    )
    return store


def test_cache_read_roles_returns_active_roles() -> None:
    client = _client(_seeded_store())
    response = client.get("/tenants/t-binky/roles", params={"actor": "key-cape"})

    assert response.status_code == 200
    assert response.json() == {"tenant_id": "t-binky", "roles": ["CUS"]}


def test_cache_read_roles_resolves_by_identifier_not_only_internal_id() -> None:
    """The real scenario key-cape's KEY-WP-0005-T02 hits: it only knows the

    tenant's profile identifier (the IAM Profile `tenant` claim value), via
    a URL path segment containing colons -- never the internal tenant_id.
    """
    client = _client(_seeded_store())
    response = client.get("/tenants/tenant:friendly:binky/roles", params={"actor": "key-cape"})

    assert response.status_code == 200
    assert response.json() == {"tenant_id": "tenant:friendly:binky", "roles": ["CUS"]}


def test_cache_read_roles_unknown_tenant_is_404() -> None:
    client = _client(_seeded_store())
    response = client.get("/tenants/does-not-exist/roles", params={"actor": "key-cape"})

    assert response.status_code == 404


def test_live_lookup_roles_returns_active_roles() -> None:
    client = _client(_seeded_store())
    response = client.get("/tenants/t-binky/roles/live", params={"actor": "flex-auth"})

    assert response.status_code == 200
    assert response.json()["roles"] == ["CUS"]


def test_live_lookup_fails_closed_on_store_outage() -> None:
    broken = _BrokenStore(_seeded_store())
    client = _client(broken)

    response = client.get("/tenants/t-binky/roles/live", params={"actor": "flex-auth"})

    assert response.status_code == 503
    assert response.json() != {"tenant_id": "t-binky", "roles": []}, (
        "outage must not be indistinguishable from a legitimate empty role list"
    )


def test_cache_read_also_fails_closed_on_store_outage() -> None:
    broken = _BrokenStore(_seeded_store())
    client = _client(broken)

    response = client.get("/tenants/t-binky/roles", params={"actor": "key-cape"})

    assert response.status_code == 503


def test_unauthorized_reads_are_denied_before_tenant_existence_is_observed() -> None:
    class _MustNotRead:
        def get_tenant(self, tenant_id):
            raise AssertionError(f"store read leaked for {tenant_id}")

        def active_roles(self, tenant_id):
            raise AssertionError(f"role read leaked for {tenant_id}")

    client = TestClient(create_app(store=_MustNotRead()))
    for path in (
        "/tenants/known?actor=unbound",
        "/tenants/unknown?actor=unbound",
        "/tenants/known/roles?actor=unbound",
        "/tenants/unknown/roles/live?actor=unbound",
    ):
        response = client.get(path)
        assert response.status_code == 403
        assert response.json()["error_code"] == "write_denied"


def test_read_requires_an_explicit_actor() -> None:
    client = _client(_seeded_store())
    assert client.get("/tenants/t-binky").status_code == 422
    assert client.get("/tenants/t-binky/roles").status_code == 422
    assert client.get("/tenants/t-binky/roles/live").status_code == 422