test(integration): address round-5 review — parse safety & timeout headroom

- _search_helpers: wrap the json.loads(search.content[0].text) parse in
  try/except (IndexError, ValueError) so empty content / malformed JSON returns
  False (keep polling) instead of escaping as a confusing traceback. Also debug-
  log an id match with a non-note doc_type to surface schema drift instead of
  silently timing out.
- test_astrolabe_session_jwt_search: drop _get_with_retry default to
  max_attempts=2 (matches the "one retry" intent) and mark both search tests
  @pytest.mark.timeout(300) so a cold model load + retry can't breach the 180s
  default pytest timeout.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-17 23:08:59 +02:00
co-authored by Claude Opus 4.8
parent 7c13c6e49a
commit 829625f2a2
2 changed files with 19 additions and 4 deletions
+16 -3
View File
@@ -35,14 +35,27 @@ async def document_is_searchable(
logger.debug("Semantic search poll error: %s", search)
return False
results = json.loads(search.content[0].text).get("results", [])
try:
results = json.loads(search.content[0].text).get("results", [])
except (IndexError, ValueError) as e: # empty content / malformed JSON
logger.debug("Semantic search parse failed: %s", e)
return False
# Token match (not contiguous substring) so multi-word terms work in the
# note_id-less fallback path.
tokens = search_term.lower().split()
for r in results:
if note_id is not None:
if r.get("id") == note_id and r.get("doc_type") == "note":
return True
if r.get("id") == note_id:
if r.get("doc_type") == "note":
return True
# id matched but not a note — surface possible schema drift
# rather than silently timing out.
logger.debug(
"search hit id=%s has doc_type=%s (expected note)",
note_id,
r.get("doc_type"),
)
else:
haystack = f"{r.get('title', '')} {r.get('excerpt', '')}".lower()
if tokens and all(t in haystack for t in tokens):
@@ -44,7 +44,7 @@ _SEARCH_TIMEOUT = httpx.Timeout(90.0)
async def _get_with_retry(
client: httpx.AsyncClient, url: str, *, max_attempts: int = 3, **kwargs
client: httpx.AsyncClient, url: str, *, max_attempts: int = 2, **kwargs
) -> httpx.Response:
"""GET, retrying on transient transport errors (timeouts/conn resets)."""
last_exc: httpx.TransportError | None = None
@@ -74,6 +74,7 @@ async def _astrolabe_configured(client: httpx.AsyncClient, auth) -> bool:
return bool(resp.json().get("success"))
@pytest.mark.timeout(300) # cold model load + retry can exceed the 180s default
async def test_session_user_searches_without_provisioning(test_users_setup):
"""A non-admin session user searches with no OAuth/provisioning step.
@@ -111,6 +112,7 @@ async def test_session_user_searches_without_provisioning(test_users_setup):
assert "results" in body and "algorithm_used" in body
@pytest.mark.timeout(300) # cold model load + retry can exceed the 180s default
async def test_admin_session_search_succeeds():
"""The same JWT-mint path works for the admin session user."""
admin_pw = os.environ["NEXTCLOUD_PASSWORD"]