test(integration): address round-4 review — type hints & small robustness
- Type the new helper signatures (CLAUDE.md A5): `mcp_client: Any` in document_is_searchable and `nc_mcp_client: Any` in _top_score. - _top_score: guard the results list directly (`if not results`) instead of via total_found, so max() can't hit an empty sequence. - _get_with_retry: replace `raise last_exc # type: ignore` with an explicit `assert last_exc is not None` then raise — clearer intent, no suppressor. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
367afa0402
commit
7c13c6e49a
@@ -6,12 +6,13 @@ the single-user sampling tests can import it.
|
|||||||
|
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
async def document_is_searchable(
|
async def document_is_searchable(
|
||||||
mcp_client, search_term: str, note_id: int | None = None
|
mcp_client: Any, search_term: str, note_id: int | None = None
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""Return True once a freshly-created document is retrievable.
|
"""Return True once a freshly-created document is retrievable.
|
||||||
|
|
||||||
|
|||||||
@@ -57,7 +57,8 @@ async def _get_with_retry(
|
|||||||
"GET %s failed (attempt %s/%s): %s", url, attempt, max_attempts, e
|
"GET %s failed (attempt %s/%s): %s", url, attempt, max_attempts, e
|
||||||
)
|
)
|
||||||
await anyio.sleep(2)
|
await anyio.sleep(2)
|
||||||
raise last_exc # type: ignore[misc]
|
assert last_exc is not None # loop ran at least once, so this is set
|
||||||
|
raise last_exc
|
||||||
|
|
||||||
|
|
||||||
async def _astrolabe_configured(client: httpx.AsyncClient, auth) -> bool:
|
async def _astrolabe_configured(client: httpx.AsyncClient, auth) -> bool:
|
||||||
|
|||||||
@@ -405,7 +405,7 @@ async def test_retrieval_quality_all_queries(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
async def _top_score(nc_mcp_client, query: str) -> float | None:
|
async def _top_score(nc_mcp_client: Any, query: str) -> float | None:
|
||||||
"""Return the best fusion score for ``query``, or None if no results."""
|
"""Return the best fusion score for ``query``, or None if no results."""
|
||||||
result = await nc_mcp_client.call_tool(
|
result = await nc_mcp_client.call_tool(
|
||||||
"nc_semantic_search",
|
"nc_semantic_search",
|
||||||
@@ -413,9 +413,10 @@ async def _top_score(nc_mcp_client, query: str) -> float | None:
|
|||||||
)
|
)
|
||||||
assert result.isError is False
|
assert result.isError is False
|
||||||
data = json.loads(result.content[0].text)
|
data = json.loads(result.content[0].text)
|
||||||
if data["total_found"] == 0:
|
results = data.get("results", [])
|
||||||
|
if not results: # guard the list directly, not via total_found
|
||||||
return None
|
return None
|
||||||
return max(r["score"] for r in data["results"])
|
return max(r["score"] for r in results)
|
||||||
|
|
||||||
|
|
||||||
async def test_no_results_for_unrelated_query(nc_mcp_client, indexed_manual_pdf):
|
async def test_no_results_for_unrelated_query(nc_mcp_client, indexed_manual_pdf):
|
||||||
|
|||||||
Reference in New Issue
Block a user