test(vector): address PR #873 round-1 review

- Extract `_app_enabled` to a module-level helper so the gate predicate is
  unit-tested directly instead of via an inline copy that could drift.
- Move `import logging` to module scope in test_scanner_app_gating.py.
- Harden `get_enabled_apps` OCS-envelope parsing (`X or {}` / `or []`) so a
  present-but-null `ocs`/`data` coerces to empty instead of raising on
  `None.get`; add parametrized malformed-envelope tests.
- Use https:// in the test request URL (SonarCloud S5332 hotspot).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-07 20:04:00 +02:00
co-authored by Claude Opus 4.8
parent 2e609cbea7
commit b11d1b17a3
4 changed files with 51 additions and 23 deletions
+5 -1
View File
@@ -214,7 +214,11 @@ class NextcloudClient:
)
response.raise_for_status()
data = response.json()
entries = data.get("ocs", {}).get("data", []) or []
# ``X or {}``/``or []`` (not ``.get(k, default)``) so a present-but-null
# ``ocs``/``data`` (``{"ocs": null}``) coerces to empty instead of
# raising AttributeError on ``None.get``.
ocs = data.get("ocs") or {}
entries = ocs.get("data") or []
enabled: set[str] = set()
for entry in entries:
# ``app`` is the canonical app id; ``id`` matches it for the apps we
+13 -6
View File
@@ -269,6 +269,16 @@ async def _get_enabled_apps_or_none(
return None
def _app_enabled(app_id: str, enabled_apps: set[str] | None) -> bool:
"""Whether ``app_id`` should be scanned for the current user.
``enabled_apps is None`` means detection failed — every app is treated as
enabled (the scan-all fallback) so a transient navigation-endpoint failure
never silently halts indexing.
"""
return enabled_apps is None or app_id in enabled_apps
async def scan_user_documents(
user_id: str,
send_stream: TaskProducer,
@@ -355,9 +365,6 @@ async def scan_user_documents(
# detection failed: fall back to scanning every app (prior behaviour).
enabled_apps = await _get_enabled_apps_or_none(nc_client, user_id, scan_id)
def _app_enabled(app_id: str) -> bool:
return enabled_apps is None or app_id in enabled_apps
# Notes (isolated so an uninstalled or disabled Notes app — whose API
# returns 404 — cannot abort scanning of the other apps; this mirrors the
# per-app try/except guards already wrapping files/news/deck below).
@@ -366,7 +373,7 @@ async def scan_user_documents(
current_time = time.time()
queued = 0
if _app_enabled("notes"):
if _app_enabled("notes", enabled_apps):
try:
queued += await scan_notes(
user_id=user_id,
@@ -703,7 +710,7 @@ async def scan_user_documents(
# Scan News items (starred + unread)
news_queued = 0
if _app_enabled("news"):
if _app_enabled("news", enabled_apps):
try:
news_queued = await scan_news_items(
user_id=user_id,
@@ -724,7 +731,7 @@ async def scan_user_documents(
# Scan Deck cards
deck_queued = 0
if _app_enabled("deck"):
if _app_enabled("deck", enabled_apps):
try:
deck_queued = await scan_deck_cards(
user_id=user_id,