Deduplicate app.py's search endpoints (item 4 from review)

Six of the seven /serp and /ops search endpoints repeated the identical
gather -> log -> filter -> flatten -> shape-a-SerpResults pattern verbatim,
differing only in which query_* function they called. Extracted into two
helpers:

- _shape_serp_results(results): flatten a gather's per-query results (or
Exceptions), surfacing the last error only when every query failed.
- _run_serp_queries(fn, params, context): run fn concurrently across every
query, log failures, and shape via _shape_serp_results.

search_patents keeps its distinct OPS-fallback loop (genuinely different
logic, not boilerplate) but now reuses _shape_serp_results for its tail.

app.py had zero test coverage before this - added tests/test_app.py first
(characterizing the endpoints' current behavior via ASGITransport, with
the query_*/ops_search functions monkeypatched rather than hitting real
search engines or needing a real Playwright browser) and confirmed those
pass against the pre-refactor code, then did the extraction under that
safety net. All 80 tests pass; app.py is 385 -> 339 lines.

app.py

@@ -104,40 +104,46 @@ ops_router = APIRouter(prefix="/ops", tags=["EPO OPS"])
 # ===================== Search endpoints =====================
 
 
-@serp_router.post("/search_scholar")
-async def search_google_scholar(params: SerpQuery):
-    """Queries google scholar for the specified query"""
-    logging.info(f"Searching Google Scholar for queries: {params.queries}")
-    results = await asyncio.gather(*[query_google_scholar(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, "google scholar search", params)
-
-    # Filter out exceptions and flatten the results
+def _shape_serp_results(results: list) -> SerpResults:
+    """Flatten a list of per-query result-lists (or Exceptions, from a
+    `return_exceptions=True` gather) into one SerpResults, surfacing the
+    last error only when every query failed.
+    """
     filtered_results = [r for r in results if not isinstance(r, Exception)]
     flattened_results = [
         item for sublist in filtered_results for item in sublist]
 
-    # all queries failed, return the last exception
     if len(filtered_results) == 0:
         return SerpResults(results=[], error=str(results[-1]))
 
     return SerpResults(results=flattened_results, error=None)
 
 
+async def _run_serp_queries(fn, params: SerpQuery, context: str) -> SerpResults:
+    """Run `fn(query, n_results)` concurrently for every query in `params`,
+    log any failures, and shape the result the way every simple /serp and
+    /ops search endpoint below expects.
+    """
+    results = await asyncio.gather(
+        *[fn(q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, context, params)
+    return _shape_serp_results(results)
+
+
+@serp_router.post("/search_scholar")
+async def search_google_scholar(params: SerpQuery):
+    """Queries google scholar for the specified query"""
+    logging.info(f"Searching Google Scholar for queries: {params.queries}")
+    return await _run_serp_queries(
+        lambda q, n: query_google_scholar(pw_browser, q, n), params, "google scholar search")
+
+
 @serp_router.post("/search_arxiv")
 async def search_arxiv(params: SerpQuery):
     """Searches arxiv for the specified queries and returns the found documents."""
     logging.info(f"Searching Arxiv for queries: {params.queries}")
-    results = await asyncio.gather(*[query_arxiv(httpx_client, q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, "arxiv search", params)
-
-    filtered_results = [r for r in results if not isinstance(r, Exception)]
-    flattened_results = [
-        item for sublist in filtered_results for item in sublist]
-
-    if len(filtered_results) == 0:
-        return SerpResults(results=[], error=str(results[-1]))
-
-    return SerpResults(results=flattened_results, error=None)
+    return await _run_serp_queries(
+        lambda q, n: query_arxiv(httpx_client, q, n), params, "arxiv search")
 
 
 @serp_router.post("/search_patents")
@@ -162,73 +168,31 @@ async def search_patents(params: SerpQuery) -> SerpResults:
                 except Exception as e:
                     logging.warning(f"OPS fallback failed: {e}")
 
-    # Filter out exceptions and flatten the results
-    filtered_results = [r for r in results if not isinstance(r, Exception)]
-    flattened_results = [
-        item for sublist in filtered_results for item in sublist]
-
-    # all queries failed, return the last exception
-    if len(filtered_results) == 0:
-        return SerpResults(results=[], error=str(results[-1]))
-
-    return SerpResults(results=flattened_results, error=None)
+    return _shape_serp_results(results)
 
 
 @serp_router.post("/search_brave")
 async def search_brave(params: SerpQuery) -> SerpResults:
     """Searches brave search for the specified queries and returns the found documents."""
     logging.info(f"Searching Brave Search for queries: {params.queries}")
-    results = await asyncio.gather(*[query_brave_search(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, "brave search", params)
-
-    # Filter out exceptions and flatten the results
-    filtered_results = [r for r in results if not isinstance(r, Exception)]
-    flattened_results = [
-        item for sublist in filtered_results for item in sublist]
-
-    # all queries failed, return the last exception
-    if len(filtered_results) == 0:
-        return SerpResults(results=[], error=str(results[-1]))
-
-    return SerpResults(results=flattened_results, error=None)
+    return await _run_serp_queries(
+        lambda q, n: query_brave_search(pw_browser, q, n), params, "brave search")
 
 
 @serp_router.post("/search_bing")
 async def search_bing(params: SerpQuery) -> SerpResults:
     """Searches Bing search for the specified queries and returns the found documents."""
     logging.info(f"Searching Bing Search for queries: {params.queries}")
-    results = await asyncio.gather(*[query_bing_search(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, "bing search", params)
-
-    # Filter out exceptions and flatten the results
-    filtered_results = [r for r in results if not isinstance(r, Exception)]
-    flattened_results = [
-        item for sublist in filtered_results for item in sublist]
-
-    # all queries failed, return the last exception
-    if len(filtered_results) == 0:
-        return SerpResults(results=[], error=str(results[-1]))
-
-    return SerpResults(results=flattened_results, error=None)
+    return await _run_serp_queries(
+        lambda q, n: query_bing_search(pw_browser, q, n), params, "bing search")
 
 
 @serp_router.post("/search_duck")
 async def search_duck(params: SerpQuery) -> SerpResults:
     """Searches duckduckgo for the specified queries and returns the found documents"""
     logging.info(f"Searching DuckDuckGo for queries: {params.queries}")
-    results = await asyncio.gather(*[query_ddg_search(q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, "duckduckgo search", params)
-
-    # Filter out exceptions and flatten the results
-    filtered_results = [r for r in results if not isinstance(r, Exception)]
-    flattened_results = [
-        item for sublist in filtered_results for item in sublist]
-
-    # all queries failed, return the last exception
-    if len(filtered_results) == 0:
-        return SerpResults(results=[], error=str(results[-1]))
-
-    return SerpResults(results=flattened_results, error=None)
+    return await _run_serp_queries(
+        lambda q, n: query_ddg_search(q, n), params, "duckduckgo search")
 
 
 async def _search_one(q: str, n_results: int) -> tuple[str, list[dict], Optional[str]]:
@@ -327,17 +291,8 @@ async def scrap_patents(params: ScrapPatentsRequest) -> PatentScrapBulkResponse:
 async def ops_keyword_search(params: SerpQuery) -> SerpResults:
     """Keyword-searches patents via the official EPO OPS API."""
     logging.info(f"Searching EPO OPS for queries: {params.queries}")
-    results = await asyncio.gather(*[ops_search(httpx_client, q, params.n_results) for q in params.queries], return_exceptions=True)
-    log_gathered_exceptions(results, "OPS search", params)
-
-    filtered_results = [r for r in results if not isinstance(r, Exception)]
-    flattened_results = [
-        item for sublist in filtered_results for item in sublist]
-
-    if len(filtered_results) == 0:
-        return SerpResults(results=[], error=str(results[-1]))
-
-    return SerpResults(results=flattened_results, error=None)
+    return await _run_serp_queries(
+        lambda q, n: ops_search(httpx_client, q, n), params, "OPS search")
 
 
 @ops_router.get("/scrap_patent/{patent_id}")

tests/test_app.py

@@ -0,0 +1,159 @@
+"""Characterization tests for app.py's search endpoints.
+
+These monkeypatch the underlying query_*/ops_search functions rather than
+hitting real search engines - app.py's own job here is just wiring
+(gather queries concurrently, log failures, flatten results, decide when to
+report an error), which is exactly what's being tested. The app's lifespan
+(which starts a real Playwright browser) is never triggered: ASGITransport
+doesn't run it, and these tests don't need it since pw_browser is never
+actually used once the query functions are patched out.
+"""
+
+import httpx
+import pytest
+from httpx import ASGITransport
+
+import app as app_module
+from serp import SerpQuery
+
+
+@pytest.fixture
+async def client():
+    transport = ASGITransport(app=app_module.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+# ------------------------- _shape_serp_results / _run_serp_queries -------------------------
+
+
+def test_shape_serp_results_flattens_successful_lists():
+    result = app_module._shape_serp_results([[{"a": 1}], [{"b": 2}, {"c": 3}]])
+    assert result.results == [{"a": 1}, {"b": 2}, {"c": 3}]
+    assert result.error is None
+
+
+def test_shape_serp_results_surfaces_last_error_when_all_failed():
+    result = app_module._shape_serp_results([ValueError("first"), RuntimeError("second")])
+    assert result.results == []
+    assert "second" in result.error
+
+
+def test_shape_serp_results_ignores_failures_when_some_succeed():
+    result = app_module._shape_serp_results([[{"a": 1}], RuntimeError("boom")])
+    assert result.results == [{"a": 1}]
+    assert result.error is None
+
+
+async def test_run_serp_queries_calls_fn_per_query_and_shapes_result():
+    params = SerpQuery(queries=["x", "y"], n_results=5)
+    calls = []
+
+    async def fn(q, n):
+        calls.append((q, n))
+        return [{"q": q}]
+
+    result = await app_module._run_serp_queries(fn, params, "test context")
+
+    assert sorted(calls) == [("x", 5), ("y", 5)]
+    assert sorted(result.results, key=str) == sorted([{"q": "x"}, {"q": "y"}], key=str)
+
+
+# ------------------------------------- endpoints -------------------------------------
+
+
+async def test_search_arxiv_flattens_results_from_all_queries(client, monkeypatch):
+    async def fake_query_arxiv(client_arg, q, n):
+        return [{"title": f"paper about {q}", "href": "x", "body": "y", "id": "1"}]
+
+    monkeypatch.setattr(app_module, "query_arxiv", fake_query_arxiv)
+
+    resp = await client.post("/serp/search_arxiv", json={"queries": ["a", "b"], "n_results": 5})
+
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["error"] is None
+    assert {r["title"] for r in data["results"]} == {"paper about a", "paper about b"}
+
+
+async def test_search_arxiv_all_queries_failing_returns_error(client, monkeypatch):
+    async def failing_query_arxiv(client_arg, q, n):
+        raise RuntimeError(f"boom for {q}")
+
+    monkeypatch.setattr(app_module, "query_arxiv", failing_query_arxiv)
+
+    resp = await client.post("/serp/search_arxiv", json={"queries": ["a"], "n_results": 5})
+
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["results"] == []
+    assert "boom for a" in data["error"]
+
+
+async def test_search_patents_does_not_fall_back_when_google_patents_succeeds(client, monkeypatch):
+    async def fake_google_patents(browser, q, n):
+        return [{"id": "US1", "href": "h", "title": "t", "body": "b"}]
+
+    ops_search_called = False
+
+    async def fake_ops_search(client_arg, q, n):
+        nonlocal ops_search_called
+        ops_search_called = True
+        return []
+
+    monkeypatch.setattr(app_module, "query_google_patents", fake_google_patents)
+    monkeypatch.setattr(app_module, "ops_search", fake_ops_search)
+    monkeypatch.setattr(app_module.ops_token_manager, "_key", "k")
+    monkeypatch.setattr(app_module.ops_token_manager, "_secret", "s")
+
+    resp = await client.post("/serp/search_patents", json={"queries": ["widget"], "n_results": 5})
+
+    data = resp.json()
+    assert data["results"] == [{"id": "US1", "href": "h", "title": "t", "body": "b"}]
+    assert ops_search_called is False
+
+
+async def test_search_patents_falls_back_to_ops_when_google_patents_empty(client, monkeypatch):
+    async def fake_google_patents(browser, q, n):
+        return []
+
+    async def fake_ops_search(client_arg, q, n):
+        return [{"title": "OPS result", "body": "", "href": "", "id": "US123"}]
+
+    monkeypatch.setattr(app_module, "query_google_patents", fake_google_patents)
+    monkeypatch.setattr(app_module, "ops_search", fake_ops_search)
+    monkeypatch.setattr(app_module.ops_token_manager, "_key", "k")
+    monkeypatch.setattr(app_module.ops_token_manager, "_secret", "s")
+
+    resp = await client.post("/serp/search_patents", json={"queries": ["widget"], "n_results": 5})
+
+    data = resp.json()
+    assert data["results"] == [{"title": "OPS result", "body": "", "href": "", "id": "US123"}]
+
+
+async def test_search_patents_no_fallback_when_ops_not_configured(client, monkeypatch):
+    async def fake_google_patents(browser, q, n):
+        return []
+
+    monkeypatch.setattr(app_module, "query_google_patents", fake_google_patents)
+    monkeypatch.setattr(app_module.ops_token_manager, "_key", None)
+    monkeypatch.setattr(app_module.ops_token_manager, "_secret", None)
+
+    resp = await client.post("/serp/search_patents", json={"queries": ["widget"], "n_results": 5})
+
+    data = resp.json()
+    assert data["results"] == []
+    assert data["error"] is None
+
+
+async def test_ops_keyword_search_flattens_results(client, monkeypatch):
+    async def fake_ops_search(client_arg, q, n):
+        return [{"title": f"OPS {q}", "body": "", "href": "", "id": "1"}]
+
+    monkeypatch.setattr(app_module, "ops_search", fake_ops_search)
+
+    resp = await client.post("/ops/search", json={"queries": ["a", "b"], "n_results": 5})
+
+    data = resp.json()
+    assert data["error"] is None
+    assert {r["title"] for r in data["results"]} == {"OPS a", "OPS b"}