CMIS compliance/test foundation

tests/cmis/opencmis-tck/README.md

@@ -0,0 +1,28 @@
+# Optional OpenCMIS TCK Harness
+
+This directory reserves the integration-test boundary for Apache Chemistry
+OpenCMIS TCK 1.1.0. The harness is intentionally optional because Java/Maven
+tooling is not part of the default Python test suite.
+
+Planned invocation shape:
+
+```sh
+mvn -Dcmis.browser.url=http://127.0.0.1:8000/cmis/compat-tck/browser \
+  -Dcmis.repository.id=kontextual-compat-tck \
+  test
+```
+
+The first target is a selected Browser Binding subset:
+
+- repository information,
+- type definitions,
+- navigation,
+- object reads,
+- content stream reads,
+- query subset,
+- relationships,
+- change log where supported.
+
+Mutation groups should only be enabled for `governed-authoring` or
+`compat-tck` profiles after the policy and audit gates are implemented.
+

tests/cmis/test_cmis_access_profiles.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from kontextual_engine import (
+    Actor,
+    ActorType,
+    CMISAccessPoint,
+    CMISAccessProfile,
+    CMISAction,
+    CMISCapability,
+    Classification,
+    KnowledgeAsset,
+    OperationContext,
+)
+
+
+def _context(actor_type: ActorType = ActorType.HUMAN) -> OperationContext:
+    return OperationContext.create(
+        Actor.create(actor_type, actor_id=f"actor-{actor_type.value}"),
+        correlation_id="corr-cmis-profile",
+    )
+
+
+def _asset(
+    asset_id: str,
+    sensitivity: str,
+    *,
+    asset_type: str = "document",
+    topics: tuple[str, ...] = ("cmis",),
+    metadata: dict | None = None,
+) -> KnowledgeAsset:
+    return KnowledgeAsset.create(
+        f"Asset {asset_id}",
+        Classification(asset_type=asset_type, sensitivity=sensitivity, topics=topics),
+        asset_id=asset_id,
+        metadata=metadata,
+    )
+
+
+def test_readonly_profile_exposes_read_capabilities_and_rejects_mutations() -> None:
+    profile = CMISAccessProfile.readonly_browser()
+    context = _context()
+
+    assert profile.has_capability(CMISCapability.REPOSITORY)
+    assert profile.has_capability(CMISCapability.CONTENT_STREAM_READ)
+    assert profile.decide_action(CMISAction.GET_OBJECT, context).allowed is True
+
+    mutation = profile.decide_action(CMISAction.CREATE_DOCUMENT, context)
+    assert mutation.allowed is False
+    assert mutation.reason == "cmis_mutation_not_allowed"
+
+
+def test_governed_authoring_profile_allows_selected_write_actions() -> None:
+    profile = CMISAccessProfile.governed_authoring()
+    context = _context()
+
+    assert profile.decide_action(CMISAction.CREATE_DOCUMENT, context).allowed is True
+    assert profile.decide_action(CMISAction.SET_CONTENT_STREAM, context).allowed is True
+    assert profile.decide_action(CMISAction.BULK_UPDATE_PROPERTIES, context).allowed is False
+
+
+def test_profiles_hide_denied_sensitivities_without_partial_exposure() -> None:
+    profile = CMISAccessProfile.readonly_browser()
+    context = _context()
+    public = _asset("asset-public", "public")
+    confidential = _asset("asset-confidential", "confidential")
+
+    assert profile.exposes_asset(public, context) is True
+
+    denied = profile.decide_asset_visibility(confidential, context)
+    assert denied.allowed is False
+    assert denied.reason == "cmis_sensitivity_denied"
+    assert denied.context["sensitivity"] == "confidential"
+
+
+def test_profile_visibility_can_be_scoped_by_type_topic_source_and_metadata() -> None:
+    profile = CMISAccessProfile(
+        name="topic-source-scope",
+        capabilities=(CMISCapability.OBJECT_READ,),
+        visible_asset_types=("document",),
+        visible_topics=("approved",),
+        visible_source_systems=("sharepoint",),
+        denied_metadata={"export_blocked": (True,)},
+    )
+    context = _context()
+
+    allowed = _asset(
+        "asset-allowed",
+        "internal",
+        topics=("approved",),
+        metadata={"source_system": "sharepoint"},
+    )
+    wrong_topic = _asset(
+        "asset-wrong-topic",
+        "internal",
+        topics=("draft",),
+        metadata={"source_system": "sharepoint"},
+    )
+    blocked = _asset(
+        "asset-blocked",
+        "internal",
+        topics=("approved",),
+        metadata={"source_system": "sharepoint", "export_blocked": True},
+    )
+
+    assert profile.exposes_asset(allowed, context) is True
+    assert profile.decide_asset_visibility(wrong_topic, context).reason == "cmis_topic_not_visible"
+    assert profile.decide_asset_visibility(blocked, context).reason == "cmis_metadata_denied"
+
+
+def test_admin_export_requires_service_account_actor() -> None:
+    profile = CMISAccessProfile.admin_export()
+    human_context = _context(ActorType.HUMAN)
+    service_context = _context(ActorType.SERVICE_ACCOUNT)
+    confidential = _asset("asset-confidential", "confidential")
+
+    assert profile.decide_action(CMISAction.GET_OBJECT, human_context).allowed is False
+    assert profile.exposes_asset(confidential, human_context) is False
+    assert profile.decide_action(CMISAction.GET_OBJECT, service_context).allowed is True
+    assert profile.exposes_asset(confidential, service_context) is True
+
+
+def test_access_point_normalizes_base_path_and_round_trips() -> None:
+    access_point = CMISAccessPoint(
+        access_point_id="cmis-readonly",
+        repository_id="kontextual-readonly",
+        profile=CMISAccessProfile.readonly_browser(),
+        base_path="cmis/readonly/browser",
+        metadata={"owner": "codex"},
+    )
+
+    serialized = access_point.to_dict()
+    restored = CMISAccessPoint.from_dict(serialized)
+
+    assert access_point.base_path == "/cmis/readonly/browser"
+    assert restored == access_point
+    assert restored.decide_action(CMISAction.GET_REPOSITORY_INFO, _context()).allowed is True
+
+
+def test_disabled_access_point_denies_all_actions_and_visibility() -> None:
+    access_point = CMISAccessPoint(
+        access_point_id="cmis-disabled",
+        repository_id="kontextual-disabled",
+        profile=CMISAccessProfile.governed_authoring(),
+        base_path="/cmis/disabled/browser",
+        enabled=False,
+    )
+    context = _context()
+
+    assert access_point.decide_action(CMISAction.GET_OBJECT, context).reason == "cmis_access_point_disabled"
+    assert access_point.exposes_asset(_asset("asset-public", "public"), context) is False
+

