generated from coulomb/repo-seed
346 lines
14 KiB
Python
346 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from kontextual_engine import (
|
|
AssetRepresentation,
|
|
Classification,
|
|
RepresentationKind,
|
|
ServiceRuntime,
|
|
Sensitivity,
|
|
)
|
|
from kontextual_engine.adapters.memory import InMemoryAssetRegistryRepository
|
|
|
|
|
|
pytestmark = pytest.mark.cmis
|
|
|
|
|
|
@pytest.fixture
|
|
def cmis_runtime() -> tuple[ServiceRuntime, object]:
|
|
runtime = ServiceRuntime(repository=InMemoryAssetRegistryRepository())
|
|
context = runtime.operation_context(actor_id="cmis-runtime", correlation_id="corr-cmis-runtime")
|
|
runtime.asset_service().create_asset(
|
|
"Runtime Source",
|
|
Classification(
|
|
asset_type="document",
|
|
sensitivity=Sensitivity.INTERNAL,
|
|
owner="Platform Knowledge",
|
|
topics=("cmis", "integration"),
|
|
),
|
|
context,
|
|
asset_id="asset-runtime-source",
|
|
representations=[
|
|
AssetRepresentation.from_content(
|
|
"asset-runtime-source",
|
|
RepresentationKind.SOURCE,
|
|
"text/markdown",
|
|
"# Runtime Source\n\nCMIS runtime fixture.",
|
|
storage_ref="memory://asset-runtime-source/source",
|
|
)
|
|
],
|
|
)
|
|
runtime.create_asset(
|
|
{
|
|
"asset_id": "asset-runtime-public",
|
|
"title": "Runtime Public",
|
|
"classification": {"asset_type": "document", "sensitivity": "public"},
|
|
},
|
|
context,
|
|
)
|
|
runtime.create_asset(
|
|
{
|
|
"asset_id": "asset-runtime-confidential",
|
|
"title": "Runtime Confidential",
|
|
"classification": {"asset_type": "document", "sensitivity": "confidential"},
|
|
},
|
|
context,
|
|
)
|
|
runtime.create_relationship(
|
|
{
|
|
"source_asset_id": "asset-runtime-source",
|
|
"target_id": "asset-runtime-public",
|
|
"predicate": "references",
|
|
"target_kind": "asset",
|
|
"confidence": 0.99,
|
|
},
|
|
context,
|
|
)
|
|
runtime.create_relationship(
|
|
{
|
|
"source_asset_id": "asset-runtime-source",
|
|
"target_id": "asset-runtime-confidential",
|
|
"predicate": "mentions_sensitive",
|
|
"target_kind": "asset",
|
|
"confidence": 0.5,
|
|
},
|
|
context,
|
|
)
|
|
return runtime, context
|
|
|
|
|
|
def test_runtime_cmis_browser_repository_types_children_and_object(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
access_points = runtime.cmis_access_points()
|
|
repository = runtime.cmis_repository_info("readonly-browser")
|
|
types = runtime.cmis_type_definitions("readonly-browser")
|
|
children = runtime.cmis_children("readonly-browser", context)
|
|
topic_children = runtime.cmis_children("readonly-browser", context, folder_id="/topics/cmis")
|
|
obj = runtime.cmis_object("readonly-browser", "cmis:asset:asset-runtime-source", context)
|
|
parents = runtime.cmis_object_parents("readonly-browser", "cmis:asset:asset-runtime-source", context)
|
|
|
|
assert access_points["count"] == 4
|
|
assert repository["repository_id"] == "kontextual-readonly-browser"
|
|
assert repository["capabilities"]["capability_get_descendants"] is False
|
|
assert repository["capabilities"]["capability_get_folder_tree"] is False
|
|
assert repository["unsupported_features"]["get_descendants"]["status"] == "unsupported"
|
|
assert {item["base_type_id"] for item in types["items"]} >= {"cmis:document", "cmis:folder"}
|
|
root_paths = {item["path"] for item in children["objects"]}
|
|
topic_object_ids = {item["object_id"] for item in topic_children["objects"]}
|
|
parent_paths = {item["path"] for item in parents["parents"]}
|
|
assert "/topics" in root_paths
|
|
assert "cmis:asset:asset-runtime-source" in topic_object_ids
|
|
assert "cmis:asset:asset-runtime-confidential" not in topic_object_ids
|
|
assert {"/topics/cmis", "/topics/integration"} <= parent_paths
|
|
assert obj["properties"]["kontextual:assetId"] == "asset-runtime-source"
|
|
|
|
|
|
def test_runtime_cmis_browser_content_query_relationships_and_changes(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
content = runtime.cmis_content_stream("readonly-browser", "cmis:asset:asset-runtime-source", context)
|
|
query = runtime.cmis_query("readonly-browser", "SELECT * FROM cmis:document", context)
|
|
relationships = runtime.cmis_relationships(
|
|
"readonly-browser",
|
|
context,
|
|
object_id="cmis:asset:asset-runtime-source",
|
|
)
|
|
changes = runtime.cmis_change_log("readonly-browser", context)
|
|
|
|
assert content["mime_type"] in {"text/plain", "text/markdown"}
|
|
assert query["total_num_items"] == 2
|
|
assert relationships["count"] == 1
|
|
assert relationships["items"][0]["properties"]["cmis:targetId"] == "cmis:asset:asset-runtime-public"
|
|
assert changes["total_num_items"] >= 3
|
|
assert all(change["object_id"] != "cmis:asset:asset-runtime-confidential" for change in changes["changes"])
|
|
|
|
|
|
def test_runtime_cmis_browser_rejects_unsupported_query_subset(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_query(
|
|
"readonly-browser",
|
|
"SELECT * FROM cmis:document JOIN cmis:relationship",
|
|
context,
|
|
)
|
|
|
|
assert "Unsupported CMIS query subset" in str(exc_info.value)
|
|
|
|
|
|
def test_runtime_cmis_governed_authoring_allows_selected_mutations(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
created = runtime.cmis_create_document(
|
|
"governed-authoring",
|
|
{
|
|
"asset_id": "asset-authored",
|
|
"name": "Authored Through CMIS",
|
|
"sensitivity": "internal",
|
|
"topics": ["cmis"],
|
|
"content": "# Authored\n\nCreated through CMIS.",
|
|
"media_type": "text/markdown",
|
|
"metadata_records": [{"key": "status", "value": "draft", "confirmed": True}],
|
|
},
|
|
context,
|
|
)
|
|
updated = runtime.cmis_update_properties(
|
|
"governed-authoring",
|
|
"cmis:asset:asset-authored",
|
|
{"properties": {"kontextual:metadata:reviewer": "codex"}},
|
|
context,
|
|
)
|
|
streamed = runtime.cmis_set_content_stream(
|
|
"governed-authoring",
|
|
"cmis:asset:asset-authored",
|
|
{"content": "# Authored\n\nUpdated stream.", "media_type": "text/markdown"},
|
|
context,
|
|
)
|
|
stream_bytes = runtime.cmis_content_stream_bytes(
|
|
"governed-authoring",
|
|
"cmis:asset:asset-authored",
|
|
context,
|
|
)
|
|
deleted = runtime.cmis_delete_object(
|
|
"governed-authoring",
|
|
"cmis:asset:asset-authored",
|
|
{},
|
|
context,
|
|
)
|
|
|
|
assert created["object_id"] == "cmis:asset:asset-authored"
|
|
assert updated["properties"]["kontextual:metadata:reviewer"] == "codex"
|
|
assert streamed["content_stream"]["mime_type"] == "text/markdown"
|
|
assert b"".join(stream_bytes.chunks) == b"# Authored\n\nUpdated stream."
|
|
assert stream_bytes.representation.storage_ref.startswith("blob://memory/")
|
|
assert deleted["deleted"] is False
|
|
assert deleted["lifecycle"] == "delete_requested"
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_object("governed-authoring", "cmis:asset:asset-authored", context)
|
|
assert "CMIS object not found" in str(exc_info.value)
|
|
|
|
|
|
def test_runtime_cmis_compat_profile_supports_workspace_folder_lifecycle(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
folder = runtime.cmis_create_folder(
|
|
"compat-tck",
|
|
{"name": "TCK Workspace", "properties": {"cmis:objectTypeId": "cmis:folder"}},
|
|
context,
|
|
)
|
|
folder_object_id = folder["object_id"]
|
|
root_children = runtime.cmis_children("compat-tck", context)
|
|
fetched = runtime.cmis_object("compat-tck", folder_object_id, context)
|
|
parents = runtime.cmis_object_parents("compat-tck", folder_object_id, context)
|
|
document = runtime.cmis_create_document(
|
|
"compat-tck",
|
|
{
|
|
"name": "Workspace Document",
|
|
"folder_id": folder_object_id,
|
|
"content": "Workspace content",
|
|
"media_type": "text/plain",
|
|
},
|
|
context,
|
|
)
|
|
folder_children = runtime.cmis_children("compat-tck", context, folder_id=folder_object_id)
|
|
document_by_path = runtime.cmis_object_by_path("compat-tck", "/TCK Workspace/Workspace Document", context)
|
|
document_parents = runtime.cmis_object_parents("compat-tck", document["object_id"], context)
|
|
browser_document_parents = runtime.cmis_browser_parents("compat-tck", document["object_id"], context)
|
|
filtered_document = runtime.cmis_browser_object(
|
|
"compat-tck",
|
|
document["object_id"],
|
|
context,
|
|
property_filter="cmis:objectId,cmis:name",
|
|
include_allowable_actions=False,
|
|
include_acl=False,
|
|
)
|
|
filtered_children = runtime.cmis_browser_children(
|
|
"compat-tck",
|
|
context,
|
|
object_id=folder_object_id,
|
|
property_filter="cmis:objectId,cmis:name",
|
|
include_allowable_actions=False,
|
|
include_acl=False,
|
|
include_path_segment=False,
|
|
)
|
|
|
|
assert folder["path"] == "/TCK Workspace"
|
|
assert folder["properties"]["kontextual:workspaceFolder"] is True
|
|
assert folder_object_id in {item["object_id"] for item in root_children["objects"]}
|
|
assert fetched["properties"]["cmis:path"] == "/TCK Workspace"
|
|
assert parents["parents"][0]["object_id"] == "cmis-root"
|
|
assert document["path"] == "/TCK Workspace/Workspace Document"
|
|
assert "cmis:path" not in document["properties"]
|
|
assert document_by_path["object_id"] == document["object_id"]
|
|
assert document_parents["count"] == 1
|
|
assert document_parents["parents"][0]["properties"]["cmis:path"] == "/TCK Workspace"
|
|
assert browser_document_parents[0]["relativePathSegment"] == "Workspace Document"
|
|
assert document["object_id"] in {item["object_id"] for item in folder_children["objects"]}
|
|
assert set(filtered_document["properties"]) == {"cmis:objectId", "cmis:name"}
|
|
assert "allowableActions" not in filtered_document
|
|
assert "pathSegment" not in filtered_children["objects"][0]
|
|
assert set(filtered_children["objects"][0]["object"]["properties"]) == {"cmis:objectId", "cmis:name"}
|
|
assert "allowableActions" not in filtered_children["objects"][0]["object"]
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_delete_object("compat-tck", folder_object_id, {}, context)
|
|
assert "CMIS folder is not empty" in str(exc_info.value)
|
|
|
|
deleted_tree = runtime.cmis_delete_tree("compat-tck", folder_object_id, {}, context)
|
|
root_children_after_delete = runtime.cmis_children("compat-tck", context)
|
|
|
|
assert deleted_tree["failedToDelete"] == []
|
|
assert folder_object_id not in {item["object_id"] for item in root_children_after_delete["objects"]}
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_object("compat-tck", folder_object_id, context)
|
|
assert "CMIS folder not found" in str(exc_info.value)
|
|
|
|
|
|
def test_runtime_cmis_workspace_folder_rename_keeps_object_id_stable(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
folder = runtime.cmis_create_folder(
|
|
"compat-tck",
|
|
{"name": "Rename Source", "properties": {"cmis:objectTypeId": "cmis:folder"}},
|
|
context,
|
|
)
|
|
renamed = runtime.cmis_update_properties(
|
|
"compat-tck",
|
|
folder["object_id"],
|
|
{"properties": {"cmis:name": "Rename Target"}},
|
|
context,
|
|
)
|
|
fetched_by_old_id = runtime.cmis_object("compat-tck", folder["object_id"], context)
|
|
fetched_by_new_path = runtime.cmis_object_by_path("compat-tck", "/Rename Target", context)
|
|
deleted = runtime.cmis_delete_object("compat-tck", folder["object_id"], {}, context)
|
|
|
|
assert renamed["object_id"] == folder["object_id"]
|
|
assert renamed["path"] == "/Rename Target"
|
|
assert fetched_by_old_id["properties"]["cmis:name"] == "Rename Target"
|
|
assert fetched_by_new_path["object_id"] == folder["object_id"]
|
|
assert deleted["deleted"] is True
|
|
|
|
|
|
def test_runtime_cmis_rejects_unsupported_standard_property_updates(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_update_properties(
|
|
"governed-authoring",
|
|
"cmis:asset:asset-runtime-source",
|
|
{"properties": {"cmis:objectTypeId": "cmis:folder"}},
|
|
context,
|
|
)
|
|
|
|
assert "Unsupported CMIS property update" in str(exc_info.value)
|
|
|
|
|
|
def test_runtime_cmis_readonly_profile_rejects_mutations(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_create_document(
|
|
"readonly-browser",
|
|
{"asset_id": "asset-readonly-denied", "name": "Denied"},
|
|
context,
|
|
)
|
|
|
|
assert "CMIS operation denied" in str(exc_info.value)
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_create_folder(
|
|
"readonly-browser",
|
|
{"name": "Denied Folder"},
|
|
context,
|
|
)
|
|
|
|
assert "CMIS operation denied" in str(exc_info.value)
|
|
|
|
|
|
def test_runtime_cmis_acl_projection_and_redaction(cmis_runtime) -> None:
|
|
runtime, context = cmis_runtime
|
|
|
|
public_acl = runtime.cmis_acl("readonly-browser", "cmis:asset:asset-runtime-public", context)
|
|
internal_acl = runtime.cmis_acl("governed-authoring", "cmis:asset:asset-runtime-source", context)
|
|
|
|
assert public_acl["is_exact"] is True
|
|
assert {entry["principal_id"] for entry in public_acl["aces"]} == {"cmis-runtime", "anyone"}
|
|
assert ["cmis:read", "cmis:write", "cmis:delete"] in [
|
|
entry["permissions"] for entry in internal_acl["aces"]
|
|
]
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
runtime.cmis_acl("readonly-browser", "cmis:asset:asset-runtime-confidential", context)
|
|
|
|
assert "CMIS object not found" in str(exc_info.value)
|