test(integration): address round-1 review — unify searchability helper

- Extract the duplicated `_document_is_searchable`/`_note_is_searchable`
  helpers into a shared, Playwright-free `tests/integration/_search_helpers.py`
  (`document_is_searchable`), used by both the plotly and sampling tests.
- Resolve the sampling Medium finding: `wait_for_vector_sync` now triggers the
  searchability path on `search_term` alone (matching the plotly variant)
  instead of requiring both `search_term` and `note_id`, removing the silent
  fall-through to the unreliable gauge-delta path.
- Tighten `_get_with_retry`'s `last_exc` annotation to `httpx.TransportError`.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/integration/_search_helpers.py

@@ -0,0 +1,43 @@
+"""Shared helpers for asserting vector-sync visibility in integration tests.
+
+Kept dependency-light (no Playwright) so both the multi-user-basic UI tests and
+the single-user sampling tests can import it.
+"""
+
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def document_is_searchable(
+    mcp_client, search_term: str, note_id: int | None = None
+) -> bool:
+    """Return True once a freshly-created document is retrievable.
+
+    Polls ``nc_semantic_search`` (hybrid: an exact unique term reliably matches
+    on the keyword side) and matches by ``note_id`` when provided, otherwise by
+    the term appearing in a result's title/excerpt. Transient errors return
+    False so callers can keep polling.
+    """
+    try:
+        search = await mcp_client.call_tool(
+            "nc_semantic_search",
+            arguments={"query": search_term, "limit": 10, "score_threshold": 0.0},
+        )
+    except Exception as e:  # transient transport/availability blip — keep polling
+        logger.debug("Semantic search poll failed: %s", e)
+        return False
+    if search.isError:
+        logger.debug("Semantic search poll error: %s", search)
+        return False
+
+    results = json.loads(search.content[0].text).get("results", [])
+    needle = search_term.lower()
+    for r in results:
+        if note_id is not None:
+            if r.get("id") == note_id and r.get("doc_type") == "note":
+                return True
+        elif needle in f"{r.get('title', '')} {r.get('excerpt', '')}".lower():
+            return True
+    return False

tests/integration/test_astrolabe_plotly_visualization.py

@@ -28,6 +28,7 @@ from playwright.async_api import Page
 
 # Import helper functions from existing test
 from tests.conftest import create_mcp_client_session
+from tests.integration._search_helpers import document_is_searchable
 from tests.integration.test_astrolabe_multi_user_background_sync import (
     complete_astrolabe_authorization,
     login_to_nextcloud,
@@ -38,38 +39,6 @@ logger = logging.getLogger(__name__)
 pytestmark = [pytest.mark.integration, pytest.mark.multi_user_basic]
 
 
-async def _document_is_searchable(
-    mcp_client, search_term: str, note_id: int | None
-) -> bool:
-    """Return True once the freshly-created document is retrievable.
-
-    Polls ``nc_semantic_search`` (hybrid: an exact unique term reliably matches
-    on the keyword side) and matches by ``note_id`` when known, otherwise by the
-    term appearing in a result's title/excerpt.
-    """
-    try:
-        search = await mcp_client.call_tool(
-            "nc_semantic_search",
-            {"query": search_term, "limit": 10, "score_threshold": 0.0},
-        )
-    except Exception as e:  # transient transport/availability blip — keep polling
-        logger.debug("Semantic search poll failed: %s", e)
-        return False
-    if search.isError:
-        logger.debug("Semantic search poll error: %s", search)
-        return False
-
-    results = json.loads(search.content[0].text).get("results", [])
-    needle = search_term.lower()
-    for r in results:
-        if note_id is not None:
-            if r.get("id") == note_id and r.get("doc_type") == "note":
-                return True
-        elif needle in f"{r.get('title', '')} {r.get('excerpt', '')}".lower():
-            return True
-    return False
-
-
 async def wait_for_vector_sync(
     mcp_client,
     initial_indexed_count: int,
@@ -132,7 +101,7 @@ async def wait_for_vector_sync(
         )
 
         if search_term is not None:
-            if await _document_is_searchable(mcp_client, search_term, note_id):
+            if await document_is_searchable(mcp_client, search_term, note_id):
                 logger.info(
                     "✓ Sync complete: document %s retrievable via semantic search",
                     note_id,

tests/integration/test_astrolabe_session_jwt_search.py

@@ -47,7 +47,7 @@ async def _get_with_retry(
     client: httpx.AsyncClient, url: str, *, retries: int = 2, **kwargs
 ) -> httpx.Response:
     """GET with retries on transient transport errors (timeouts/conn resets)."""
-    last_exc: Exception | None = None
+    last_exc: httpx.TransportError | None = None
     for attempt in range(retries + 1):
         try:
             return await client.get(url, **kwargs)

tests/integration/test_sampling.py

@@ -14,33 +14,16 @@ vector database with indexed test data.
 """
 
 import json
-import logging
 from unittest.mock import MagicMock
 
 import anyio
 import pytest
 from mcp.types import CreateMessageResult, TextContent
 
+from tests.integration._search_helpers import document_is_searchable
+
 pytestmark = pytest.mark.integration
 
-logger = logging.getLogger(__name__)
-
-
-async def _note_is_searchable(nc_mcp_client, search_term: str, note_id: int) -> bool:
-    """Return True once ``note_id`` is retrievable via semantic search."""
-    try:
-        search = await nc_mcp_client.call_tool(
-            "nc_semantic_search",
-            arguments={"query": search_term, "limit": 10, "score_threshold": 0.0},
-        )
-    except Exception as e:  # transient blip — keep polling
-        logger.debug("Semantic search poll failed: %s", e)
-        return False
-    if search.isError:
-        return False
-    results = json.loads(search.content[0].text).get("results", [])
-    return any(r.get("id") == note_id and r.get("doc_type") == "note" for r in results)
-
 
 async def wait_for_vector_sync(
     nc_mcp_client,
@@ -55,11 +38,12 @@ async def wait_for_vector_sync(
 
     Args:
         nc_mcp_client: MCP client to poll status with.
-        search_term/note_id: If set (preferred), wait until that specific
-            document is retrievable via ``nc_semantic_search``. This is robust
-            against full-corpus re-scan churn, where the corpus-wide
-            ``indexed_count`` gauge is non-monotonic and ``indexed_count >
-            initial`` can never hold even though the document is indexed.
+        search_term: If set (preferred), wait until a document matching this
+            term is retrievable via ``nc_semantic_search``. Robust against
+            full-corpus re-scan churn, where the corpus-wide ``indexed_count``
+            gauge is non-monotonic and ``indexed_count > initial`` can never
+            hold even though the document is indexed.
+        note_id: Optional exact-match document id paired with ``search_term``.
         initial_indexed_count: Legacy gauge-delta fallback when no search_term
             is given: wait until indexed_count exceeds this value and
             pending_count reaches 0.
@@ -77,9 +61,9 @@ async def wait_for_vector_sync(
         )
         status_data = json.loads(sync_status.content[0].text)
 
-        if search_term is not None and note_id is not None:
+        if search_term is not None:
             # Robust signal: wait for the specific document to be retrievable
-            if await _note_is_searchable(nc_mcp_client, search_term, note_id):
+            if await document_is_searchable(nc_mcp_client, search_term, note_id):
                 break
         elif initial_indexed_count is not None:
             # Legacy: wait for new document(s) to be indexed (gauge delta)