tests/cmis/test_cmis_contract_examples.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+
+FIXTURE_PATH = (
+    Path(__file__).resolve().parents[2] / "examples" / "cmis" / "capability-fixtures.json"
+)
+
+REQUIRED_GROUPS = {
+    "repository-type",
+    "navigation",
+    "object-content",
+    "versioning",
+    "discovery-query",
+    "relationships",
+    "acl-policy",
+    "change-log",
+    "retention-renditions-bulk",
+}
+REQUIRED_PROFILES = {"readonly-browser", "governed-authoring", "admin-export", "compat-tck"}
+
+
+def _catalog() -> dict:
+    return json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
+
+
+def test_cmis_fixture_catalog_declares_target_standard_and_profiles() -> None:
+    catalog = _catalog()
+
+    assert catalog["standard"] == {
+        "name": "CMIS",
+        "version": "1.1",
+        "primary_binding": "browser",
+        "deferred_bindings": ["atompub", "web_services"],
+    }
+    assert set(catalog["profiles"]) == REQUIRED_PROFILES
+    assert all(profile["binding"] == "browser" for profile in catalog["profiles"].values())
+
+
+def test_cmis_fixture_groups_cover_required_capabilities() -> None:
+    catalog = _catalog()
+    groups = {group["id"]: group for group in catalog["capability_groups"]}
+
+    assert set(groups) == REQUIRED_GROUPS
+    for group in groups.values():
+        assert group["examples"], group["id"]
+        assert group["must_validate"], group["id"]
+        assert set(group["supported_profiles"]).issubset(REQUIRED_PROFILES)
+
+
+def test_unsupported_cmis_features_have_structured_diagnostics() -> None:
+    catalog = _catalog()
+    diagnostics = catalog["unsupported_diagnostics"]
+
+    unsupported = {
+        feature
+        for group in catalog["capability_groups"]
+        for feature in group.get("unsupported", [])
+    }
+    unsupported.update(catalog["standard"]["deferred_bindings"])
+
+    assert unsupported <= set(diagnostics)
+    assert all(diagnostics[feature] for feature in unsupported)
+
+
+def test_profile_visibility_and_mutation_expectations_are_explicit() -> None:
+    catalog = _catalog()
+    groups = {group["id"] for group in catalog["capability_groups"]}
+
+    for profile_name, profile in catalog["profiles"].items():
+        expectations = catalog["profile_expectations"][profile_name]
+
+        assert set(expectations["must_expose"]).issubset(groups)
+        assert "must_not_expose_objects" in expectations
+        if profile["mutations"]:
+            assert expectations["must_authorize_actions"] == [
+                "create_document",
+                "update_properties",
+                "delete_object",
+                "set_content_stream",
+            ]
+        else:
+            assert expectations["must_reject_actions"] == [
+                "create_document",
+                "update_properties",
+                "delete_object",
+                "set_content_stream",
+            ]
+
+
+def test_readonly_and_governed_profiles_hide_confidential_fixture_objects() -> None:
+    catalog = _catalog()
+    objects = {item["id"]: item for item in catalog["objects"]}
+    confidential = {
+        object_id
+        for object_id, item in objects.items()
+        if item.get("sensitivity") == "confidential"
+    }
+
+    assert confidential
+    for profile_name in ["readonly-browser", "governed-authoring", "compat-tck"]:
+        denied = set(catalog["profile_expectations"][profile_name]["must_not_expose_objects"])
+        assert confidential <= denied
+
+
+def test_admin_export_is_the_only_profile_for_deferred_extension_group() -> None:
+    catalog = _catalog()
+    extension = next(
+        group
+        for group in catalog["capability_groups"]
+        if group["id"] == "retention-renditions-bulk"
+    )
+
+    assert extension["supported_profiles"] == ["admin-export"]
+    for profile_name, expectations in catalog["profile_expectations"].items():
+        exposes_extension = "retention-renditions-bulk" in expectations["must_expose"]
+        assert exposes_extension is (profile_name == "admin-export")
+