kontextual-engine/tests/cmis/test_cmis_browser_binding_api.py

from __future__ import annotations

import pytest

from kontextual_engine import (
    AssetRepresentation,
    Classification,
    RepresentationKind,
    ServiceRuntime,
    Sensitivity,
    create_app,
)
from kontextual_engine.adapters.memory import InMemoryAssetRegistryRepository


pytestmark = pytest.mark.cmis


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

    runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
    context = runtime.operation_context(actor_id="cmis-test", correlation_id="corr-cmis-api")
    runtime.asset_service().create_asset(
        "Source",
        Classification(
            asset_type="document",
            sensitivity=Sensitivity.INTERNAL,
            owner="Platform Knowledge",
            topics=("cmis",),
        ),
        context,
        asset_id="asset-source",
        representations=[
            AssetRepresentation.from_content(
                "asset-source",
                RepresentationKind.SOURCE,
                "text/markdown",
                "# Source\n\nCMIS Browser Binding test fixture.",
                storage_ref="memory://asset-source/source",
            )
        ],
    )
    runtime.create_asset(
        {
            "asset_id": "asset-public",
            "title": "Public Target",
            "classification": {"asset_type": "document", "sensitivity": "public"},
        },
        context,
    )
    runtime.create_asset(
        {
            "asset_id": "asset-confidential",
            "title": "Confidential Target",
            "classification": {"asset_type": "document", "sensitivity": "confidential"},
        },
        context,
    )
    runtime.create_relationship(
        {
            "source_asset_id": "asset-source",
            "target_id": "asset-public",
            "predicate": "references",
            "target_kind": "asset",
            "confidence": 1.0,
        },
        context,
    )
    with TestClient(create_app(runtime)) as test_client:
        yield test_client


def test_cmis_browser_binding_routes_are_advertised_in_openapi(cmis_client) -> None:
    paths = cmis_client.get("/openapi.json").json()["paths"]

    assert "/cmis" in paths
    assert "/cmis/{access_point_id}/browser" in paths
    assert "/cmis/{access_point_id}/browser/types" in paths
    assert "/cmis/{access_point_id}/browser/children" in paths
    assert "/cmis/{access_point_id}/browser/object/{object_id}" in paths
    assert "/cmis/{access_point_id}/browser/content/{object_id}" in paths
    assert "/cmis/{access_point_id}/browser/query" in paths
    assert "/cmis/{access_point_id}/browser/relationships" in paths
    assert "/cmis/{access_point_id}/browser/changes" in paths


def test_cmis_repository_info_and_type_definitions(cmis_client) -> None:
    access_points = cmis_client.get("/cmis").json()
    repository = cmis_client.get("/cmis/readonly-browser/browser").json()
    types = cmis_client.get("/cmis/readonly-browser/browser/types").json()

    assert access_points["count"] == 4
    assert repository["repository_id"] == "kontextual-readonly-browser"
    assert repository["cmis_version_supported"] == "1.1"
    assert repository["capabilities"]["capability_query"] == "metadataonly"
    assert {item["base_type_id"] for item in types["items"]} >= {
        "cmis:document",
        "cmis:folder",
        "cmis:relationship",
    }


def test_cmis_readonly_children_object_content_query_relationships_and_changes(cmis_client) -> None:
    children = cmis_client.get("/cmis/readonly-browser/browser/children").json()
    object_response = cmis_client.get(
        "/cmis/readonly-browser/browser/object/cmis:asset:asset-source"
    ).json()
    content = cmis_client.get(
        "/cmis/readonly-browser/browser/content/cmis:asset:asset-source"
    ).json()
    query = cmis_client.get(
        "/cmis/readonly-browser/browser/query",
        params={"q": "SELECT * FROM cmis:document"},
    ).json()
    relationships = cmis_client.get(
        "/cmis/readonly-browser/browser/relationships",
        params={"object_id": "cmis:asset:asset-source"},
    ).json()
    changes = cmis_client.get("/cmis/readonly-browser/browser/changes").json()

    child_ids = {item["object_id"] for item in children["objects"]}
    assert "cmis:asset:asset-source" in child_ids
    assert "cmis:asset:asset-public" in child_ids
    assert "cmis:asset:asset-confidential" not in child_ids
    assert object_response["properties"]["kontextual:assetId"] == "asset-source"
    assert "get_content_stream" in object_response["allowable_actions"]
    assert content["mime_type"] == "text/markdown"
    assert query["total_num_items"] == children["total_num_items"]
    assert relationships["count"] == 1
    assert relationships["items"][0]["properties"]["cmis:targetId"] == "cmis:asset:asset-public"
    assert changes["total_num_items"] >= 3


def test_cmis_profile_gates_visibility_by_access_point(cmis_client) -> None:
    readonly = cmis_client.get("/cmis/readonly-browser/browser/children").json()
    admin_denied = cmis_client.get("/cmis/admin-export/browser/children")
    admin_allowed = cmis_client.get(
        "/cmis/admin-export/browser/children",
        headers={"X-Actor-Type": "service_account", "X-Actor-Id": "svc-export"},
    ).json()

    readonly_ids = {item["object_id"] for item in readonly["objects"]}
    admin_ids = {item["object_id"] for item in admin_allowed["objects"]}

    assert admin_denied.status_code == 403
    assert "cmis:asset:asset-confidential" not in readonly_ids
    assert "cmis:asset:asset-confidential" in admin_ids


def test_cmis_query_reports_unsupported_subset_diagnostics(cmis_client) -> None:
    response = cmis_client.get(
        "/cmis/readonly-browser/browser/query",
        params={"q": "SELECT * FROM cmis:document JOIN cmis:relationship"},
    )

    assert response.status_code == 422
    assert response.json()["detail"]["details"]["supported"] == [
        "SELECT * FROM cmis:document",
        "SELECT * FROM kontextual:document",
    ]