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/root" 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 assert "/cmis/{access_point_id}/browser/folder" in paths def test_cmis_repository_info_and_type_definitions(cmis_client) -> None: access_points = cmis_client.get("/cmis").json() service_document = cmis_client.get("/cmis/readonly-browser/browser").json() repository = service_document["kontextual-readonly-browser"] types = cmis_client.get("/cmis/readonly-browser/browser/types").json() browser_types = cmis_client.get( "/cmis/readonly-browser/browser", params={"cmisselector": "typeChildren"}, ).json() browser_types_with_properties = cmis_client.get( "/cmis/readonly-browser/browser", params={"cmisselector": "typeChildren", "includePropertyDefinitions": "true"}, ).json() browser_type_descendants = cmis_client.get( "/cmis/readonly-browser/browser", params={"cmisselector": "typeDescendants"}, ).json() browser_type_definition = cmis_client.get( "/cmis/readonly-browser/browser", params={"cmisselector": "typeDefinition", "typeId": "cmis:document"}, ).json() root_policies = cmis_client.get( "/cmis/readonly-browser/browser/root", params={"cmisselector": "policies"}, ).json() root_object = cmis_client.get( "/cmis/readonly-browser/browser/root", params={"cmisselector": "object"}, ).json() assert access_points["count"] == 4 assert repository["repositoryId"] == "kontextual-readonly-browser" assert repository["cmisVersionSupported"] == "1.1" assert repository["repositoryUrl"].endswith("/cmis/readonly-browser/browser") assert repository["rootFolderUrl"].endswith("/cmis/readonly-browser/browser/root") assert repository["capabilities"]["capabilityQuery"] == "metadataonly" assert repository["capabilities"]["capabilityGetDescendants"] is False assert browser_types["types"][0]["id"] == "cmis:document" assert "propertyDefinitions" not in browser_types["types"][0] assert "propertyDefinitions" in browser_types_with_properties["types"][0] assert browser_type_descendants[0]["type"]["id"] == "cmis:document" assert "propertyDefinitions" not in browser_type_descendants[0]["type"] assert "propertyDefinitions" in browser_type_definition assert browser_type_descendants[0]["children"] == [] assert root_policies == [] assert root_object["properties"]["kontextual:filingSource"]["value"] == "root" assert root_object["properties"]["kontextual:workspaceFolder"]["value"] is False assert "kontextual:assetId" in browser_type_definition["propertyDefinitions"] assert browser_type_definition["propertyDefinitions"]["kontextual:topics"]["cardinality"] == "multi" assert "cmis:path" in browser_type_definition["propertyDefinitions"] 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: root_children = cmis_client.get("/cmis/readonly-browser/browser/children").json() children = cmis_client.get( "/cmis/readonly-browser/browser/children", params={"folder_id": "cmis:folder:assets::document"}, ).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() root_ids = {item["object_id"] for item in root_children["objects"]} child_ids = {item["object_id"] for item in children["objects"]} assert "cmis:folder:assets" in root_ids 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", params={"folder_id": "cmis:folder:assets::document"}, ).json() admin_denied = cmis_client.get("/cmis/admin-export/browser/children") admin_allowed = cmis_client.get( "/cmis/admin-export/browser/children", params={"folder_id": "cmis:folder:assets::document"}, 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_browser_binding_create_folder_action_creates_workspace_folder(cmis_client) -> None: created = cmis_client.post( "/cmis/compat-tck/browser/root", data={ "cmisaction": "createFolder", "propertyId[0]": "cmis:objectTypeId", "propertyValue[0]": "cmis:folder", "propertyId[1]": "cmis:name", "propertyValue[1]": "Action Workspace", }, ) folder = created.json() folder_id = folder["properties"]["cmis:objectId"]["value"] root_children = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "children"}, ).json() fetched = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "object", "objectId": folder_id}, ).json() parent = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "parent", "objectId": folder_id}, ).json() document = cmis_client.post( "/cmis/compat-tck/browser/root", params={"objectId": folder_id}, data={ "cmisaction": "createDocument", "propertyId[0]": "cmis:objectTypeId", "propertyValue[0]": "cmis:document", "propertyId[1]": "cmis:name", "propertyValue[1]": "Multipart Document", }, files={"content": ("multipart.txt", b"Multipart content", "text/plain")}, ) document_path = document.json()["properties"]["cmis:path"]["value"] fetched_document_by_path = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "object", "path": document_path}, ).json() document_parents = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "parents", "objectId": document.json()["properties"]["cmis:objectId"]["value"]}, ).json() fetched_folder_by_path = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "object", "path": "/Action Workspace"}, ).json() deleted_tree = cmis_client.post( "/cmis/compat-tck/browser/root", data={"cmisaction": "deleteTree", "objectId": folder_id}, ) root_children_after_delete = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "children"}, ).json() fetched_after_delete = cmis_client.get( "/cmis/compat-tck/browser/root", params={"cmisselector": "object", "objectId": folder_id}, ) deleted = cmis_client.post( "/cmis/compat-tck/browser", data={"cmisaction": "delete", "objectId": folder_id}, ) assert created.status_code == 200 assert folder["properties"]["cmis:name"]["value"] == "Action Workspace" assert folder["properties"]["kontextual:workspaceFolder"]["value"] is True assert folder["allowableActions"]["canDeleteTree"] is True assert any(item["object"]["properties"]["cmis:objectId"]["value"] == folder_id for item in root_children["objects"]) assert fetched["properties"]["cmis:path"]["value"] == "/Action Workspace" assert parent["properties"]["cmis:objectId"]["value"] == "cmis-root" assert document.status_code == 200 assert document.json()["properties"]["cmis:name"]["value"] == "Multipart Document" assert document.json()["properties"]["cmis:contentStreamLength"]["value"] == 17 assert document.json()["properties"]["cmis:isLatestVersion"]["value"] is True assert document.json()["properties"]["cmis:secondaryObjectTypeIds"]["value"] == [] assert document.json()["allowableActions"]["canGetFolderParent"] is False assert document_path == "/Action Workspace/Multipart Document" assert fetched_document_by_path["properties"]["cmis:name"]["value"] == "Multipart Document" assert document_parents[0]["object"]["properties"]["cmis:path"]["value"] == "/Action Workspace" assert fetched_folder_by_path["properties"]["cmis:objectId"]["value"] == folder_id assert deleted_tree.json()["failedToDelete"] == [] assert all( item["object"]["properties"]["cmis:objectId"]["value"] != folder_id for item in root_children_after_delete["objects"] ) assert fetched_after_delete.status_code == 404 assert deleted.status_code == 404 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"}, ) folder_response = cmis_client.post( "/cmis/readonly-browser/browser/root", data={ "cmisaction": "createFolder", "propertyId[0]": "cmis:name", "propertyValue[0]": "Denied Folder", }, ) assert response.status_code == 403 assert folder_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:"]