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/content-bytes/{object_id}" in paths assert "/cmis/{access_point_id}/browser/acl/{object_id}" in paths assert "/cmis/{access_point_id}/browser/parents/{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 repository["capabilities"]["capability_get_descendants"] is False assert repository["unsupported_features"]["multifiling"]["status"] == "projection_only" 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", ] def test_cmis_governed_authoring_routes_allow_selected_mutations(cmis_client) -> None: created = cmis_client.post( "/cmis/governed-authoring/browser/document", json={ "asset_id": "asset-api-authored", "name": "API Authored", "content": "# API Authored", "media_type": "text/markdown", }, ) updated = cmis_client.post( "/cmis/governed-authoring/browser/object/cmis:asset:asset-api-authored/properties", json={"properties": {"kontextual:metadata:status": "draft"}}, ) streamed = cmis_client.post( "/cmis/governed-authoring/browser/object/cmis:asset:asset-api-authored/content", json={"content": "# Updated", "media_type": "text/markdown"}, ) byte_stream = cmis_client.get( "/cmis/governed-authoring/browser/content-bytes/cmis:asset:asset-api-authored", ) deleted = cmis_client.post( "/cmis/governed-authoring/browser/object/cmis:asset:asset-api-authored/delete", json={}, ) assert created.status_code == 200 assert updated.json()["properties"]["kontextual:metadata:status"] == "draft" assert streamed.json()["content_stream"]["mime_type"] == "text/markdown" assert byte_stream.content == b"# Updated" assert byte_stream.headers["etag"].startswith("sha256:") assert deleted.json()["lifecycle"] == "delete_requested" def test_cmis_readonly_route_rejects_mutation(cmis_client) -> None: response = cmis_client.post( "/cmis/readonly-browser/browser/document", json={"asset_id": "asset-api-readonly-denied", "name": "Denied"}, ) assert response.status_code == 403 def test_cmis_rejects_unsupported_standard_property_update_with_diagnostics(cmis_client) -> None: response = cmis_client.post( "/cmis/governed-authoring/browser/object/cmis:asset:asset-source/properties", json={"properties": {"cmis:name": "Renamed"}}, ) assert response.status_code == 422 assert response.json()["detail"]["details"]["property"] == "cmis:name" assert response.json()["detail"]["details"]["supported"] == ["kontextual:metadata:"]