admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(search): verify shared files by global file id (ACL-aware)

Browse Source 
The ACL-aware vector filter (PR #813) expands a user's search to documents
whose owner shared them, but verify-on-read still re-checked each file by
PATH under the *searching* user's WebDAV root. Nextcloud mounts received
shares at the recipient's root by basename, so a nested shared file (e.g.
owner's /docs/report.pdf) 404s for the recipient and was silently dropped —
defeating the filter for everything but root-level files.

Verify files by their global Nextcloud file id instead (the file doc_id IS
that id): WebDAVClient.get_file_info_by_id was insufficient (the dav/meta
endpoint only resolves the user's own storage, not shares), so add
WebDAVClient.file_accessible_by_id which runs a WebDAV SEARCH over the user's
whole tree (incl. mounted shares) filtered on oc:fileid. Empirically this
resolves owned, directly-shared, and folder-shared files; an empty result is
a definitive drop, transport errors are kept as transient.

- search/verification.py: _verify_files now checks file_accessible_by_id.
- client/webdav.py: add file_accessible_by_id (SEARCH by fileid).
- tests/integration/test_acl_owner_filter.py: filter matrix vs real Qdrant.
- tests/integration/test_acl_shared_search.py: real-Nextcloud share -> search.
- tests/integration/test_verify_on_read.py: nested shared file kept for the
  recipient; unshared file dropped.
- tests/unit/search/test_verification.py: id-based verifier semantics.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-05-28 23:05:40 +02:00
co-authored by Claude Opus 4.8
parent dc0653c415
commit bf35200bab
8 changed files with 586 additions and 920 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/client/webdav.py
+43
View File
@@ -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 = (
"<d:eq><d:prop><oc:fileid/></d:prop>"
f"<d:literal>{int(file_id)}</d:literal></d:eq>"
)
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.
nextcloud_mcp_server/search/verification.py
+25 -23
View File
@@ -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,
tests/integration/test_acl_owner_filter.py
+173
View File
@@ -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
tests/integration/test_acl_shared_search.py
+186
View File
@@ -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"
tests/integration/test_astrolabe_login_flow_provisioning.py
-139
View File
@@ -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()
tests/integration/test_astrolabe_token_refresh.py
-702
View File
@@ -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)
tests/integration/test_verify_on_read.py
+116 -1
View File
@@ -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)
tests/unit/search/test_verification.py
+43 -55
View File
@@ -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")