diff --git a/tests/integration/_search_helpers.py b/tests/integration/_search_helpers.py new file mode 100644 index 00000000..db531c7d --- /dev/null +++ b/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 diff --git a/tests/integration/test_astrolabe_plotly_visualization.py b/tests/integration/test_astrolabe_plotly_visualization.py index 88ae2b5c..42fa6edc 100644 --- a/tests/integration/test_astrolabe_plotly_visualization.py +++ b/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, diff --git a/tests/integration/test_astrolabe_session_jwt_search.py b/tests/integration/test_astrolabe_session_jwt_search.py index 8e0885f3..6ad6bcf2 100644 --- a/tests/integration/test_astrolabe_session_jwt_search.py +++ b/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) diff --git a/tests/integration/test_sampling.py b/tests/integration/test_sampling.py index da8978fc..293e5dc7 100644 --- a/tests/integration/test_sampling.py +++ b/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)