kontextual-engine/tests/test_service_api.py

import pytest

from kontextual_engine import AuthorizationError, OperationContext, PolicyDecision, ServiceRuntime, create_app
from kontextual_engine.adapters.memory import InMemoryAssetRegistryRepository


def test_service_runtime_health_readiness_and_version_are_importable_without_fastapi() -> None:
    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())

    assert runtime.health()["status"] == "ok"
    assert runtime.readiness()["ready"] is True
    assert runtime.readiness()["checks"]["asset_registry"]["repository"] == (
        "InMemoryAssetRegistryRepository"
    )
    assert runtime.version()["api_version"] == "v1"


def test_service_runtime_exposes_asset_metadata_relationship_audit_and_policy_operations() -> None:
    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
    context = runtime.operation_context(actor_id="user-api", correlation_id="corr-api")

    created = runtime.create_asset(
        {
            "asset_id": "asset-api",
            "title": "API Asset",
            "classification": {
                "asset_type": "document",
                "sensitivity": "internal",
                "owner": "Platform Knowledge",
            },
            "metadata_records": [
                {
                    "key": "status",
                    "value": "draft",
                    "confirmed": True,
                    "provenance": {"producer": "api-test"},
                }
            ],
        },
        context,
    )
    successor = runtime.create_asset(
        {
            "asset_id": "asset-successor",
            "title": "Successor",
            "classification": {"asset_type": "document", "sensitivity": "public"},
        },
        context,
    )
    metadata = runtime.add_metadata_record(
        "asset-api",
        {"key": "reviewer", "value": "codex", "confirmed": True},
        context,
    )
    lifecycle = runtime.transition_lifecycle(
        "asset-api",
        {"lifecycle": "retired"},
        context,
    )
    relationship = runtime.create_relationship(
        {
            "source_asset_id": "asset-api",
            "target_id": "asset-successor",
            "predicate": "superseded_by",
            "target_kind": "asset",
            "confidence": 1.0,
        },
        context,
    )
    audit = runtime.list_audit_events(target="asset:asset-api")
    policy = runtime.evaluate_policy(
        {"action": "asset.retrieve", "resource": "asset:asset-api"},
        context,
    )

    assert created["asset"]["id"] == "asset-api"
    assert successor["asset"]["id"] == "asset-successor"
    assert runtime.get_asset("asset-api")["lifecycle"] == "retired"
    assert runtime.list_assets(asset_type="document")["count"] == 2
    assert metadata["version"]["change_type"] == "metadata_changed"
    assert lifecycle["asset"]["lifecycle"] == "retired"
    assert [record["key"] for record in runtime.list_metadata_records("asset-api")["items"]] == [
        "status",
        "reviewer",
    ]
    assert relationship["relationship"]["predicate"] == "superseded_by"
    assert runtime.list_relationships(source_id="asset-api")["count"] == 1
    assert [event["operation"] for event in audit["items"]] == [
        "asset.create",
        "asset.metadata.add",
        "asset.lifecycle.transition",
        "asset.relationship.add",
    ]
    assert policy["effect"] == "allow"


def test_service_runtime_policy_denial_blocks_protected_asset_operation() -> None:
    runtime = ServiceRuntime(
        repository=InMemoryAssetRegistryRepository(),
        policy_gateway=DenyCreatePolicy(),
    )

    with pytest.raises(AuthorizationError) as exc_info:
        runtime.create_asset(
            {
                "asset_id": "asset-denied",
                "title": "Denied",
                "classification": {"asset_type": "document"},
            },
            runtime.operation_context(actor_id="user-denied"),
        )

    assert exc_info.value.details["policy_decision"]["effect"] == "deny"
    assert runtime.list_assets()["count"] == 0
    assert runtime.list_audit_events(target="asset:asset-denied")["items"][0]["outcome"] == "denied"


def test_create_app_reports_missing_optional_dependency_when_fastapi_is_absent() -> None:
    try:
        import fastapi  # noqa: F401
    except ImportError:
        with pytest.raises(RuntimeError, match=r"kontextual-engine\[service\]"):
            create_app()
    else:
        pytest.skip("FastAPI is installed; missing-dependency path is not active")


@pytest.fixture
def client():
    pytest.importorskip("fastapi")
    pytest.importorskip("httpx")
    from fastapi.testclient import TestClient

    app = create_app(ServiceRuntime(repository=InMemoryAssetRegistryRepository()))
    with TestClient(app) as test_client:
        yield test_client


def test_service_health_readiness_version_and_openapi_contracts(client) -> None:
    health = client.get("/health")
    ready = client.get("/ready")
    version = client.get("/version")
    versioned_health = client.get("/api/v1/health")
    openapi = client.get("/openapi.json")

    assert health.status_code == 200
    assert health.json()["status"] == "ok"
    assert ready.status_code == 200
    assert ready.json()["ready"] is True
    assert ready.json()["checks"]["asset_registry"]["status"] == "ok"
    assert version.status_code == 200
    assert version.json()["api_version"] == "v1"
    assert versioned_health.status_code == 200
    assert versioned_health.json()["api_version"] == "v1"
    assert openapi.status_code == 200
    paths = openapi.json()["paths"]
    assert "/health" in paths
    assert "/ready" in paths
    assert "/version" in paths
    assert "/api/v1/health" in paths
    assert "/api/v1/ready" in paths
    assert "/api/v1/version" in paths
    assert "/api/v1/assets" in paths
    assert "/api/v1/relationships" in paths
    assert "/api/v1/audit/events" in paths
    assert "/api/v1/policy/evaluate" in paths


def test_create_app_attaches_runtime_to_application_state(client) -> None:
    assert client.app.state.kontextual_runtime.api_version == "v1"


class DenyCreatePolicy:
    def authorize(
        self,
        context: OperationContext,
        action: str,
        resource: str,
        *,
        resource_metadata: dict[str, str] | None = None,
    ) -> PolicyDecision:
        if action == "asset.create":
            return PolicyDecision.deny(
                context.actor.id,
                action,
                resource,
                reason="asset creation disabled",
                context={"resource_metadata": resource_metadata or {}},
            )
        return PolicyDecision.allow(context.actor.id, action, resource)