diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index f2068187..5fc63f09 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -1073,6 +1073,49 @@ class WebDAVClient(BaseNextcloudClient): limit=limit, ) + async def file_accessible_by_id(self, file_id: int) -> bool: + """ACL-aware access check for a file by its global Nextcloud file ID. + + Used by verify-on-read (ADR-019). Searches the authenticated user's + whole files tree — which *includes mounted shares* — via WebDAV SEARCH + (RFC 5323) filtered on ``oc:fileid``, returning True iff the user can + currently access the file. + + This is the only check that resolves shared files correctly: + + - :meth:`get_file_info` resolves a path under the caller's *own* root, + so it 404s on a file shared into the caller's account (Nextcloud + mounts received shares at the recipient's root by basename, a + different path than the owner indexed). + - The ``/remote.php/dav/meta/{id}/`` endpoint resolves only the user's + *own* storage, so it 404s on shared files too. + + SEARCH-by-fileid handles all cases: owned files, directly-shared files, + and files reachable via a shared parent folder (verified empirically). + + Args: + file_id: Nextcloud internal (global) file ID. + + Returns: + True if the user can access the file, False if it is not present + in their tree (not owned and not shared with them). + + Raises: + HTTPStatusError: On transport/server errors — callers treat these + as transient (keep the result), not as a definitive denial. + """ + where = ( + "" + f"{int(file_id)}" + ) + results = await self.search_files( + scope="", # user's whole files tree, incl. mounted shares + where_conditions=where, + properties=["fileid"], + limit=1, + ) + return len(results) > 0 + async def _get_file_info_by_id(self, file_id: int) -> Dict[str, Any]: """Get file information by Nextcloud file ID using WebDAV. diff --git a/nextcloud_mcp_server/search/verification.py b/nextcloud_mcp_server/search/verification.py index 480f3515..ee4588df 100644 --- a/nextcloud_mcp_server/search/verification.py +++ b/nextcloud_mcp_server/search/verification.py @@ -138,39 +138,39 @@ async def _verify_files( async def check(result: SearchResult) -> None: doc_id = result.id # file_path is propagated from the Qdrant payload by the algorithm - # layer (bm25_hybrid.py / semantic.py). No extra Qdrant round-trip. + # layer (bm25_hybrid.py / semantic.py); kept here only for log context. file_path = (result.metadata or {}).get("path") - if not file_path: - # Cannot verify without a path; treat as accessible to avoid - # silently dropping legitimate results when payload is missing - # (legacy data, or a future doc_type that doesn't propagate path). + + # Verify by *global* file ID via an ACL-aware WebDAV SEARCH, NOT by + # path. For files the vector ``doc_id`` IS the Nextcloud file ID, and + # file_accessible_by_id searches the user's whole tree (incl. mounted + # shares), so a file an owner shared with this user verifies as + # accessible even though it lives at a different path under the owner's + # root. A path-based check (the old behaviour) would 404 on shared + # files mounted at the recipient's root by basename and silently drop + # legitimate ACL-aware-search results. + # + # Hoisted cast mirrors _verify_notes: a malformed id keeps the result + # (fail open) with a specific log line rather than a generic + # "unexpected error" from the catch-all below. + try: + file_id_int = int(doc_id) + except (TypeError, ValueError) as e: logger.warning( - "No file path in metadata for file_id %s; keeping result " - "(verification skipped)", + "Non-numeric file id %r (%s): %s; keeping result", doc_id, + file_path, + e, ) accessible.add(doc_id) return async with semaphore: try: - info = await client.webdav.get_file_info(file_path) - if info is None: - # Contract (see WebDAVClient.get_file_info docstring): - # `None` means a malformed PROPFIND response — an - # ambiguous state, not a definitive 404. Treat as - # transient and KEEP the result rather than evicting. - # Real 404s raise HTTPStatusError and land in the - # _is_definitive_404_or_403 branch below. - logger.warning( - "Malformed PROPFIND response verifying file %s (%s); " - "keeping result (ambiguous state, not a definitive 404)", - doc_id, - file_path, - ) + if await client.webdav.file_accessible_by_id(file_id_int): accessible.add(doc_id) - return - accessible.add(doc_id) + # else: definitively inaccessible (not owned, not shared) — + # drop and let the caller schedule eviction. except HTTPStatusError as e: if _is_definitive_404_or_403(e): return @@ -183,6 +183,8 @@ async def _verify_files( ) accessible.add(doc_id) except Exception as e: + # Network blip / unexpected WebDAV error — ambiguous, not a + # definitive denial. Keep the result; the next query re-verifies. logger.warning( "Unexpected error verifying file %s (%s): %s; keeping result", doc_id, diff --git a/tests/integration/test_acl_owner_filter.py b/tests/integration/test_acl_owner_filter.py new file mode 100644 index 00000000..068362c3 --- /dev/null +++ b/tests/integration/test_acl_owner_filter.py @@ -0,0 +1,173 @@ +"""ACL-aware ownership filter — deterministic, in-memory Qdrant. + +Proves the query-time ownership expansion added for ACL-aware search +(``search/access_filter.build_ownership_filter`` → +``SemanticSearchAlgorithm``): a user finds documents whose owner shared them +(``owner_id`` ∈ accessible_owners), does not find documents owned by users who +have not shared with them, and legacy points carrying only ``user_id`` stay +findable by their original indexer. + +This complements ``tests/unit/search/test_access_filter.py`` (filter +construction in isolation) by exercising the filter against a real Qdrant +engine through the actual search algorithm — no Nextcloud, no verification +layer, no background sync, so it is fast and deterministic. The full +real-Nextcloud flow (share + verify-on-read) lives in +``test_acl_shared_search.py``. +""" + +import pytest +from qdrant_client import AsyncQdrantClient +from qdrant_client.models import Distance, PointStruct, VectorParams + +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.embedding import SimpleEmbeddingProvider +from nextcloud_mcp_server.search.semantic import SemanticSearchAlgorithm + +pytestmark = pytest.mark.integration + +# Same text for every point so cosine similarity to the query is ~identical: +# the *filter*, not the score, must decide what each user sees. +_DOC_TEXT = "Quarterly infrastructure budget planning and resource allocation" + +# (point_id, doc_id, owner_id, user_id) — owner_id=None mimics a legacy point +# indexed before the owner_id payload field existed. +_ALICE_FILE = (101, "101", "alice", "alice") +_CHARLIE_FILE = (102, "102", "charlie", "charlie") +_LEGACY_DAVE_FILE = (103, "103", None, "dave") + + +@pytest.fixture +async def seeded_collection(monkeypatch): + """In-memory Qdrant seeded with three file points, wired into the algorithm. + + Yields the ``SimpleEmbeddingProvider`` so the test can build a query vector + identical to the one the algorithm will generate. + """ + provider = SimpleEmbeddingProvider(dimension=384) + client = AsyncQdrantClient(":memory:") + collection = get_settings().get_collection_name() + + # The production collection uses a named "dense" vector (see + # vector/qdrant_client.py); the semantic algorithm queries using="dense". + await client.create_collection( + collection_name=collection, + vectors_config={"dense": VectorParams(size=384, distance=Distance.COSINE)}, + ) + + embedding = await provider.embed(_DOC_TEXT) + points = [] + for point_id, doc_id, owner_id, user_id in ( + _ALICE_FILE, + _CHARLIE_FILE, + _LEGACY_DAVE_FILE, + ): + payload = { + "doc_id": doc_id, + "doc_type": "file", + "user_id": user_id, + "is_placeholder": False, + "file_path": f"docs/{doc_id}.txt", + "title": f"file {doc_id}", + "excerpt": _DOC_TEXT, + "chunk_index": 0, + "total_chunks": 1, + } + # Legacy points carry no owner_id at all. + if owner_id is not None: + payload["owner_id"] = owner_id + points.append( + PointStruct(id=point_id, vector={"dense": embedding}, payload=payload) + ) + + await client.upsert(collection_name=collection, points=points, wait=True) + + # Point the algorithm at the in-memory client + deterministic embeddings. + async def _fake_get_qdrant_client(): + return client + + monkeypatch.setattr( + "nextcloud_mcp_server.search.semantic.get_qdrant_client", + _fake_get_qdrant_client, + ) + monkeypatch.setattr( + "nextcloud_mcp_server.search.semantic.get_embedding_service", + lambda: provider, + ) + + yield provider + + await client.close() + + +def _ids(results): + return {r.id for r in results} + + +async def test_shared_owner_is_visible_unshared_is_not(seeded_collection): + """Bob sees Alice's file (shared → owner in accessible_owners), not Charlie's.""" + algo = SemanticSearchAlgorithm(score_threshold=0.0) + + results = await algo.search( + query=_DOC_TEXT, + user_id="bob", + limit=10, + doc_type="file", + accessible_owners=["bob", "alice"], + ) + + found = _ids(results) + assert "101" in found, "Alice's shared file must be discoverable by Bob" + assert "102" not in found, "Charlie's unshared file must NOT be visible to Bob" + assert "103" not in found, "Legacy file owned by dave must NOT be visible to Bob" + + +async def test_no_shares_sees_only_own(seeded_collection): + """With no shares, Bob (who owns nothing here) gets nothing.""" + algo = SemanticSearchAlgorithm(score_threshold=0.0) + + results = await algo.search( + query=_DOC_TEXT, + user_id="bob", + limit=10, + doc_type="file", + accessible_owners=["bob"], + ) + + assert _ids(results) == set() + + +async def test_legacy_user_id_point_still_found_by_indexer(seeded_collection): + """A pre-owner_id point stays findable by its original indexer via the + legacy ``user_id`` OR-branch in build_ownership_filter.""" + algo = SemanticSearchAlgorithm(score_threshold=0.0) + + results = await algo.search( + query=_DOC_TEXT, + user_id="dave", + limit=10, + doc_type="file", + accessible_owners=["dave"], + ) + + found = _ids(results) + assert "103" in found, "dave must still find his own legacy (user_id-only) file" + assert "101" not in found + assert "102" not in found + + +async def test_owner_sees_own_new_style_point(seeded_collection): + """Alice finds her own file via the owner_id branch.""" + algo = SemanticSearchAlgorithm(score_threshold=0.0) + + results = await algo.search( + query=_DOC_TEXT, + user_id="alice", + limit=10, + doc_type="file", + accessible_owners=["alice"], + ) + + found = _ids(results) + assert "101" in found + assert "102" not in found + assert "103" not in found diff --git a/tests/integration/test_acl_shared_search.py b/tests/integration/test_acl_shared_search.py new file mode 100644 index 00000000..4302f2a3 --- /dev/null +++ b/tests/integration/test_acl_shared_search.py @@ -0,0 +1,186 @@ +"""End-to-end ACL-aware semantic search against a real Nextcloud (PR #813). + +This is the card-120 acceptance criterion exercised across the *new* code +paths together: + +1. ``list_accessible_owners`` resolves the querying user's real OCS shares into + the set of owner UIDs they may search. +2. ``SemanticSearchAlgorithm`` applies the expanded ownership filter in Qdrant. +3. ``verify_search_results`` re-checks each hit against real Nextcloud + (ACL-aware, by global file id). + +Qdrant is in-memory and seeded directly with one point owned by *alice* — this +deliberately stands in for the background scanner (whose only relevant change +is writing ``owner_id`` into the payload, covered separately). Nextcloud itself +is real, so the share lookup (step 1) and the verification (step 3) exercise +the live OCS Sharing + WebDAV APIs. The result: bob, with whom alice shared the +file, finds it without having indexed anything; diana, with no share, does not. + +The pure-filter matrix lives in ``test_acl_owner_filter.py`` and the +verification layer in ``test_verify_on_read.py``; this test is the glue that +proves the real share → accessible_owners → filter → verify chain. +""" + +import os +import uuid + +import pytest +from httpx import BasicAuth +from qdrant_client import AsyncQdrantClient +from qdrant_client.models import Distance, PointStruct, VectorParams + +from nextcloud_mcp_server.client import NextcloudClient +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.embedding import SimpleEmbeddingProvider +from nextcloud_mcp_server.search.access_filter import list_accessible_owners +from nextcloud_mcp_server.search.semantic import SemanticSearchAlgorithm +from nextcloud_mcp_server.search.verification import verify_search_results + +pytestmark = pytest.mark.integration + +_DOC_TEXT = "Confidential quarterly infrastructure budget and capacity plan" + + +def _user_client(username: str, password: str) -> NextcloudClient: + return NextcloudClient( + base_url=os.environ["NEXTCLOUD_HOST"], + username=username, + auth=BasicAuth(username, password), + password=password, + ) + + +@pytest.fixture +async def acl_users(test_users_setup): + """alice (owner), bob (recipient), diana (no access) direct clients.""" + clients = { + name: _user_client(name, test_users_setup[name]["password"]) + for name in ("alice", "bob", "diana") + } + try: + yield clients + finally: + for c in clients.values(): + await c._client.aclose() + + +@pytest.fixture +async def shared_file(acl_users): + """alice creates a nested file and shares it with bob (not diana). + + Yields (file_id, owner_relative_path); cleans up the directory after. + """ + alice = acl_users["alice"] + suffix = uuid.uuid4().hex[:8] + test_dir = f"acl_e2e_{suffix}" + nested = f"{test_dir}/reports" + path = f"{nested}/budget.txt" + + await alice.webdav.create_directory(test_dir) + await alice.webdav.create_directory(nested) + await alice.webdav.write_file(path, _DOC_TEXT.encode(), "text/plain") + file_id = (await alice.webdav.get_file_info(path))["id"] + await alice.sharing.create_share( + path=f"/{path}", share_with="bob", share_type=0, permissions=1 + ) + try: + yield file_id, path + finally: + await alice.webdav.delete_resource(test_dir) + + +@pytest.fixture +async def seeded_semantic(monkeypatch, shared_file): + """In-memory Qdrant carrying alice's file point, wired into the algorithm. + + Stands in for the background scanner: the point carries ``owner_id=alice`` + exactly as the scanner now writes it. + """ + file_id, path = shared_file + provider = SimpleEmbeddingProvider(dimension=384) + client = AsyncQdrantClient(":memory:") + collection = get_settings().get_collection_name() + await client.create_collection( + collection_name=collection, + vectors_config={"dense": VectorParams(size=384, distance=Distance.COSINE)}, + ) + await client.upsert( + collection_name=collection, + points=[ + PointStruct( + id=int(file_id), + vector={"dense": await provider.embed(_DOC_TEXT)}, + payload={ + "doc_id": str(file_id), + "doc_type": "file", + "owner_id": "alice", + "user_id": "alice", + "is_placeholder": False, + "file_path": path, + "title": "budget.txt", + "excerpt": _DOC_TEXT, + "chunk_index": 0, + "total_chunks": 1, + }, + ) + ], + wait=True, + ) + + async def _fake_get_qdrant_client(): + return client + + monkeypatch.setattr( + "nextcloud_mcp_server.search.semantic.get_qdrant_client", + _fake_get_qdrant_client, + ) + monkeypatch.setattr( + "nextcloud_mcp_server.search.semantic.get_embedding_service", + lambda: provider, + ) + yield file_id + await client.close() + + +async def _search_as(user_client, file_id_unused) -> list: + """Run the full new chain (share lookup → filter → verify) as a user.""" + accessible_owners = await list_accessible_owners( + user_client.sharing, user_client.username + ) + algo = SemanticSearchAlgorithm(score_threshold=0.0) + unverified = await algo.search( + query=_DOC_TEXT, + user_id=user_client.username, + limit=10, + doc_type="file", + accessible_owners=accessible_owners, + ) + kept, _dropped = await verify_search_results(user_client, unverified) + return kept + + +async def test_recipient_finds_shared_file_without_indexing(acl_users, seeded_semantic): + """Bob finds alice's shared file end-to-end: real share lookup expands his + accessible owners to include alice, the filter surfaces her point, and + real verification confirms his ACL access — all without bob indexing.""" + file_id = seeded_semantic + # Sanity: the live OCS lookup really does expand bob to include alice. + owners = await list_accessible_owners(acl_users["bob"].sharing, "bob") + assert "alice" in owners, "OCS shared-with-me must surface alice as an owner" + + kept = await _search_as(acl_users["bob"], file_id) + + assert [r.id for r in kept] == [str(file_id)], ( + "bob must find alice's shared file via semantic search" + ) + + +async def test_non_recipient_does_not_find_file(acl_users, seeded_semantic): + """Diana, with no share, never sees the file: her accessible-owners set + excludes alice, so the ownership filter drops the point before verification.""" + owners = await list_accessible_owners(acl_users["diana"].sharing, "diana") + assert "alice" not in owners + + kept = await _search_as(acl_users["diana"], seeded_semantic) + + assert kept == [], "diana (no share) must not find alice's file" diff --git a/tests/integration/test_astrolabe_login_flow_provisioning.py b/tests/integration/test_astrolabe_login_flow_provisioning.py deleted file mode 100644 index 15d2abbb..00000000 --- a/tests/integration/test_astrolabe_login_flow_provisioning.py +++ /dev/null @@ -1,139 +0,0 @@ -"""Integration test for Astrolabe's "Enable Semantic Search" OAuth flow on -the `mcp-login-flow` profile. - -Cross-system interface test. Brings together Astrolabe (Nextcloud PHP app -installed at container start by ``app-hooks/post-installation``) with the -``mcp-login-flow`` MCP server over OAuth + the management API. Mirrors -the production-shaped flow that PR #773's recent -`ALLOWED_MGMT_CLIENT` ↔ `astrolabeMcpClientOAuth00000000000` drift was -masking — every management API call from Astrolabe (e.g. -``/api/v1/users/admin/session``) was returning 401 because the -real-deployment client id was not in the test-fixture allowlist, so the -Astrolabe settings page never updated to reflect a successful -authorization. - -This test is **regression coverage** for that class of drift. If the -Astrolabe client id ever falls out of `mcp-login-flow`'s -``ALLOWED_MGMT_CLIENT`` again, the post-redirect assertions here will -fail because the page state stays on ``oauth-required.php``. - -Requires the login-flow stack to be running: - - MCP_SERVER_URL=http://mcp-login-flow:8004 \\ - docker compose --profile login-flow up -d app db mcp-login-flow - -The ``app-hooks/before-starting/26-configure-astrolabe-oauth.sh`` hook -creates the OAuth client with the production-shaped id -``astrolabeMcpClientOAuth00000000000`` automatically when -``MCP_SERVER_URL`` is set, so no fixture-level OIDC client creation is -needed here. -""" - -import logging -import os -import re - -import pytest -from playwright.async_api import Page - -# Reuse helpers from the multi-user-basic Astrolabe test for login + nav. -from tests.integration.test_astrolabe_multi_user_background_sync import ( - login_to_nextcloud, - navigate_to_astrolabe_settings, -) - -logger = logging.getLogger(__name__) - -pytestmark = [pytest.mark.integration, pytest.mark.login_flow] - -NEXTCLOUD_URL = "http://localhost:8080" -ASTROLABE_SETTINGS_URL = f"{NEXTCLOUD_URL}/settings/user/astrolabe" - - -async def _click_enable_semantic_search(page: Page) -> None: - """Click the "Enable Semantic Search" OAuth link on the - ``oauth-required.php`` template that login-flow mode renders to a - not-yet-authorized user. - - Astrolabe's own e2e helper (``third_party/astrolabe/tests/e2e/helpers/ - authorize.ts``) targets the same link by accessible name. - """ - enable_link = page.get_by_role("link", name="Enable Semantic Search") - await enable_link.wait_for(state="visible", timeout=10_000) - logger.info("Clicking 'Enable Semantic Search' OAuth link") - await enable_link.click() - - -async def _grant_oidc_consent(page: Page) -> None: - """Click "Allow" on the Nextcloud OIDC consent screen, if shown. - - Nextcloud may auto-redirect for already-trusted clients, in which - case the consent button never appears — that's not an error. - """ - allow_button = page.get_by_role("button", name=re.compile(r"^allow$", re.I)) - try: - await allow_button.wait_for(state="visible", timeout=10_000) - logger.info("Clicking 'Allow' on OIDC consent") - await allow_button.click(force=True) - except Exception: - logger.info( - "OIDC consent screen not visible — assuming auto-grant for " - "already-trusted client" - ) - - -@pytest.mark.timeout(180) -async def test_enable_semantic_search_completes_oauth_for_login_flow(browser): - """Click the "Enable Semantic Search" link, grant consent, and assert - the post-redirect page reflects a completed authorization. - - The success criterion is intentionally negative: after the OAuth - flow, the original "Enable Semantic Search" link must be gone. If - Astrolabe's management API call is rejected by the MCP server (HTTP - 401, the original bug), the page falls back to the same - ``oauth-required.php`` template and the link reappears — making this - test the canary for the drift class. - """ - admin_password = os.getenv("NEXTCLOUD_PASSWORD") - if admin_password is None: - raise RuntimeError("NEXTCLOUD_PASSWORD must be set") - page = await browser.new_page() - try: - await login_to_nextcloud(page, "admin", admin_password) - await navigate_to_astrolabe_settings(page) - - # Sanity-check we're on the not-yet-authorized template. - enable_link = page.get_by_role("link", name="Enable Semantic Search") - if await enable_link.count() == 0: - pytest.skip( - "Astrolabe is already authorized for admin (oauth-required.php " - "not rendered). Reset by clearing the user's OAuth tokens " - "before re-running this test." - ) - - await _click_enable_semantic_search(page) - await _grant_oidc_consent(page) - - # OAuth callback returns to /apps/astrolabe/oauth/callback then the - # controller redirects to /settings/user/astrolabe. - await page.wait_for_url(re.compile(r"/settings/user/astrolabe"), timeout=30_000) - await page.wait_for_load_state("networkidle", timeout=15_000) - - # Regression assertion for the ALLOWED_MGMT_CLIENT drift bug: - # the page must have moved past oauth-required.php. If - # Astrolabe's management API call to /api/v1/users/{id}/session - # is rejected (401), the session lookup falls back to "no token", - # and the same oauth-required.php template re-renders with the - # link still present. - post_auth_count = await page.get_by_role( - "link", name="Enable Semantic Search" - ).count() - assert post_auth_count == 0, ( - "'Enable Semantic Search' link still visible after completing " - "OAuth flow — Astrolabe could not read the user's session from " - "the MCP server. Most likely cause: " - "`astrolabeMcpClientOAuth00000000000` missing from " - "`ALLOWED_MGMT_CLIENT` on `mcp-login-flow`." - ) - finally: - await page.close() diff --git a/tests/integration/test_astrolabe_token_refresh.py b/tests/integration/test_astrolabe_token_refresh.py deleted file mode 100644 index e9b9c553..00000000 --- a/tests/integration/test_astrolabe_token_refresh.py +++ /dev/null @@ -1,702 +0,0 @@ -"""Integration tests for Astrolabe token refresh flow. - -Cross-system interface test: Tests the MCP server's integration with the -Astrolabe Nextcloud app, which is installed from the Nextcloud app store via -app-hooks/post-installation/20-install-astrolabe-app.sh. Astrolabe source -lives in a separate repository (https://github.com/cbcoutinho/astrolabe). - -Tests the token refresh mechanism between Astrolabe (Nextcloud app) -and the MCP server backend in a multi-user basic auth deployment. - -This test verifies: -1. User provisions access via Astrolabe personal settings -2. Token is stored encrypted in Nextcloud database -3. Token expires (simulated via database manipulation) -4. MCP server requests new token via refresh -5. Astrolabe refreshes token with IdP -6. New token is stored and used successfully - -Note: The mcp-multi-user-basic deployment uses "hybrid mode" which requires -BOTH OAuth authorization AND app password for full configuration. These tests -focus on the app password/credential storage aspects and verify database state -directly rather than relying on UI elements that require both steps. -""" - -import logging -import re -import subprocess - -import anyio -import pytest -from playwright.async_api import Page - -pytestmark = [pytest.mark.integration, pytest.mark.multi_user_basic] - -logger = logging.getLogger(__name__) - - -async def login_to_nextcloud(page: Page, username: str, password: str): - """Helper function to login to Nextcloud via Playwright. - - Args: - page: Playwright page instance - username: Nextcloud username - password: Nextcloud password - """ - nextcloud_url = "http://localhost:8080" - - logger.info("Logging in to Nextcloud as %s...", username) - await page.goto(f"{nextcloud_url}/login", wait_until="networkidle") - - # Fill in login form - await page.wait_for_selector('input[name="user"]', timeout=10000) - await page.fill('input[name="user"]', username) - await page.fill('input[name="password"]', password) - - # Submit form - await page.click('button[type="submit"]') - await page.wait_for_load_state("networkidle", timeout=30000) - - # Verify logged in (should redirect away from login page) - current_url = page.url - assert "/login" not in current_url, ( - f"Login failed for {username}, still on login page" - ) - logger.info("✓ Successfully logged in as %s", username) - - -async def generate_app_password( - page: Page, username: str, app_name: str = "Astrolabe Test" -) -> str: - """Generate an app password in Nextcloud Security settings. - - Args: - page: Playwright page instance (must be authenticated) - username: Username (for logging) - app_name: Name for the app password - - Returns: - The generated app password string - """ - logger.info("Generating app password for %s...", username) - - nextcloud_url = "http://localhost:8080" - - # Navigate to Security settings - await page.goto(f"{nextcloud_url}/settings/user/security", wait_until="networkidle") - logger.info("Navigated to Security settings") - - # Fill the app password input field - app_password_input = page.locator('input[placeholder="App name"]') - await app_password_input.fill(app_name) - logger.info("Entered app name: %s", app_name) - - # Wait for Vue.js to react and enable the button - await anyio.sleep(1.0) - - # Click the create button - create_button = page.locator( - 'button[type="submit"]:has-text("Create new app password")' - ) - await create_button.click() - logger.info("Clicked create app password button") - - # Wait for app password to be generated - await anyio.sleep(3) - - # Find the generated app password - app_password = None - try: - await page.wait_for_selector('text="New app password"', timeout=10000) - logger.info("App password dialog appeared") - - all_inputs = await page.locator('input[type="text"]').all() - for idx, input_elem in enumerate(all_inputs): - try: - value = await input_elem.input_value() - if value and "-" in value and len(value) > 20: - app_password = value.strip() - logger.info("Found app password in input %s", idx) - break - except Exception: - continue - except Exception as e: - logger.error("Failed to find app password dialog: %s", e) - - if not app_password: - screenshot_path = f"/tmp/app_password_generation_{username}.png" - await page.screenshot(path=screenshot_path) - raise ValueError( - f"Could not find generated app password. Screenshot: {screenshot_path}" - ) - - # Validate password format - if not re.match( - r"^[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}$", - app_password, - ): - raise ValueError(f"App password format validation failed: {app_password}") - - logger.info("✓ Generated app password for %s", username) - - # Close the dialog - close_button = page.get_by_role("button", name="Close") - await close_button.click() - await anyio.sleep(0.5) - - return app_password - - -async def save_app_password_in_astrolabe( - page: Page, username: str, app_password: str -) -> bool: - """Save app password in Astrolabe settings (Step 2 of hybrid mode). - - This function only saves the app password - it does NOT verify the "Active" - badge since that requires both OAuth and app password in hybrid mode. - - Args: - page: Playwright page instance - username: Username (for logging) - app_password: App password to enter - - Returns: - True if the password was saved successfully (based on network response) - """ - logger.info("Saving app password in Astrolabe for %s...", username) - - nextcloud_url = "http://localhost:8080" - - # Track network responses - credentials_response_status = None - - def capture_response(resp): - nonlocal credentials_response_status - if "background-sync/credentials" in resp.url or "storeAppPassword" in resp.url: - credentials_response_status = resp.status - logger.info("Credentials endpoint response: %s %s", resp.status, resp.url) - - page.on("response", capture_response) - - # Navigate to Astrolabe settings - await page.goto( - f"{nextcloud_url}/settings/user/astrolabe", wait_until="networkidle" - ) - await anyio.sleep(1) - - # Check if Step 2 already shows "Complete" - try: - complete_badge = page.locator('text="Complete"').first - if await complete_badge.is_visible(timeout=2000): - logger.info("✓ App password already configured for %s", username) - return True - except Exception: - pass - - # Find the app password input field - app_password_input = page.get_by_placeholder("xxxxx-xxxxx-xxxxx-xxxxx-xxxxx") - - try: - await app_password_input.wait_for(timeout=5000, state="visible") - logger.info("Found app password input field") - except Exception: - screenshot_path = f"/tmp/astrolabe_no_password_field_{username}.png" - await page.screenshot(path=screenshot_path) - raise ValueError( - f"Could not find app password input field. Screenshot: {screenshot_path}" - ) - - # Enter the app password - await app_password_input.fill(app_password) - logger.info("Entered app password for %s", username) - - await anyio.sleep(0.5) - - # Click Save button - save_button = page.get_by_role("button", name="Save") - await save_button.click() - logger.info("Clicked Save button") - - # Wait for the request to complete and page to reload - await page.wait_for_load_state("networkidle", timeout=15000) - await anyio.sleep(2) - - # Verify the save was successful by checking network response - if credentials_response_status == 200: - logger.info("✓ App password saved successfully for %s", username) - return True - else: - logger.error( - "App password save failed for %s, status: %s", - username, - credentials_response_status, - ) - screenshot_path = f"/tmp/astrolabe_save_failed_{username}.png" - await page.screenshot(path=screenshot_path) - return False - - -def get_background_sync_credentials(username: str) -> dict | None: - """Get background sync credentials for a user from the database. - - Args: - username: Nextcloud username - - Returns: - Dict with credential details, or None if not found - """ - query = f""" - SELECT configkey, configvalue - FROM oc_preferences - WHERE userid = '{username}' - AND appid = 'astrolabe' - AND configkey IN ('background_sync_password', 'background_sync_type', 'background_sync_provisioned_at') - ORDER BY configkey; - """ - - try: - result = subprocess.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "mariadb", - "-u", - "root", - "-ppassword", - "nextcloud", - "-e", - query, - ], - capture_output=True, - text=True, - timeout=10, - ) - - output = result.stdout - if "background_sync_type" in output: - return { - "has_password": "background_sync_password" in output, - "has_type": "background_sync_type" in output, - "has_timestamp": "background_sync_provisioned_at" in output, - "is_app_password": "app_password" in output, - } - return None - - except Exception as e: - logger.error("Error getting credentials for %s: %s", username, e) - return None - - -def delete_user_credentials(username: str) -> bool: - """Delete all stored credentials for a user (for cleanup). - - Args: - username: Nextcloud username - - Returns: - True if successful - """ - query = f""" - DELETE FROM oc_preferences - WHERE userid = '{username}' - AND appid = 'astrolabe' - AND configkey IN ('oauth_tokens', 'background_sync_password', 'background_sync_type', 'background_sync_provisioned_at'); - """ - - try: - result = subprocess.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "mariadb", - "-u", - "root", - "-ppassword", - "nextcloud", - "-e", - query, - ], - capture_output=True, - text=True, - timeout=10, - ) - - logger.info("Deleted credentials for %s", username) - return result.returncode == 0 - - except Exception as e: - logger.error("Error deleting credentials for %s: %s", username, e) - return False - - -@pytest.mark.integration -@pytest.mark.multi_user_basic -async def test_app_password_storage_and_cleanup( - browser, - nc_client, - test_users_setup, - configure_astrolabe_for_mcp_server, -): - """Test that app passwords are stored and cleaned up correctly. - - This test verifies: - 1. User can save app password in Astrolabe settings - 2. Password is stored encrypted in the database - 3. Credentials can be revoked and are deleted from database - - Note: In hybrid mode (mcp-multi-user-basic), this only tests Step 2 - (app password storage). The "Active" badge requires both OAuth and - app password, which is tested separately. - """ - # Configure Astrolabe for mcp-multi-user-basic - logger.info("Configuring Astrolabe for mcp-multi-user-basic server...") - await configure_astrolabe_for_mcp_server( - mcp_server_internal_url="http://mcp-multi-user-basic:8000", - mcp_server_public_url="http://localhost:8003", - ) - - username = "alice" - user_config = test_users_setup[username] - password = user_config["password"] - - # Cleanup any existing credentials - delete_user_credentials(username) - - context = await browser.new_context(ignore_https_errors=True) - page = await context.new_page() - - try: - # Step 1: Login - await login_to_nextcloud(page, username, password) - - # Step 2: Verify no credentials exist initially - initial_creds = get_background_sync_credentials(username) - assert initial_creds is None, f"Expected no credentials, found: {initial_creds}" - logger.info("✓ Verified no initial credentials") - - # Step 3: Generate app password - app_password = await generate_app_password(page, username) - assert app_password, "Failed to generate app password" - - # Step 4: Save app password in Astrolabe - save_success = await save_app_password_in_astrolabe( - page, username, app_password - ) - assert save_success, "Failed to save app password" - - # Step 5: Verify credentials are stored in database - stored_creds = get_background_sync_credentials(username) - assert stored_creds is not None, "Expected credentials to be stored" - assert stored_creds["has_password"], "Expected password to be stored" - assert stored_creds["has_type"], "Expected type to be stored" - assert stored_creds["is_app_password"], "Expected type to be 'app_password'" - logger.info("✓ Verified credentials stored in database") - - # Step 6: Verify password is encrypted (not plaintext) - query = f""" - SELECT configvalue - FROM oc_preferences - WHERE userid = '{username}' - AND appid = 'astrolabe' - AND configkey = 'background_sync_password'; - """ - - result = subprocess.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "mariadb", - "-u", - "root", - "-ppassword", - "nextcloud", - "-N", - "-e", - query, - ], - capture_output=True, - text=True, - timeout=10, - ) - - encrypted_value = result.stdout.strip() - assert app_password not in encrypted_value, "Password appears in plaintext!" - assert len(encrypted_value) > len(app_password), ( - "Encrypted value should be longer" - ) - logger.info("✓ Verified password is encrypted") - - finally: - await context.close() - # Cleanup - delete_user_credentials(username) - - -@pytest.mark.integration -@pytest.mark.multi_user_basic -async def test_credential_isolation_between_users( - browser, - nc_client, - test_users_setup, - configure_astrolabe_for_mcp_server, -): - """Test that credentials are properly isolated between users. - - This test verifies: - 1. Multiple users can provision credentials independently - 2. Each user's encrypted credentials are unique - 3. Deleting one user's credentials doesn't affect others - """ - await configure_astrolabe_for_mcp_server( - mcp_server_internal_url="http://mcp-multi-user-basic:8000", - mcp_server_public_url="http://localhost:8003", - ) - - test_users = ["alice", "bob"] - user_passwords = {} - - # Cleanup all users first - for username in test_users: - delete_user_credentials(username) - - # Provision each user - for username in test_users: - user_config = test_users_setup[username] - password = user_config["password"] - - context = await browser.new_context(ignore_https_errors=True) - page = await context.new_page() - - try: - await login_to_nextcloud(page, username, password) - app_password = await generate_app_password( - page, username, f"Test {username}" - ) - save_success = await save_app_password_in_astrolabe( - page, username, app_password - ) - - assert save_success, f"Failed to save app password for {username}" - user_passwords[username] = app_password - - # Verify stored - creds = get_background_sync_credentials(username) - assert creds is not None, f"Credentials not stored for {username}" - logger.info("✓ Credentials provisioned for %s", username) - - finally: - await context.close() - - # Verify isolation - get encrypted values - encrypted_values = {} - for username in test_users: - query = f""" - SELECT configvalue - FROM oc_preferences - WHERE userid = '{username}' - AND appid = 'astrolabe' - AND configkey = 'background_sync_password'; - """ - - result = subprocess.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "mariadb", - "-u", - "root", - "-ppassword", - "nextcloud", - "-N", - "-e", - query, - ], - capture_output=True, - text=True, - timeout=10, - ) - encrypted_values[username] = result.stdout.strip() - - # Different users should have different encrypted values - assert encrypted_values["alice"] != encrypted_values["bob"], ( - "Different users should have different encrypted values" - ) - logger.info("✓ Verified credentials are unique per user") - - # Delete alice's credentials and verify bob's are unaffected - delete_user_credentials("alice") - - alice_creds = get_background_sync_credentials("alice") - bob_creds = get_background_sync_credentials("bob") - - assert alice_creds is None, "Alice's credentials should be deleted" - assert bob_creds is not None, "Bob's credentials should still exist" - logger.info("✓ Verified credential deletion is isolated") - - # Cleanup - for username in test_users: - delete_user_credentials(username) - - -@pytest.mark.integration -@pytest.mark.multi_user_basic -async def test_credential_revoke_and_reprovision( - browser, - nc_client, - test_users_setup, - configure_astrolabe_for_mcp_server, -): - """Test that credentials can be revoked and reprovisioned. - - This test verifies: - 1. User provisions credentials - 2. User revokes credentials (deletes from database) - 3. User provisions again with new app password - 4. New credentials are stored correctly - - Note: The UI prevents overwriting credentials directly - users must - revoke first before provisioning new credentials. - """ - await configure_astrolabe_for_mcp_server( - mcp_server_internal_url="http://mcp-multi-user-basic:8000", - mcp_server_public_url="http://localhost:8003", - ) - - username = "alice" - user_config = test_users_setup[username] - password = user_config["password"] - - delete_user_credentials(username) - - context = await browser.new_context(ignore_https_errors=True) - page = await context.new_page() - - try: - await login_to_nextcloud(page, username, password) - - # First provisioning - app_password_1 = await generate_app_password(page, username, "First Password") - await save_app_password_in_astrolabe(page, username, app_password_1) - - # Get first encrypted value - query = f""" - SELECT configvalue - FROM oc_preferences - WHERE userid = '{username}' - AND appid = 'astrolabe' - AND configkey = 'background_sync_password'; - """ - - result1 = subprocess.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "mariadb", - "-u", - "root", - "-ppassword", - "nextcloud", - "-N", - "-e", - query, - ], - capture_output=True, - text=True, - timeout=10, - ) - first_encrypted = result1.stdout.strip() - assert first_encrypted, "First credential should be stored" - logger.info("✓ First credential stored") - - # Revoke credentials (simulating user clicking "Revoke Access") - delete_user_credentials(username) - logger.info("✓ Credentials revoked") - - # Verify credentials are gone - creds_after_revoke = get_background_sync_credentials(username) - assert creds_after_revoke is None, "Credentials should be deleted after revoke" - - # Second provisioning with different password - app_password_2 = await generate_app_password(page, username, "Second Password") - await save_app_password_in_astrolabe(page, username, app_password_2) - - result2 = subprocess.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "mariadb", - "-u", - "root", - "-ppassword", - "nextcloud", - "-N", - "-e", - query, - ], - capture_output=True, - text=True, - timeout=10, - ) - second_encrypted = result2.stdout.strip() - assert second_encrypted, "Second credential should be stored" - logger.info("✓ Second credential stored") - - # Verify the encrypted values are different (different passwords) - assert first_encrypted != second_encrypted, ( - "Different passwords should produce different encrypted values" - ) - - # Verify only one row exists - count_query = f""" - SELECT COUNT(*) - FROM oc_preferences - WHERE userid = '{username}' - AND appid = 'astrolabe' - AND configkey = 'background_sync_password'; - """ - - count_result = subprocess.run( - [ - "docker", - "compose", - "exec", - "-T", - "db", - "mariadb", - "-u", - "root", - "-ppassword", - "nextcloud", - "-N", - "-e", - count_query, - ], - capture_output=True, - text=True, - timeout=10, - ) - count = int(count_result.stdout.strip()) - assert count == 1, f"Expected 1 credential row, found {count}" - logger.info("✓ Verified clean reprovision after revoke") - - finally: - await context.close() - delete_user_credentials(username) diff --git a/tests/integration/test_verify_on_read.py b/tests/integration/test_verify_on_read.py index ef0e27cf..bb405ca4 100644 --- a/tests/integration/test_verify_on_read.py +++ b/tests/integration/test_verify_on_read.py @@ -23,10 +23,11 @@ behaviour separately. """ import logging +import os import uuid import pytest -from httpx import HTTPStatusError +from httpx import BasicAuth, HTTPStatusError from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.search import verification @@ -48,6 +49,28 @@ def _result_for_note(note_id: int) -> SearchResult: ) +def _result_for_file(file_id: int, path: str) -> SearchResult: + # Mirrors what the algorithm layer propagates: doc_id IS the global file id, + # ``path`` is carried in metadata (owner-relative) for log context only. + return SearchResult( + id=file_id, + doc_type="file", + title=path.split("/")[-1], + excerpt="...", + score=0.9, + metadata={"path": path}, + ) + + +def _user_client(username: str, password: str) -> NextcloudClient: + return NextcloudClient( + base_url=os.environ["NEXTCLOUD_HOST"], + username=username, + auth=BasicAuth(username, password), + password=password, + ) + + async def test_verify_keeps_accessible_note( nc_client: NextcloudClient, temporary_note: dict, mocker ): @@ -170,3 +193,95 @@ async def test_verify_dedupes_chunks_of_same_document( assert dropped_count == 0 # ...but verification only fetched the note ONCE assert spy_get_note.await_count == 1 + + +# --------------------------------------------------------------------------- +# File verifier — cross-user shared access (ACL-aware search, PR #813) +# --------------------------------------------------------------------------- +# +# These exercise the verifier fix that makes ACL-aware search actually work +# end-to-end: a file an owner shared with another user must survive +# verify-on-read for the *recipient*, even when it lives in a subfolder of the +# owner's tree (Nextcloud mounts received shares at the recipient's root by +# basename, so the owner-relative path does NOT resolve under the recipient's +# root). The fix verifies by global file id, which is ACL-aware. + + +@pytest.fixture +async def alice_bob_clients(test_users_setup): + """Direct NextcloudClients for alice (owner) and bob (recipient).""" + alice = _user_client("alice", test_users_setup["alice"]["password"]) + bob = _user_client("bob", test_users_setup["bob"]["password"]) + try: + yield alice, bob + finally: + await alice._client.aclose() + await bob._client.aclose() + + +async def test_verify_keeps_nested_file_shared_with_recipient( + alice_bob_clients, mocker +): + """The PR #813 acceptance check at the verifier layer. + + Alice owns a file in a *subfolder* and shares it with Bob. Verifying the + result as Bob must KEEP it — proving the id-based check sees the share. + A path-based check (the old behaviour) would 404 here and wrongly drop it. + """ + spy_evict = mocker.AsyncMock() + mocker.patch.object(verification, "delete_document_points", spy_evict) + + alice, bob = alice_bob_clients + suffix = uuid.uuid4().hex[:8] + test_dir = f"acl_verify_{suffix}" + nested_dir = f"{test_dir}/reports" + shared_path = f"{nested_dir}/shared.txt" + + await alice.webdav.create_directory(test_dir) + await alice.webdav.create_directory(nested_dir) + await alice.webdav.write_file(shared_path, b"alice's shared report", "text/plain") + file_id = (await alice.webdav.get_file_info(shared_path))["id"] + + await alice.sharing.create_share( + path=f"/{shared_path}", share_with="bob", share_type=0, permissions=1 + ) + + try: + kept, dropped_count = await verify_search_results( + bob, [_result_for_file(file_id, shared_path)] + ) + + assert [r.id for r in kept] == [file_id], ( + "a nested file shared with bob must pass verification for bob" + ) + assert dropped_count == 0 + spy_evict.assert_not_awaited() + finally: + await alice.webdav.delete_resource(test_dir) + + +async def test_verify_drops_unshared_file_for_other_user(alice_bob_clients, mocker): + """Negative control: a file Alice did NOT share is inaccessible to Bob and + must be dropped + scheduled for eviction under his identity.""" + spy_evict = mocker.AsyncMock() + mocker.patch.object(verification, "delete_document_points", spy_evict) + + alice, bob = alice_bob_clients + suffix = uuid.uuid4().hex[:8] + test_dir = f"acl_verify_priv_{suffix}" + private_path = f"{test_dir}/private.txt" + + await alice.webdav.create_directory(test_dir) + await alice.webdav.write_file(private_path, b"alice's private note", "text/plain") + file_id = (await alice.webdav.get_file_info(private_path))["id"] + + try: + kept, dropped_count = await verify_search_results( + bob, [_result_for_file(file_id, private_path)] + ) + + assert kept == [], "an unshared file must not pass verification for bob" + assert dropped_count == 1 + spy_evict.assert_awaited_once_with(file_id, "file", bob.username) + finally: + await alice.webdav.delete_resource(test_dir) diff --git a/tests/unit/search/test_verification.py b/tests/unit/search/test_verification.py index 3cee6981..8bd3c102 100644 --- a/tests/unit/search/test_verification.py +++ b/tests/unit/search/test_verification.py @@ -438,10 +438,16 @@ async def test_verify_news_items_malformed_api_response_keeps_all(mocker): @pytest.mark.unit -async def test_verify_files_uses_path_from_metadata(mocker): - """File verifier reads path from SearchResult.metadata, no Qdrant round-trip.""" +async def test_verify_files_accessible_by_global_id_is_kept(mocker): + """File verifier resolves the file by its global ID (the doc_id), ACL-aware. + + This is what lets a recipient verify a file an owner shared with them: + file_accessible_by_id searches the user's whole tree (incl. mounted + shares) by global file id, not a path under the caller's own root (which + would 404 on shared files mounted at a different path). + """ webdav_client = SimpleNamespace( - get_file_info=mocker.AsyncMock(return_value={"id": 100}) + file_accessible_by_id=mocker.AsyncMock(return_value=True) ) client = SimpleNamespace(webdav=webdav_client, username="alice") @@ -452,14 +458,15 @@ async def test_verify_files_uses_path_from_metadata(mocker): ) assert result == {"100"} - webdav_client.get_file_info.assert_awaited_once_with("Documents/foo.txt") + webdav_client.file_accessible_by_id.assert_awaited_once_with(100) @pytest.mark.unit -async def test_verify_files_404_drops(mocker): - """get_file_info raising HTTPStatusError(404) is a definitive drop.""" +async def test_verify_files_inaccessible_id_drops(mocker): + """file_accessible_by_id returning False (file not in the user's tree) is a + definitive drop — the file is neither owned by nor shared with the user.""" webdav_client = SimpleNamespace( - get_file_info=mocker.AsyncMock(side_effect=_http_error(404)) + file_accessible_by_id=mocker.AsyncMock(return_value=False) ) client = SimpleNamespace(webdav=webdav_client, username="alice") @@ -473,68 +480,49 @@ async def test_verify_files_404_drops(mocker): @pytest.mark.unit -async def test_verify_files_malformed_propfind_keeps_result(mocker): - """get_file_info returning None means malformed PROPFIND — keep the result. +async def test_verify_files_403_404_drops(mocker): + """A 403/404 raised by the SEARCH call is treated as a definitive drop, + consistent with the shared _is_definitive_404_or_403 policy used by every + verifier. (Normal inaccessibility surfaces as an empty result set, not a + status code, and is covered by test_verify_files_inaccessible_id_drops.)""" + for status in (403, 404): + webdav_client = SimpleNamespace( + file_accessible_by_id=mocker.AsyncMock(side_effect=_http_error(status)) + ) + client = SimpleNamespace(webdav=webdav_client, username="alice") - Per the contract change in webdav.py: ``None`` is now reserved for the - ambiguous "malformed XML" case. Real 404s raise HTTPStatusError. The - file verifier must NOT evict on the ambiguous case (we cannot tell - whether the file exists), only log a warning and keep the result. - """ - webdav_client = SimpleNamespace(get_file_info=mocker.AsyncMock(return_value=None)) - client = SimpleNamespace(webdav=webdav_client, username="alice") + result = await _verify_files( + client, + [_make_result(124, doc_type="file", metadata={"path": "x.txt"})], + _sem(), + ) - result = await _verify_files( - client, - [_make_result(123, doc_type="file", metadata={"path": "brittle.txt"})], - _sem(), - ) - - assert result == {"123"}, "ambiguous None must keep result, not evict" + assert result == set(), f"{status} on the SEARCH call must drop" @pytest.mark.unit -async def test_verify_files_403_drops(mocker): - """get_file_info raising HTTPStatusError(403) is a definitive drop.""" +async def test_verify_files_non_numeric_id_keeps_unverified(mocker): + """Without a numeric file id we cannot verify — fail open, don't drop.""" webdav_client = SimpleNamespace( - get_file_info=mocker.AsyncMock(side_effect=_http_error(403)) + file_accessible_by_id=mocker.AsyncMock( + side_effect=AssertionError("must not be called") + ) ) client = SimpleNamespace(webdav=webdav_client, username="alice") result = await _verify_files( client, - [_make_result(124, doc_type="file", metadata={"path": "forbidden.txt"})], + [_make_result("not-a-file-id", doc_type="file", metadata={"path": "x.txt"})], _sem(), ) - - assert result == set() - - -@pytest.mark.unit -async def test_verify_files_missing_path_metadata_keeps_unverified(mocker): - """Without a path in metadata we cannot verify — fail open, don't drop.""" - webdav_client = SimpleNamespace( - get_file_info=mocker.AsyncMock(side_effect=AssertionError("must not be called")) - ) - client = SimpleNamespace(webdav=webdav_client, username="alice") - - # No metadata at all - result = await _verify_files(client, [_make_result(555, doc_type="file")], _sem()) - assert result == {"555"} - webdav_client.get_file_info.assert_not_awaited() - - # Metadata present but no "path" key - result = await _verify_files( - client, [_make_result(556, doc_type="file", metadata={})], _sem() - ) - assert result == {"556"} - webdav_client.get_file_info.assert_not_awaited() + assert result == {"not-a-file-id"} + webdav_client.file_accessible_by_id.assert_not_awaited() @pytest.mark.unit async def test_verify_files_transient_5xx_keeps(mocker): webdav_client = SimpleNamespace( - get_file_info=mocker.AsyncMock(side_effect=_http_error(503)) + file_accessible_by_id=mocker.AsyncMock(side_effect=_http_error(503)) ) client = SimpleNamespace(webdav=webdav_client, username="alice") @@ -549,9 +537,9 @@ async def test_verify_files_transient_5xx_keeps(mocker): @pytest.mark.unit async def test_verify_files_429_keeps_as_transient(mocker): - """HTTP 429 from get_file_info must NOT silently drop file results.""" + """HTTP 429 from the SEARCH call must NOT silently drop file results.""" webdav_client = SimpleNamespace( - get_file_info=mocker.AsyncMock(side_effect=_http_error(429)) + file_accessible_by_id=mocker.AsyncMock(side_effect=_http_error(429)) ) client = SimpleNamespace(webdav=webdav_client, username="alice") @@ -566,14 +554,14 @@ async def test_verify_files_429_keeps_as_transient(mocker): @pytest.mark.unit async def test_verify_files_unexpected_exception_keeps(mocker): - """A non-HTTP exception from get_file_info must not drop the result. + """A non-HTTP exception from file_accessible_by_id must not drop the result. The catch-all ``except Exception`` branch in the file verifier exists so a bug in the WebDAV client (or an httpx ConnectError on a flaky network) cannot silently shrink result pages. """ webdav_client = SimpleNamespace( - get_file_info=mocker.AsyncMock(side_effect=RuntimeError("dav blew up")) + file_accessible_by_id=mocker.AsyncMock(side_effect=RuntimeError("dav blew up")) ) client = SimpleNamespace(webdav=webdav_client, username="alice")