kontextual-engine/tests/cmis/test_cmis_compliance_flags.py

from __future__ import annotations

from kontextual_engine import (
    Actor,
    ActorType,
    CMISAccessPoint,
    CMISAccessProfile,
    CMISAction,
    CMISBaseType,
    CMISCapability,
    CMISDomainMapper,
    Classification,
    KnowledgeAsset,
    LifecycleState,
    OperationContext,
)


EXPECTED_CAPABILITY_KEYS = {
    "capability_content_stream_updatability",
    "capability_changes",
    "capability_renditions",
    "capability_get_descendants",
    "capability_get_folder_tree",
    "capability_order_by",
    "capability_multifiling",
    "capability_unfiling",
    "capability_version_specific_filing",
    "capability_pwc_searchable",
    "capability_pwc_updatable",
    "capability_all_versions_searchable",
    "capability_query",
    "capability_join",
    "capability_acl",
    "capability_new_type_settable_attributes",
}


def _context() -> OperationContext:
    return OperationContext.create(
        Actor.create(ActorType.HUMAN, actor_id="actor-cmis-compliance"),
        correlation_id="corr-cmis-compliance",
    )


def _mapper(profile: CMISAccessProfile | None = None) -> CMISDomainMapper:
    return CMISDomainMapper(
        CMISAccessPoint(
            access_point_id="cmis-compliance",
            repository_id="kontextual-compliance",
            profile=profile or CMISAccessProfile.readonly_browser(),
            base_path="/cmis/compliance/browser",
            metadata={"repository_name": "Kontextual Compliance Repository"},
        )
    )


def test_repository_info_uses_complete_conservative_cmis_11_capability_flags() -> None:
    repository = _mapper().repository_info()
    capabilities = repository["capabilities"]

    assert EXPECTED_CAPABILITY_KEYS <= set(capabilities)
    assert capabilities["capability_content_stream_updatability"] == "none"
    assert capabilities["capability_changes"] == "objectidsonly"
    assert capabilities["capability_renditions"] == "none"
    assert capabilities["capability_get_descendants"] is False
    assert capabilities["capability_get_folder_tree"] is False
    assert capabilities["capability_order_by"] == "common"
    assert capabilities["capability_multifiling"] is False
    assert capabilities["capability_unfiling"] is False
    assert capabilities["capability_version_specific_filing"] is False
    assert capabilities["capability_pwc_searchable"] is False
    assert capabilities["capability_pwc_updatable"] is False
    assert capabilities["capability_all_versions_searchable"] is False
    assert capabilities["capability_query"] == "metadataonly"
    assert capabilities["capability_join"] == "none"
    assert capabilities["capability_acl"] == "discover"
    assert set(capabilities["capability_new_type_settable_attributes"].values()) == {False}


def test_authoring_profile_only_advertises_supported_content_write_semantics() -> None:
    repository = _mapper(CMISAccessProfile.governed_authoring()).repository_info()
    capabilities = repository["capabilities"]

    assert capabilities["capability_content_stream_updatability"] == "anytime"
    assert capabilities["capability_multifiling"] is False
    assert capabilities["capability_renditions"] == "none"


def test_repository_info_exposes_nonstandard_projection_feature_and_unsupported_catalog() -> None:
    repository = _mapper().repository_info()
    feature_ids = {feature["id"] for feature in repository["repository_features"]}
    unsupported = repository["unsupported_features"]

    assert "urn:kontextual:cmis:feature:projection-parentage" in feature_ids
    assert unsupported["multifiling"]["status"] == "projection_only"
    assert unsupported["multifiling"]["standard_flag"] == "capability_multifiling"
    assert unsupported["get_descendants"]["standard_flag"] == "capability_get_descendants"
    assert unsupported["rendition_streams"]["standard_flag"] == "capability_renditions"
    assert repository["compliance"]["posture"] == "declared-browser-binding-subset"


def test_type_definitions_do_not_claim_unimplemented_cmis_services() -> None:
    types = {definition["base_type_id"]: definition for definition in _mapper().type_definitions()}
    document = types[CMISBaseType.DOCUMENT.value]
    folder = types[CMISBaseType.FOLDER.value]
    relationship = types[CMISBaseType.RELATIONSHIP.value]
    policy = types[CMISBaseType.POLICY.value]

    assert document["versionable"] is False
    assert document["controllable_acl"] is True
    assert document["queryable"] is True
    assert document["creatable"] is False
    assert "cmis:contentStreamLength" in document["property_definitions"]
    assert folder["controllable_acl"] is False
    assert folder["queryable"] is False
    assert relationship["creatable"] is False
    assert relationship["queryable"] is False
    assert policy["controllable_policy"] is False


def test_profile_denies_unimplemented_mutations_even_when_related_capability_exists() -> None:
    profile = CMISAccessProfile(
        name="acl-manager-shape",
        capabilities=(CMISCapability.ACL,),
        allow_mutations=True,
    )
    decision = profile.decide_action(CMISAction.APPLY_ACL, _context())

    assert decision.allowed is False
    assert decision.reason == "cmis_operation_not_implemented"


def test_deleted_assets_are_not_exposed_as_live_cmis_objects() -> None:
    profile = CMISAccessProfile.readonly_browser()
    asset = KnowledgeAsset.create(
        "Deleted",
        Classification(asset_type="document", sensitivity="internal"),
        asset_id="asset-deleted",
    ).transition_lifecycle(LifecycleState.DELETE_REQUESTED)

    decision = profile.decide_asset_visibility(asset, _context())

    assert decision.allowed is False
    assert decision.reason == "cmis_lifecycle_not_visible"