diff --git a/tests/contract/test_astrolabe_capabilities_consumer.py b/tests/contract/test_astrolabe_capabilities_consumer.py new file mode 100644 index 00000000..8bea1176 --- /dev/null +++ b/tests/contract/test_astrolabe_capabilities_consumer.py @@ -0,0 +1,108 @@ +"""Consumer contract: nextcloud-mcp-server -> astrolabe OCS capabilities. + +The MCP server reads which content sources an admin has approved for semantic +search via :func:`nextcloud_mcp_server.capabilities.allowed_doc_types`, which +calls :meth:`NextcloudClient.capabilities` → +``GET /ocs/v2.php/cloud/capabilities`` and parses +``ocs.data.capabilities.astrolabe.semantic_search.enabled_doc_types`` (the +``OCA\\Astrolabe\\Capabilities`` provider on the astrolabe side). + +This pact pins the request shape and the two states the consumer branches on: + +- some sources approved -> ``enabled_doc_types`` is a non-empty list, parsed to + the corresponding frozenset (the search/scan/purge gates restrict to it) +- every source disabled -> ``enabled_doc_types`` is ``[]``, parsed to an empty + frozenset (an all-disabled admin config — distinct from "no restriction") + +The "no astrolabe block" / fail-open (``None``) case is internal defensive +handling for older astrolabe versions, not a contract obligation of the current +provider, so it is covered by a unit test (``tests/unit/test_capabilities.py``) +rather than a pact interaction. + +Only the capability block astrolabe owns is pinned — the real OCS response also +carries many other apps' capabilities, which pact ignores (unspecified keys are +allowed in the provider response). + +See ADR-029 for the overall contract-testing architecture. +""" + +import pytest +from httpx import BasicAuth + +from nextcloud_mcp_server.capabilities import allowed_doc_types, clear_cache +from nextcloud_mcp_server.client import NextcloudClient + +pytestmark = pytest.mark.contract + + +def _ocs_capabilities(enabled_doc_types: list[str]) -> dict: + """Minimal OCS envelope carrying just astrolabe's semantic_search block.""" + return { + "ocs": { + "meta": {"status": "ok"}, + "data": { + "capabilities": { + "astrolabe": { + "semantic_search": { + "enabled_doc_types": enabled_doc_types, + } + } + } + }, + } + } + + +async def test_capabilities_report_admin_approved_doc_types(consumer_pact): + """Approved sources are returned and parsed into the allow-set.""" + clear_cache() + ( + consumer_pact.upon_receiving("a request for OCS capabilities") + .given("astrolabe has approved file and note for semantic search") + .with_request("GET", "/ocs/v2.php/cloud/capabilities") + .with_header("OCS-APIRequest", "true") + .will_respond_with(200) + .with_body( + _ocs_capabilities(["file", "note"]), + content_type="application/json", + ) + ) + + with consumer_pact.serve() as srv: + client = NextcloudClient( + base_url=str(srv.url), + username="admin", + auth=BasicAuth("admin", "app-password"), + ) + allowed = await allowed_doc_types(client, "admin") + + assert allowed == frozenset({"file", "note"}) + + +async def test_capabilities_report_all_sources_disabled(consumer_pact): + """An empty allow-set (admin disabled everything) is distinct from None.""" + clear_cache() + ( + consumer_pact.upon_receiving( + "a request for OCS capabilities with every source disabled" + ) + .given("astrolabe has disabled all sources for semantic search") + .with_request("GET", "/ocs/v2.php/cloud/capabilities") + .with_header("OCS-APIRequest", "true") + .will_respond_with(200) + .with_body( + _ocs_capabilities([]), + content_type="application/json", + ) + ) + + with consumer_pact.serve() as srv: + client = NextcloudClient( + base_url=str(srv.url), + username="admin", + auth=BasicAuth("admin", "app-password"), + ) + allowed = await allowed_doc_types(client, "admin") + + # Present-but-empty => empty frozenset (restrict everything), NOT None. + assert allowed == frozenset() diff --git a/tests/contract/test_mcp_provider_verification.py b/tests/contract/test_mcp_provider_verification.py index db5754ca..5b35123c 100644 --- a/tests/contract/test_mcp_provider_verification.py +++ b/tests/contract/test_mcp_provider_verification.py @@ -70,7 +70,22 @@ pytestmark = [ # fixtures, etc.). Keep the keys identical to the astrolabe ``given(...)`` # strings. Unhandled states fall through to ``_dispatch_state`` which logs and # no-ops, so state-less interactions still verify. +def _state_admin_can_purge() -> None: + """Provider state for astrolabe's consent-purge pact + (``POST /api/v1/vector-sync/purge``). + + Full verification of this authenticated endpoint (admin OAuth token + + Nextcloud admin-group check + Qdrant delete) needs the live-stack auth + test-hook that is the ADR-029 phase-4 follow-up. Until then the interaction + rides the broker's pending flow (see ``include_pending`` below); this handler + is registered so the dispatcher recognises the state by name rather than + logging an "unhandled state" warning. + """ + return None + + _PROVIDER_STATES: dict[str, Callable[[], None]] = { + "an admin can purge indexed documents": _state_admin_can_purge, # "a webhook is registered for user alice": _state_webhook_registered, # "vector sync has indexed documents": _state_vector_sync_ran, # "the search index returns a hit for 'budget'": _state_search_has_hit, @@ -103,9 +118,23 @@ def test_verify_astrolabe_consumer_pacts() -> None: verifier.state_handler(_dispatch_state, teardown=True) if _BROKER_URL and _BROKER_USERNAME and _BROKER_PASSWORD: - verifier.broker_source( - _BROKER_URL, username=_BROKER_USERNAME, password=_BROKER_PASSWORD + # selector=True to opt into pending pacts: a new/authenticated contract + # (e.g. the consent-purge endpoint) reports as *pending* instead of + # failing this build until provider verification of the authenticated + # surface is stood up (ADR-029 phase 4). Already-verified interactions + # (GET /api/v1/status) stay blocking. Empty consumer selectors keep the + # default "latest pacts for this provider" fetch. + broker = verifier.broker_source( + _BROKER_URL, + username=_BROKER_USERNAME, + password=_BROKER_PASSWORD, + selector=True, ) + broker.include_pending() + provider_branch = os.environ.get("PACT_PROVIDER_BRANCH") + if provider_branch: + broker.provider_branch(provider_branch) + broker.build() else: assert _LOCAL_PACT_DIR is not None # guaranteed by module skipif verifier.add_source(_LOCAL_PACT_DIR)