Add test harness and fix TDD/architecture review findings (#3)

.github/workflows/deploy-to-hf.yml

@@ -7,7 +7,28 @@ on:
   workflow_dispatch:
 
 jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install dependencies
+        run: pip install -r requirements-dev.txt
+
+      - name: Install Playwright's Chromium
+        run: playwright install --with-deps chromium
+
+      - name: Run tests
+        run: pytest
+
   deploy:
+    needs: test
     runs-on: ubuntu-latest
     steps:
       - name: Checkout

README.md

@@ -13,3 +13,16 @@ pinned: true
 # `SERPent`
 
 `SERPent` provides a SERP / scrapping API for use by AI agents / projects.
+
+## Running tests
+
+```
+pip install -r requirements.txt -r requirements-dev.txt
+playwright install chromium   # if not already present on the machine
+pytest
+```
+
+The suite mocks all outbound HTTP (via `respx`) and drives the Playwright-based
+scrapers against local fixture HTML instead of live search engines - see
+`tests/helpers.py` for how - so it runs fully offline and doesn't depend on,
+or risk tripping, any site's anti-bot measures.

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}")

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_mode = auto
+testpaths = tests

requirements-dev.txt

@@ -0,0 +1,7 @@
+# Test-only dependencies. Install with:
+#   pip install -r requirements.txt -r requirements-dev.txt
+#   playwright install chromium   # if not already present on the machine
+-r requirements.txt
+pytest>=8,<10
+pytest-asyncio>=0.24
+respx>=0.21

scrap.py

@@ -38,7 +38,18 @@ async def scrap_patent_async(client: AsyncClient, patent_url: str) -> PatentScra
     response = await client.get(patent_url, headers=headers)
     response.raise_for_status()
 
-    soup = BeautifulSoup(response.text, "html.parser")
+    return parse_patent_html(response.text, patent_url)
+
+
+def parse_patent_html(html: str, patent_url: str) -> PatentScrapResult:
+    """Parse a Google Patents patent page into a PatentScrapResult.
+
+    Pure function of the page's HTML - no network I/O - so it can be tested
+    directly against saved/fixture HTML without mocking a client.
+    `patent_url` is only used to identify the page in the error message
+    below.
+    """
+    soup = BeautifulSoup(html, "html.parser")
 
     # Abstract
     abstract_div = soup.find("div", {"class": "abstract"})

serp.py

@@ -82,187 +82,193 @@ async def playwright_open_page(browser: Optional[Browser]):
             await context.close()
 
 
-async def query_google_scholar(browser: Browser, q: str, n_results: int = 10):
-    """Queries google scholar for the specified query and number of results. Returns relevant papers"""
+async def _block_stylesheet_and_image_resources(route, request):
+    """Shared `page.route("**/*", ...)` handler for every scraper below: skip
+    fetching stylesheets/images, since only the page's text content matters
+    here and this meaningfully speeds up navigation.
+    """
+    if request.resource_type in ["stylesheet", "image"]:
+        await route.abort()
+    else:
+        await route.continue_()
 
-    async with playwright_open_page(browser) as page:
 
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
+async def _extract_google_scholar_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Google Scholar results page."""
+    await page.wait_for_selector("div.gs_ri")
+
+    items = await page.locator("div.gs_ri").all()
+    results = []
+    for item in items[:n_results]:
+        title = await item.locator("h3").inner_text(timeout=1000)
+        body = await item.locator("div.gs_rs").inner_text(timeout=1000)
+        href = await item.locator("h3 > a").get_attribute("href")
+
+        results.append({
+            "title": title,
+            "body": body,
+            "href": href
+        })
 
-        await page.route("**/*", _block_resources)
+    return results
+
+
+async def query_google_scholar(browser: Browser, q: str, n_results: int = 10):
+    """Queries google scholar for the specified query and number of results. Returns relevant papers"""
+    async with playwright_open_page(browser) as page:
+        await page.route("**/*", _block_stylesheet_and_image_resources)
 
         url = f"https://scholar.google.com/scholar?q={quote_plus(q)}&num={n_results}"
         await page.goto(url)
 
-        await page.wait_for_selector("div.gs_ri")
+        return await _extract_google_scholar_results(page, n_results)
 
-        items = await page.locator("div.gs_ri").all()
-        results = []
-        for item in items[:n_results]:
-            title = await item.locator("h3").inner_text(timeout=1000)
-            body = await item.locator("div.gs_rs").inner_text(timeout=1000)
-            href = await item.locator("h3 > a").get_attribute("href")
 
-            results.append({
-                "title": title,
-                "body": body,
-                "href": href
-            })
+# regex to locate a patent id, e.g. "US11930446B2" or "EP4760514A1"
+PATENT_ID_REGEX = r"\b[A-Z]{2}\d{6,}(?:[A-Z]\d?)?\b"
 
-    return results
 
+async def _extract_google_patents_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Google Patents search page."""
+    # Wait for at least one search result item to appear
+    # This ensures the page has loaded enough to start scraping
+    await page.wait_for_function(
+        f"""() => document.querySelectorAll('search-result-item').length >= 1""",
+        timeout=30_000
+    )
 
-async def query_google_patents(browser: Browser, q: str, n_results: int = 10):
-    """Queries google patents for the specified query and number of results. Returns relevant patents"""
+    items = await page.locator("search-result-item").all()
+    results = []
+    for item in items:
+        text = " ".join(await item.locator("span").all_inner_texts())
+        match = re.search(PATENT_ID_REGEX, text)
+        if not match:
+            continue
 
-    # regex to locate a patent id
-    PATENT_ID_REGEX = r"\b[A-Z]{2}\d{6,}(?:[A-Z]\d?)?\b"
+        patent_id = match.group()
 
-    async with playwright_open_page(browser) as page:
+        try:
+            title = await item.locator("h3, h4").first.inner_text(timeout=1000)
+            body = await item.locator("div.abstract, div.result-snippet, .snippet, .result-text").first.inner_text(timeout=1000)
+        except Exception:
+            continue  # If we can't get title or body, skip this item
+
+        results.append({
+            "id": patent_id,
+            "href": f"https://patents.google.com/patent/{patent_id}/en",
+            "title": title,
+            "body": body
+        })
+
+    return results[:n_results]
 
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
 
-        await page.route("**/*", _block_resources)
+async def query_google_patents(browser: Browser, q: str, n_results: int = 10):
+    """Queries google patents for the specified query and number of results. Returns relevant patents"""
+    async with playwright_open_page(browser) as page:
+        await page.route("**/*", _block_stylesheet_and_image_resources)
 
         url = f"https://patents.google.com/?q={quote_plus(q)}&num={n_results}"
         await page.goto(url)
 
-        # Wait for at least one search result item to appear
-        # This ensures the page has loaded enough to start scraping
-        await page.wait_for_function(
-            f"""() => document.querySelectorAll('search-result-item').length >= 1""",
-            timeout=30_000
-        )
-
-        items = await page.locator("search-result-item").all()
-        results = []
-        for item in items:
-            text = " ".join(await item.locator("span").all_inner_texts())
-            match = re.search(PATENT_ID_REGEX, text)
-            if not match:
-                continue
-
-            patent_id = match.group()
-
-            try:
-                title = await item.locator("h3, h4").first.inner_text(timeout=1000)
-                body = await item.locator("div.abstract, div.result-snippet, .snippet, .result-text").first.inner_text(timeout=1000)
-            except:
-                continue  # If we can't get title or body, skip this item
+        return await _extract_google_patents_results(page, n_results)
 
-            results.append({
-                "id": patent_id,
-                "href": f"https://patents.google.com/patent/{patent_id}/en",
-                "title": title,
-                "body": body
-            })
 
-    return results[:n_results]
+async def _extract_brave_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Brave Search results page.
 
+    Raises BraveSearchBlockedException if the page looks like Brave's
+    anti-bot interstitial rather than a results page.
+    """
+    results_cards = await page.locator('.snippet').all()
 
-async def query_brave_search(browser: Browser, q: str, n_results: int = 10):
-    """Queries Brave Search for the specified query."""
+    if len(results_cards) == 0:
+        page_content = await page.content()
 
-    async with playwright_open_page(browser) as page:
+        if "suspicious" in page_content:
+            raise BraveSearchBlockedException()
 
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
+    results = []
 
-        await page.route("**/*", _block_resources)
+    for result in results_cards:
+        title = await result.locator('.title').all_inner_texts()
+        description = await result.locator('.snippet-description').all_inner_texts()
+        url = await result.locator('a').nth(0).get_attribute('href')
 
-        url = f"https://search.brave.com/search?q={quote_plus(q)}"
-        await page.goto(url)
+        # Filter out results with no URL or brave-specific URLs
+        if url is None or url.startswith('/'):
+            continue
 
-        results_cards = await page.locator('.snippet').all()
+        results.append({
+            "title": title[0] if title else "",
+            "body": description[0] if description else "",
+            "href": url
+        })
 
-        if len(results_cards) == 0:
-            page_content = await page.content()
+        if len(results) >= n_results:
+            break
 
-            if "suspicious" in page_content:
-                raise BraveSearchBlockedException()
+    return results
 
-        results = []
 
-        for result in results_cards:
-            title = await result.locator('.title').all_inner_texts()
-            description = await result.locator('.snippet-description').all_inner_texts()
-            url = await result.locator('a').nth(0).get_attribute('href')
+async def query_brave_search(browser: Browser, q: str, n_results: int = 10):
+    """Queries Brave Search for the specified query."""
+    async with playwright_open_page(browser) as page:
+        await page.route("**/*", _block_stylesheet_and_image_resources)
+
+        url = f"https://search.brave.com/search?q={quote_plus(q)}"
+        await page.goto(url)
+
+        return await _extract_brave_results(page, n_results)
 
-            # Filter out results with no URL or brave-specific URLs
-            if url is None or url.startswith('/'):
-                continue
 
+async def _extract_bing_results(page: Page, n_results: int) -> list[dict]:
+    """Extract results from an already-loaded Bing results page."""
+    await page.wait_for_selector("li.b_algo")
+
+    results = []
+
+    items = await page.query_selector_all("li.b_algo")
+    for item in items[:n_results]:
+        title_el = await item.query_selector("h2 > a")
+        url = await title_el.get_attribute("href") if title_el else None
+        title = await title_el.inner_text() if title_el else ""
+
+        snippet = ""
+
+        # Try several fallback selectors
+        for selector in [
+            "div.b_caption p",  # typical snippet
+            "div.b_caption",    # sometimes snippet is here
+            "div.b_snippet",    # used in some result types
+            "div.b_text",       # used in some panels
+            "p"                 # fallback to any paragraph
+        ]:
+            snippet_el = await item.query_selector(selector)
+            if snippet_el:
+                snippet = await snippet_el.inner_text()
+                if snippet.strip():
+                    break
+
+        if title and url:
             results.append({
-                "title": title[0] if title else "",
-                "body": description[0] if description else "",
-                "href": url
+                "title": title.strip(),
+                "href": url.strip(),
+                "body": snippet.strip()
             })
 
-            if len(results) >= n_results:
-                break
-
-        return results
+    return results
 
 
 async def query_bing_search(browser: Browser, q: str, n_results: int = 10):
     """Queries bing search for the specified query"""
     async with playwright_open_page(browser) as page:
-        async def _block_resources(route, request):
-            if request.resource_type in ["stylesheet", "image"]:
-                await route.abort()
-            else:
-                await route.continue_()
-
-        await page.route("**/*", _block_resources)
+        await page.route("**/*", _block_stylesheet_and_image_resources)
 
         url = f"https://www.bing.com/search?q={quote_plus(q)}"
         await page.goto(url)
 
-        await page.wait_for_selector("li.b_algo")
-
-        results = []
-
-        items = await page.query_selector_all("li.b_algo")
-        for item in items[:n_results]:
-            title_el = await item.query_selector("h2 > a")
-            url = await title_el.get_attribute("href") if title_el else None
-            title = await title_el.inner_text() if title_el else ""
-
-            snippet = ""
-
-            # Try several fallback selectors
-            for selector in [
-                "div.b_caption p",  # typical snippet
-                "div.b_caption",    # sometimes snippet is here
-                "div.b_snippet",    # used in some result types
-                "div.b_text",       # used in some panels
-                "p"                 # fallback to any paragraph
-            ]:
-                snippet_el = await item.query_selector(selector)
-                if snippet_el:
-                    snippet = await snippet_el.inner_text()
-                    if snippet.strip():
-                        break
-
-            if title and url:
-                results.append({
-                    "title": title.strip(),
-                    "href": url.strip(),
-                    "body": snippet.strip()
-                })
-
-        return results
+        return await _extract_bing_results(page, n_results)
 
 
 async def query_ddg_search(q: str, n_results: int = 10):

tests/conftest.py

@@ -0,0 +1,61 @@
+import os
+import sys
+from pathlib import Path
+
+import pytest_asyncio
+
+# The app modules (serp.py, scrap.py, ops.py, utils.py) live at the repo
+# root, not inside a package - make sure they're importable regardless of
+# how pytest resolves its rootdir.
+REPO_ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO_ROOT))
+
+
+def _find_chromium_executable() -> str | None:
+    """Fallback lookup for environments where the browser Playwright's pip
+    package expects doesn't match what's actually on disk (e.g. a sandbox
+    image that pre-installs a specific Chromium revision separately from
+    however `pip install playwright` resolved). Mirrors the workaround
+    documented for the JS Playwright test runner in this environment.
+    """
+    browsers_path = os.environ.get("PLAYWRIGHT_BROWSERS_PATH")
+    if not browsers_path:
+        return None
+    root = Path(browsers_path)
+    if not root.is_dir():
+        return None
+    candidates = sorted(root.glob("chromium-*/chrome-linux/chrome"))
+    return str(candidates[-1]) if candidates else None
+
+
+@pytest_asyncio.fixture
+async def browser():
+    """A real headless Chromium instance, exactly like the one `app.py`'s
+    lifespan starts in production.
+
+    Function-scoped (one launch per test) rather than session-scoped: each
+    pytest-asyncio test runs on its own event loop by default, and a
+    session-scoped instance would be created on one loop's setup and then
+    awaited from a different, later loop in the next test. Playwright's
+    connection keeps a background reader task bound to the loop it was
+    created on, so calls from another loop don't error - they just hang
+    forever waiting for a reply the old loop's task will never deliver.
+    Confirmed by reproducing this exact hang with a session-scoped version
+    of this fixture, then ruling out the scraping logic itself by running
+    the same call in a plain script outside pytest (single event loop),
+    where it completed instantly. A fresh browser per test costs roughly a
+    second of extra startup but sidesteps the whole cross-loop class of bug.
+    """
+    from playwright.async_api import async_playwright
+
+    pw = await async_playwright().start()
+    try:
+        b = await pw.chromium.launch(headless=True)
+    except Exception:
+        executable_path = _find_chromium_executable()
+        if not executable_path:
+            raise
+        b = await pw.chromium.launch(headless=True, executable_path=executable_path)
+    yield b
+    await b.close()
+    await pw.stop()

tests/fixtures/arxiv_response.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<feed xmlns="http://www.w3.org/2005/Atom">
+  <entry>
+    <id>http://arxiv.org/abs/2401.00001v1</id>
+    <title>
+      A Study of Widgets
+    </title>
+    <summary>
+      This paper studies widgets in depth.
+    </summary>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/2401.00002v1</id>
+    <title>Gadgets and Their Applications</title>
+    <summary>An overview of gadgets.</summary>
+  </entry>
+</feed>

tests/fixtures/bing_results.html

@@ -0,0 +1,13 @@
+<!DOCTYPE html>
+<html>
+<body>
+<li class="b_algo">
+  <h2><a href="https://example.org/a">Result A</a></h2>
+  <div class="b_caption"><p>Snippet A</p></div>
+</li>
+<li class="b_algo">
+  <h2><a href="https://example.org/b">Result B</a></h2>
+  <div class="b_caption"><p>Snippet B</p></div>
+</li>
+</body>
+</html>

tests/fixtures/brave_blocked.html

@@ -0,0 +1,6 @@
+<!DOCTYPE html>
+<html>
+<body>
+<p>Your request looks suspicious. Please verify you are human to continue.</p>
+</body>
+</html>

tests/fixtures/brave_results.html

@@ -0,0 +1,13 @@
+<!DOCTYPE html>
+<html>
+<body>
+<div class="snippet">
+  <a href="https://example.org/brave-a"><span class="title">Brave Result A</span></a>
+  <p class="snippet-description">Brave snippet A</p>
+</div>
+<div class="snippet">
+  <a href="https://example.org/brave-b"><span class="title">Brave Result B</span></a>
+  <p class="snippet-description">Brave snippet B</p>
+</div>
+</body>
+</html>

tests/fixtures/ops/biblio_response.json

@@ -0,0 +1,37 @@
+{
+  "ops:world-patent-data": {
+    "exchange-documents": {
+      "exchange-document": {
+        "@country": "US",
+        "@doc-number": "11930446",
+        "bibliographic-data": {
+          "publication-reference": {
+            "document-id": [
+              {"@document-id-type": "epodoc", "doc-number": {"$": "US11930446"}}
+            ]
+          },
+          "invention-title": {"@lang": "en", "$": "Widget apparatus"},
+          "patent-classifications": {
+            "patent-classification": [
+              {
+                "section": {"$": "G"},
+                "class": {"$": "06"},
+                "subclass": {"$": "F"},
+                "main-group": {"$": "17"},
+                "subgroup": {"$": "30"}
+              },
+              {
+                "section": {"$": "H"},
+                "class": {"$": "04"},
+                "subclass": {"$": "L"},
+                "main-group": {"$": "9"},
+                "subgroup": {"$": "00"}
+              }
+            ]
+          }
+        },
+        "abstract": {"@lang": "en", "p": [{"$": "An apparatus comprising a widget."}]}
+      }
+    }
+  }
+}

tests/fixtures/ops/claims_response.json

@@ -0,0 +1,15 @@
+{
+  "ops:world-patent-data": {
+    "ftxt:fulltext-documents": {
+      "ftxt:fulltext-document": {
+        "claims": {
+          "@lang": "en",
+          "claim": [
+            {"claim-text": {"$": "1. A widget comprising a gadget."}},
+            {"claim-text": {"$": "2. The widget of claim 1, further comprising a mechanism."}}
+          ]
+        }
+      }
+    }
+  }
+}

tests/fixtures/ops/description_response.json

@@ -0,0 +1,15 @@
+{
+  "ops:world-patent-data": {
+    "ftxt:fulltext-documents": {
+      "ftxt:fulltext-document": {
+        "description": {
+          "@lang": "en",
+          "p": [
+            {"$": "This invention relates to widgets."},
+            {"$": "The widget comprises a gadget and a mechanism."}
+          ]
+        }
+      }
+    }
+  }
+}

tests/fixtures/ops/search_not_found_response.json

@@ -0,0 +1,6 @@
+{
+  "fault": {
+    "faultstring": "SERVER.EntityNotFound",
+    "detail": {"errorcode": "SERVER.EntityNotFound"}
+  }
+}

tests/fixtures/ops/search_response.json

@@ -0,0 +1,45 @@
+{
+  "ops:world-patent-data": {
+    "ops:biblio-search": {
+      "ops:search-result": {
+        "exchange-documents": [
+          {
+            "exchange-document": {
+              "@country": "US",
+              "@doc-number": "11930446",
+              "bibliographic-data": {
+                "publication-reference": {
+                  "document-id": [
+                    {"@document-id-type": "epodoc", "doc-number": {"$": "US11930446"}}
+                  ]
+                },
+                "invention-title": [
+                  {"@lang": "en", "$": "Widget apparatus"},
+                  {"@lang": "de", "$": "Widgetvorrichtung"}
+                ]
+              },
+              "abstract": [
+                {"@lang": "en", "p": [{"$": "An apparatus comprising a widget."}]}
+              ]
+            }
+          },
+          {
+            "exchange-document": {
+              "@country": "EP",
+              "@doc-number": "4760514",
+              "bibliographic-data": {
+                "publication-reference": {
+                  "document-id": [
+                    {"@document-id-type": "epodoc", "doc-number": {"$": "EP4760514"}}
+                  ]
+                },
+                "invention-title": {"@lang": "en", "$": "Gadget system"}
+              },
+              "abstract": {"@lang": "en", "p": {"$": "A system comprising a gadget."}}
+            }
+          }
+        ]
+      }
+    }
+  }
+}

tests/fixtures/ops/token_response.json

@@ -0,0 +1,5 @@
+{
+  "access_token": "test-access-token-123",
+  "expires_in": 1199,
+  "token_type": "Bearer"
+}

tests/fixtures/patents/full_patent.html

@@ -0,0 +1,41 @@
+<!DOCTYPE html>
+<html>
+<head>
+<meta name="DC.title" content="Widget with improved gadget mechanism">
+</head>
+<body>
+
+<div class="abstract">A widget comprising a gadget and a mechanism for improving widget performance.</div>
+
+<section itemprop="description">
+<div>TECHNICAL FIELD</div>
+<div>This invention relates to widgets and gadgets for testing purposes.</div>
+<div>BACKGROUND OF THE INVENTION</div>
+<div>Prior art widgets suffered from several problems described herein.</div>
+<div>SUMMARY OF THE INVENTION</div>
+<div>The present invention solves the problems of the prior art by providing an improved widget.</div>
+<div>DETAILED DESCRIPTION</div>
+<div>The following describes example embodiments of the widget in detail.</div>
+</section>
+
+<section itemprop="claims">
+<div class="claim">1. A widget comprising a gadget.</div>
+<div class="claim">2. The widget of claim 1, further comprising a mechanism.</div>
+</section>
+
+<ul class="classifications">
+  <li><span>G06F</span>
+    <ul>
+      <li><span>G06F17/30</span> Database structures therefor</li>
+      <li><span>G06F17/50</span> Other database related</li>
+    </ul>
+  </li>
+  <li><span>H04L</span>
+    <ul>
+      <li><span>H04L9/00</span> Cryptographic mechanisms</li>
+    </ul>
+  </li>
+</ul>
+
+</body>
+</html>

tests/fixtures/patents/missing_title.html

@@ -0,0 +1,8 @@
+<!DOCTYPE html>
+<html>
+<head>
+</head>
+<body>
+<p>This is not a valid Google Patents page (interstitial, error page, or markup change) - it has no meta[name=DC.title].</p>
+</body>
+</html>

tests/fixtures/patents/search_results.html

@@ -0,0 +1,15 @@
+<!DOCTYPE html>
+<html>
+<body>
+<search-result-item>
+  <span>US11930446B2</span>
+  <h3>Widget apparatus</h3>
+  <div class="abstract">An apparatus comprising a widget.</div>
+</search-result-item>
+<search-result-item>
+  <span>EP4760514A1</span>
+  <h4>Gadget system</h4>
+  <div class="result-snippet">A system comprising a gadget.</div>
+</search-result-item>
+</body>
+</html>

tests/fixtures/scholar_results.html

@@ -0,0 +1,13 @@
+<!DOCTYPE html>
+<html>
+<body>
+<div class="gs_ri">
+  <h3><a href="https://example.org/scholar-a">Scholar Paper A</a></h3>
+  <div class="gs_rs">Scholar snippet A</div>
+</div>
+<div class="gs_ri">
+  <h3><a href="https://example.org/scholar-b">Scholar Paper B</a></h3>
+  <div class="gs_rs">Scholar snippet B</div>
+</div>
+</body>
+</html>

tests/helpers.py

@@ -0,0 +1,73 @@
+"""Shared test helpers: fixture loading and a fake-navigation Playwright
+Browser wrapper used to characterize the scraper functions in `serp.py`
+without hitting live search engines.
+"""
+
+from pathlib import Path
+
+FIXTURES_DIR = Path(__file__).resolve().parent / "fixtures"
+
+
+def load_fixture(*parts: str) -> str:
+    """Read a text fixture from tests/fixtures/<parts...>."""
+    return FIXTURES_DIR.joinpath(*parts).read_text(encoding="utf-8")
+
+
+class _GotoPatchingContext:
+    """Wraps a real Playwright BrowserContext so every Page it creates loads
+    fixture HTML via `set_content()` instead of navigating over the network.
+    """
+
+    def __init__(self, real_context, html: str):
+        self._real_context = real_context
+        self._html = html
+
+    async def new_page(self):
+        page = await self._real_context.new_page()
+        html = self._html
+
+        async def fake_goto(url, **kwargs):
+            # The scraper functions don't use the response object `goto`
+            # normally returns, so a plain None is a safe stand-in.
+            await page.set_content(html)
+            return None
+
+        page.goto = fake_goto
+        return page
+
+    async def close(self):
+        await self._real_context.close()
+
+
+class _GotoPatchingBrowser:
+    """Wraps a real Playwright Browser the same way, so it can be passed
+    directly to `serp.py` functions in place of the real `pw_browser`.
+    """
+
+    def __init__(self, real_browser, html: str):
+        self._real_browser = real_browser
+        self._html = html
+
+    async def new_context(self, **kwargs):
+        real_context = await self._real_browser.new_context(**kwargs)
+        return _GotoPatchingContext(real_context, self._html)
+
+
+def make_fixture_browser(real_browser, html: str) -> _GotoPatchingBrowser:
+    """Build a Browser-like object for characterizing `serp.py`'s scraper
+    functions against fixture HTML.
+
+    Why this exists: each scraper (e.g. `query_bing_search`) registers its
+    own catch-all `page.route("**/*", ...)` resource blocker *inside* the
+    function, and that handler calls `route.continue_()` (not
+    `route.fallback()`) for the main document request. Playwright invokes
+    the most-recently-registered matching route first, so any route we
+    register on the page *before* calling the function never runs -
+    verified empirically, it's not a documented guarantee either way.
+    Patching `page.goto` itself, on a page created through this wrapper, is
+    the only reliable seam into these functions without touching their
+    source (see the code review this test suite followed from: the
+    fetch/navigate and parse logic aren't separated, which is exactly what
+    makes this workaround necessary).
+    """
+    return _GotoPatchingBrowser(real_browser, html)

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"}

tests/test_ops_client.py

@@ -0,0 +1,180 @@
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+import respx
+
+import ops as ops_module
+from ops import OPSNotConfigured, _OPSTokenManager, _ops_get, ops_scrap_patent, ops_search
+
+FIXTURES = Path(__file__).parent / "fixtures" / "ops"
+TOKEN_URL = "https://ops.epo.org/3.2/auth/accesstoken"
+OPS_BASE = "https://ops.epo.org/3.2/rest-services"
+
+
+def load_json(name: str):
+    return json.loads((FIXTURES / name).read_text())
+
+
+def mock_token_endpoint():
+    return respx.post(TOKEN_URL).mock(
+        return_value=httpx.Response(200, json=load_json("token_response.json"))
+    )
+
+
+@pytest.fixture
+def fresh_token_manager(monkeypatch):
+    """A token manager with test credentials, swapped in for the module-level
+    singleton so `_ops_get`/`ops_search`/etc. (which reach for the module
+    global `token_manager` directly) pick it up.
+    """
+    tm = _OPSTokenManager()
+    tm._key = "test-key"
+    tm._secret = "test-secret"
+    monkeypatch.setattr(ops_module, "token_manager", tm)
+    return tm
+
+
+class TestTokenManager:
+    async def test_not_configured_raises(self):
+        tm = _OPSTokenManager()
+        tm._key = None
+        tm._secret = None
+        async with httpx.AsyncClient() as client:
+            with pytest.raises(OPSNotConfigured):
+                await tm.get_token(client)
+
+    async def test_fetches_and_caches_the_token(self):
+        tm = _OPSTokenManager()
+        tm._key = "k"
+        tm._secret = "s"
+        with respx.mock:
+            route = mock_token_endpoint()
+            async with httpx.AsyncClient() as client:
+                token1 = await tm.get_token(client)
+                token2 = await tm.get_token(client)
+
+        assert token1 == "test-access-token-123"
+        assert token2 == "test-access-token-123"
+        assert route.call_count == 1
+
+    async def test_force_refresh_refetches_even_if_cached(self):
+        tm = _OPSTokenManager()
+        tm._key = "k"
+        tm._secret = "s"
+        with respx.mock:
+            route = mock_token_endpoint()
+            async with httpx.AsyncClient() as client:
+                await tm.get_token(client)
+                await tm.get_token(client, force=True)
+
+        assert route.call_count == 2
+
+
+class TestOpsGet:
+    async def test_retries_once_when_the_token_is_stale(self, fresh_token_manager):
+        with respx.mock:
+            mock_token_endpoint()
+            get_route = respx.get(f"{OPS_BASE}/some/path").mock(
+                side_effect=[
+                    httpx.Response(401, text="Invalid access token."),
+                    httpx.Response(200, json={"ok": True}),
+                ]
+            )
+            async with httpx.AsyncClient() as client:
+                data = await _ops_get(client, "/some/path")
+
+        assert data == {"ok": True}
+        assert get_route.call_count == 2
+
+    async def test_propagates_non_token_errors(self, fresh_token_manager):
+        with respx.mock:
+            mock_token_endpoint()
+            respx.get(f"{OPS_BASE}/some/path").mock(return_value=httpx.Response(500, text="server error"))
+            async with httpx.AsyncClient() as client:
+                with pytest.raises(httpx.HTTPStatusError):
+                    await _ops_get(client, "/some/path")
+
+
+class TestOpsSearch:
+    async def test_maps_ops_search_results_to_serp_dicts(self, fresh_token_manager):
+        with respx.mock:
+            mock_token_endpoint()
+            respx.get(f"{OPS_BASE}/published-data/search/biblio").mock(
+                return_value=httpx.Response(200, json=load_json("search_response.json"))
+            )
+            async with httpx.AsyncClient() as client:
+                results = await ops_search(client, "widget", 10)
+
+        assert results == [
+            {
+                "title": "Widget apparatus",
+                "body": "An apparatus comprising a widget.",
+                "href": "https://worldwide.espacenet.com/patent/search?q=US11930446",
+                "id": "US11930446",
+            },
+            {
+                "title": "Gadget system",
+                "body": "A system comprising a gadget.",
+                "href": "https://worldwide.espacenet.com/patent/search?q=EP4760514",
+                "id": "EP4760514",
+            },
+        ]
+
+    async def test_returns_empty_list_on_404_not_found(self, fresh_token_manager):
+        with respx.mock:
+            mock_token_endpoint()
+            respx.get(f"{OPS_BASE}/published-data/search/biblio").mock(
+                return_value=httpx.Response(404, json=load_json("search_not_found_response.json"))
+            )
+            async with httpx.AsyncClient() as client:
+                results = await ops_search(client, "nonexistent widget xyz", 10)
+
+        assert results == []
+
+
+class TestOpsScrapPatent:
+    async def test_assembles_a_full_patent_from_three_endpoints(self, fresh_token_manager):
+        number = "US11930446"
+        with respx.mock:
+            mock_token_endpoint()
+            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/biblio").mock(
+                return_value=httpx.Response(200, json=load_json("biblio_response.json"))
+            )
+            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/claims").mock(
+                return_value=httpx.Response(200, json=load_json("claims_response.json"))
+            )
+            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/description").mock(
+                return_value=httpx.Response(200, json=load_json("description_response.json"))
+            )
+            async with httpx.AsyncClient() as client:
+                # Kind-code suffix ("B2") should be normalized away before
+                # any of the three requests go out.
+                result = await ops_scrap_patent(client, "US11930446B2")
+
+        assert result.title == "Widget apparatus"
+        assert result.abstract == "An apparatus comprising a widget."
+        assert result.claims == "1. A widget comprising a gadget.\n2. The widget of claim 1, further comprising a mechanism."
+        assert result.description == "This invention relates to widgets.\nThe widget comprises a gadget and a mechanism."
+        assert {c.code for c in result.classifications} == {"G06F17/30", "H04L9/00"}
+
+    async def test_missing_fulltext_degrades_to_none_instead_of_raising(self, fresh_token_manager):
+        number = "US11930446"
+        with respx.mock:
+            mock_token_endpoint()
+            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/biblio").mock(
+                return_value=httpx.Response(200, json=load_json("biblio_response.json"))
+            )
+            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/claims").mock(
+                return_value=httpx.Response(404)
+            )
+            respx.get(f"{OPS_BASE}/published-data/publication/epodoc/{number}/description").mock(
+                return_value=httpx.Response(404)
+            )
+            async with httpx.AsyncClient() as client:
+                result = await ops_scrap_patent(client, number)
+
+        assert result.title == "Widget apparatus"
+        assert result.claims is None
+        assert result.description is None

tests/test_ops_parsing.py

@@ -0,0 +1,215 @@
+"""Unit tests for ops.py's pure JSON-walking helpers.
+
+OPS's XML-to-JSON conversion is the trickiest part of this module: text
+nodes become {"$": "..."}, attributes become "@name" keys, and a repeated
+XML element becomes a list *unless there's only one*, in which case it stays
+a bare dict. Every helper here has to handle both shapes; these tests pin
+both for each one.
+"""
+
+import pytest
+
+from ops import (
+    _as_list,
+    _text,
+    _pick_lang,
+    _epodoc_number,
+    _extract_title,
+    _extract_abstract,
+    _extract_classifications,
+    _fulltext_paragraphs,
+    _to_cql,
+    _normalize_epodoc,
+)
+
+
+class TestAsList:
+    def test_none_becomes_empty_list(self):
+        assert _as_list(None) == []
+
+    def test_single_dict_is_wrapped(self):
+        assert _as_list({"a": 1}) == [{"a": 1}]
+
+    def test_list_passes_through_unchanged(self):
+        assert _as_list([{"a": 1}, {"a": 2}]) == [{"a": 1}, {"a": 2}]
+
+
+class TestText:
+    def test_extracts_dollar_key(self):
+        assert _text({"$": "hello"}) == "hello"
+
+    def test_plain_string_passes_through(self):
+        assert _text("hello") == "hello"
+
+    def test_none_for_anything_else(self):
+        assert _text(None) is None
+        assert _text({"@lang": "en"}) is None
+
+
+class TestPickLang:
+    def test_picks_requested_language_from_list(self):
+        nodes = [{"@lang": "de", "$": "Hallo"}, {"@lang": "en", "$": "Hello"}]
+        assert _pick_lang(nodes) == {"@lang": "en", "$": "Hello"}
+
+    def test_falls_back_to_first_when_language_missing(self):
+        nodes = [{"@lang": "de", "$": "Hallo"}, {"@lang": "fr", "$": "Bonjour"}]
+        assert _pick_lang(nodes) == {"@lang": "de", "$": "Hallo"}
+
+    def test_single_dict_not_wrapped_in_a_list_still_works(self):
+        node = {"@lang": "en", "$": "Hello"}
+        assert _pick_lang(node) == node
+
+    def test_empty_input_returns_none(self):
+        assert _pick_lang(None) is None
+        assert _pick_lang([]) is None
+
+
+class TestEpodocNumber:
+    def test_prefers_the_epodoc_document_id(self):
+        exchange_doc = {
+            "@country": "US",
+            "@doc-number": "11930446",
+            "bibliographic-data": {
+                "publication-reference": {
+                    "document-id": [
+                        {"@document-id-type": "docdb", "doc-number": {"$": "11930446"}},
+                        {"@document-id-type": "epodoc", "doc-number": {"$": "US11930446"}},
+                    ]
+                }
+            },
+        }
+        assert _epodoc_number(exchange_doc) == "US11930446"
+
+    def test_falls_back_to_country_plus_doc_number(self):
+        exchange_doc = {"@country": "US", "@doc-number": "11930446", "bibliographic-data": {}}
+        assert _epodoc_number(exchange_doc) == "US11930446"
+
+    def test_none_when_nothing_available(self):
+        assert _epodoc_number({"bibliographic-data": {}}) is None
+
+
+class TestExtractTitle:
+    def test_from_a_list_of_languages(self):
+        bd = {"invention-title": [{"@lang": "en", "$": "Widget"}, {"@lang": "de", "$": "Widget-DE"}]}
+        assert _extract_title(bd) == "Widget"
+
+    def test_from_a_single_bare_dict(self):
+        bd = {"invention-title": {"@lang": "en", "$": "Widget"}}
+        assert _extract_title(bd) == "Widget"
+
+    def test_none_when_missing(self):
+        assert _extract_title({}) is None
+
+
+class TestExtractAbstract:
+    def test_joins_paragraphs_from_a_list(self):
+        doc = {"abstract": [{"@lang": "en", "p": [{"$": "Part one."}, {"$": "Part two."}]}]}
+        assert _extract_abstract(doc) == "Part one. Part two."
+
+    def test_single_paragraph_not_wrapped_in_a_list(self):
+        doc = {"abstract": {"@lang": "en", "p": {"$": "Only paragraph."}}}
+        assert _extract_abstract(doc) == "Only paragraph."
+
+    def test_none_when_missing(self):
+        assert _extract_abstract({}) is None
+
+
+class TestExtractClassifications:
+    def test_builds_codes_from_the_five_part_symbol(self):
+        bd = {
+            "patent-classifications": {
+                "patent-classification": [
+                    {
+                        "section": {"$": "G"},
+                        "class": {"$": "06"},
+                        "subclass": {"$": "F"},
+                        "main-group": {"$": "17"},
+                        "subgroup": {"$": "30"},
+                    }
+                ]
+            }
+        }
+        codes = _extract_classifications(bd)
+        assert len(codes) == 1
+        assert codes[0].code == "G06F17/30"
+        assert codes[0].description == ""
+
+    def test_dedupes_and_skips_incomplete_entries(self):
+        bd = {
+            "patent-classifications": {
+                "patent-classification": [
+                    {"section": {"$": "G"}, "class": {"$": "06"}, "subclass": {"$": "F"}, "main-group": {"$": "17"}, "subgroup": {"$": "30"}},
+                    {"section": {"$": "G"}, "class": {"$": "06"}, "subclass": {"$": "F"}, "main-group": {"$": "17"}, "subgroup": {"$": "30"}},
+                    {"section": {"$": "H"}},  # incomplete - no class/subclass/main-group
+                ]
+            }
+        }
+        codes = _extract_classifications(bd)
+        assert [c.code for c in codes] == ["G06F17/30"]
+
+    def test_none_when_no_classifications(self):
+        assert _extract_classifications({}) is None
+
+
+class TestFulltextParagraphs:
+    def test_claims_join_claim_text_lines(self):
+        data = {
+            "ops:world-patent-data": {
+                "ftxt:fulltext-documents": {
+                    "ftxt:fulltext-document": {
+                        "claims": {
+                            "@lang": "en",
+                            "claim": [
+                                {"claim-text": {"$": "1. A widget."}},
+                                {"claim-text": [{"$": "2. A gadget,"}, {"$": "further comprising a mechanism."}]},
+                            ],
+                        }
+                    }
+                }
+            }
+        }
+        assert _fulltext_paragraphs(data, "claims") == (
+            "1. A widget.\n2. A gadget,\nfurther comprising a mechanism."
+        )
+
+    def test_description_joins_paragraphs(self):
+        data = {
+            "ops:world-patent-data": {
+                "ftxt:fulltext-documents": {
+                    "ftxt:fulltext-document": {
+                        "description": {"@lang": "en", "p": [{"$": "Para one."}, {"$": "Para two."}]}
+                    }
+                }
+            }
+        }
+        assert _fulltext_paragraphs(data, "description") == "Para one.\nPara two."
+
+    def test_none_when_document_missing(self):
+        assert _fulltext_paragraphs({"ops:world-patent-data": {}}, "claims") is None
+
+
+@pytest.mark.parametrize(
+    "raw, expected",
+    [
+        ("widget", 'txt all "widget"'),
+        ("agentic ai", 'txt all "agentic ai"'),
+        ('pn = "US1234"', 'pn = "US1234"'),
+        ("", ""),
+        ('has "quotes"', 'txt all "has  quotes"'),
+    ],
+)
+def test_to_cql(raw, expected):
+    assert _to_cql(raw) == expected
+
+
+@pytest.mark.parametrize(
+    "raw, expected",
+    [
+        ("US11930446B2", "US11930446"),
+        ("EP4760514A1", "EP4760514"),
+        ("US 11930446 B2", "US11930446"),
+        ("US11930446", "US11930446"),
+    ],
+)
+def test_normalize_epodoc(raw, expected):
+    assert _normalize_epodoc(raw) == expected

tests/test_scrap.py

@@ -0,0 +1,91 @@
+import httpx
+import pytest
+import respx
+
+from helpers import load_fixture
+from scrap import parse_patent_html, scrap_patent_async, scrap_patent_bulk_async
+
+FULL_PATENT_HTML = load_fixture("patents", "full_patent.html")
+MISSING_TITLE_HTML = load_fixture("patents", "missing_title.html")
+
+
+def test_parse_patent_html_extracts_all_fields_with_no_network_involved():
+    """`scrap_patent_async` is now a thin fetch that delegates to this - a
+    plain, synchronous function of the HTML string, so the parsing logic
+    (including the regex-based section splitting, which is the part most
+    likely to break when Google Patents changes its markup) can be tested
+    directly against a saved page with no client/respx/event loop needed.
+    """
+    result = parse_patent_html(FULL_PATENT_HTML, "https://patents.google.com/patent/US11930446B2/en")
+
+    assert result.title == "Widget with improved gadget mechanism"
+    codes = {c.code: c.description for c in result.classifications}
+    assert codes == {
+        "G06F17/30": "Database structures therefor",
+        "G06F17/50": "Other database related",
+        "H04L9/00": "Cryptographic mechanisms",
+    }
+
+
+def test_parse_patent_html_raises_when_page_has_no_title():
+    with pytest.raises(ValueError):
+        parse_patent_html(MISSING_TITLE_HTML, "https://patents.google.com/patent/BOGUS/en")
+
+
+async def test_scrap_patent_async_extracts_all_fields():
+    url = "https://patents.google.com/patent/US11930446B2/en"
+    with respx.mock:
+        respx.get(url).mock(return_value=httpx.Response(200, text=FULL_PATENT_HTML))
+        async with httpx.AsyncClient() as client:
+            result = await scrap_patent_async(client, url)
+
+    assert result.title == "Widget with improved gadget mechanism"
+    assert result.abstract == "A widget comprising a gadget and a mechanism for improving widget performance."
+    assert result.field_of_invention == "This invention relates to widgets and gadgets for testing purposes."
+    assert result.background == "Prior art widgets suffered from several problems described herein."
+    assert "1. A widget comprising a gadget." in result.claims
+    assert "2. The widget of claim 1, further comprising a mechanism." in result.claims
+
+    codes = {c.code: c.description for c in result.classifications}
+    assert codes == {
+        "G06F17/30": "Database structures therefor",
+        "G06F17/50": "Other database related",
+        "H04L9/00": "Cryptographic mechanisms",
+    }
+
+
+async def test_scrap_patent_async_raises_when_page_has_no_title():
+    """Pins the documented behavior at scrap.py: an unexpected page layout
+    (interstitial, non-patent page, markup change) raises ValueError rather
+    than crashing on `None.get(...)`, so callers can treat it as a failed
+    scrape.
+    """
+    url = "https://patents.google.com/patent/BOGUS/en"
+    with respx.mock:
+        respx.get(url).mock(return_value=httpx.Response(200, text=MISSING_TITLE_HTML))
+        async with httpx.AsyncClient() as client:
+            with pytest.raises(ValueError):
+                await scrap_patent_async(client, url)
+
+
+async def test_scrap_patent_async_raises_on_http_error():
+    url = "https://patents.google.com/patent/US00000000/en"
+    with respx.mock:
+        respx.get(url).mock(return_value=httpx.Response(404))
+        async with httpx.AsyncClient() as client:
+            with pytest.raises(httpx.HTTPStatusError):
+                await scrap_patent_async(client, url)
+
+
+async def test_scrap_patent_bulk_async_separates_successes_from_failures():
+    ok_url = "https://patents.google.com/patent/US_OK/en"
+    fail_url = "https://patents.google.com/patent/US_FAIL/en"
+    with respx.mock:
+        respx.get(ok_url).mock(return_value=httpx.Response(200, text=FULL_PATENT_HTML))
+        respx.get(fail_url).mock(return_value=httpx.Response(404))
+        async with httpx.AsyncClient() as client:
+            result = await scrap_patent_bulk_async(client, ["US_OK", "US_FAIL"])
+
+    assert len(result.patents) == 1
+    assert result.patents[0].title == "Widget with improved gadget mechanism"
+    assert result.failed_ids == ["US_FAIL"]

tests/test_serp_arxiv.py

@@ -0,0 +1,31 @@
+import httpx
+import respx
+
+from helpers import load_fixture
+from serp import query_arxiv
+
+ARXIV_XML = load_fixture("arxiv_response.xml")
+
+
+async def test_query_arxiv_parses_the_atom_feed():
+    with respx.mock:
+        respx.get("https://export.arxiv.org/api/query").mock(
+            return_value=httpx.Response(200, text=ARXIV_XML, headers={"content-type": "application/atom+xml"})
+        )
+        async with httpx.AsyncClient() as client:
+            results = await query_arxiv(client, "widgets", 2)
+
+    assert results == [
+        {
+            "title": "A Study of Widgets",
+            "href": "http://arxiv.org/pdf/2401.00001v1",
+            "body": "This paper studies widgets in depth.",
+            "id": "http://arxiv.org/abs/2401.00001v1",
+        },
+        {
+            "title": "Gadgets and Their Applications",
+            "href": "http://arxiv.org/pdf/2401.00002v1",
+            "body": "An overview of gadgets.",
+            "id": "http://arxiv.org/abs/2401.00002v1",
+        },
+    ]

tests/test_serp_ddg.py

@@ -0,0 +1,44 @@
+import pytest
+
+import serp
+from serp import DuckDuckGoBlockedException, query_ddg_search
+
+
+class _FakeDDGS:
+    def __init__(self, results):
+        self._results = results
+
+    def text(self, q, max_results=10):
+        return iter(self._results)
+
+
+async def test_maps_duckduckgo_results_to_serp_dicts(monkeypatch):
+    monkeypatch.setattr(
+        serp,
+        "DDGS",
+        lambda: _FakeDDGS(
+            [
+                {"title": "A", "body": "Body A", "href": "https://a.example"},
+                {"title": "B", "body": "Body B", "href": "https://b.example"},
+            ]
+        ),
+    )
+
+    results = await query_ddg_search("test query", 10)
+
+    assert results == [
+        {"title": "A", "body": "Body A", "href": "https://a.example"},
+        {"title": "B", "body": "Body B", "href": "https://b.example"},
+    ]
+
+
+async def test_raises_blocked_exception_on_empty_results(monkeypatch):
+    """Pins the fix from commit d341a7a: duckduckgo_search doesn't always
+    raise when it's rate-limited/blocked - it can just parse zero results
+    from the anti-bot page. An empty list must be treated as a failure so
+    the `/serp/search` fallback chain moves on to the next backend.
+    """
+    monkeypatch.setattr(serp, "DDGS", lambda: _FakeDDGS([]))
+
+    with pytest.raises(DuckDuckGoBlockedException):
+        await query_ddg_search("test query", 10)

tests/test_serp_models.py

@@ -0,0 +1,25 @@
+import pytest
+
+from serp import SerpQuery
+
+
+@pytest.mark.parametrize(
+    "raw, expected",
+    [
+        (10, 10),
+        (1, 1),
+        (100, 100),
+        (0, 1),
+        (-5, 1),
+        (500, 100),
+        (101, 100),
+    ],
+)
+def test_n_results_is_clamped_to_1_100(raw, expected):
+    q = SerpQuery(queries=["test"], n_results=raw)
+    assert q.n_results == expected
+
+
+def test_n_results_defaults_to_10():
+    q = SerpQuery(queries=["test"])
+    assert q.n_results == 10

tests/test_serp_playwright.py

@@ -0,0 +1,109 @@
+"""Characterization tests for the Playwright-driven scrapers in serp.py.
+
+These run against a real headless Chromium (same engine production uses) so
+that DOM/CSS-selector behavior is genuinely exercised, but the target page is
+served locally from a fixture instead of navigating to the live site - see
+`helpers.make_fixture_browser` for how, and why a simpler
+`page.route()`-based approach doesn't work here.
+"""
+
+import pytest
+
+from helpers import load_fixture, make_fixture_browser
+from serp import (
+    BraveSearchBlockedException,
+    BrowserUnavailableError,
+    _extract_bing_results,
+    playwright_open_page,
+    query_bing_search,
+    query_brave_search,
+    query_google_patents,
+    query_google_scholar,
+)
+
+
+async def test_extract_bing_results_works_on_an_already_loaded_page(browser):
+    """Demonstrates the payoff of splitting fetch (navigate) from parse
+    (extract): `_extract_bing_results` takes a Page that's already loaded,
+    so this needs nothing beyond `set_content` - no goto-patching wrapper,
+    no browser-as-a-parameter indirection.
+    """
+    context = await browser.new_context()
+    page = await context.new_page()
+    await page.set_content(load_fixture("bing_results.html"))
+
+    results = await _extract_bing_results(page, 10)
+
+    assert results == [
+        {"title": "Result A", "href": "https://example.org/a", "body": "Snippet A"},
+        {"title": "Result B", "href": "https://example.org/b", "body": "Snippet B"},
+    ]
+
+    await context.close()
+
+
+async def test_bing_search_parses_results(browser):
+    fixture_browser = make_fixture_browser(browser, load_fixture("bing_results.html"))
+
+    results = await query_bing_search(fixture_browser, "test query", 10)
+
+    assert results == [
+        {"title": "Result A", "href": "https://example.org/a", "body": "Snippet A"},
+        {"title": "Result B", "href": "https://example.org/b", "body": "Snippet B"},
+    ]
+
+
+async def test_brave_search_parses_results(browser):
+    fixture_browser = make_fixture_browser(browser, load_fixture("brave_results.html"))
+
+    results = await query_brave_search(fixture_browser, "test query", 10)
+
+    assert results == [
+        {"title": "Brave Result A", "body": "Brave snippet A", "href": "https://example.org/brave-a"},
+        {"title": "Brave Result B", "body": "Brave snippet B", "href": "https://example.org/brave-b"},
+    ]
+
+
+async def test_brave_search_detects_the_suspicious_block_page(browser):
+    fixture_browser = make_fixture_browser(browser, load_fixture("brave_blocked.html"))
+
+    with pytest.raises(BraveSearchBlockedException):
+        await query_brave_search(fixture_browser, "test query", 10)
+
+
+async def test_google_scholar_parses_results(browser):
+    fixture_browser = make_fixture_browser(browser, load_fixture("scholar_results.html"))
+
+    results = await query_google_scholar(fixture_browser, "test query", 10)
+
+    assert results == [
+        {"title": "Scholar Paper A", "body": "Scholar snippet A", "href": "https://example.org/scholar-a"},
+        {"title": "Scholar Paper B", "body": "Scholar snippet B", "href": "https://example.org/scholar-b"},
+    ]
+
+
+async def test_google_patents_parses_results(browser):
+    fixture_browser = make_fixture_browser(browser, load_fixture("patents", "search_results.html"))
+
+    results = await query_google_patents(fixture_browser, "test query", 10)
+
+    assert results == [
+        {
+            "id": "US11930446B2",
+            "href": "https://patents.google.com/patent/US11930446B2/en",
+            "title": "Widget apparatus",
+            "body": "An apparatus comprising a widget.",
+        },
+        {
+            "id": "EP4760514A1",
+            "href": "https://patents.google.com/patent/EP4760514A1/en",
+            "title": "Gadget system",
+            "body": "A system comprising a gadget.",
+        },
+    ]
+
+
+async def test_playwright_open_page_raises_when_browser_unavailable():
+    with pytest.raises(BrowserUnavailableError):
+        async with playwright_open_page(None):
+            pass

tests/test_utils.py

@@ -0,0 +1,40 @@
+import logging
+
+from serp import SerpQuery
+from utils import log_gathered_exceptions
+
+
+def test_logs_only_the_exceptions_with_their_query(caplog):
+    params = SerpQuery(queries=["q1", "q2", "q3"])
+    results = ["ok result", ValueError("boom"), "another ok"]
+
+    with caplog.at_level(logging.WARNING):
+        log_gathered_exceptions(results, "test context", params)
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 1
+    assert "q2" in warnings[0].message
+    assert "boom" in warnings[0].message
+    assert "test context" in warnings[0].message
+
+
+def test_logs_nothing_when_all_results_succeeded(caplog):
+    params = SerpQuery(queries=["q1", "q2"])
+    results = ["ok", "also ok"]
+
+    with caplog.at_level(logging.WARNING):
+        log_gathered_exceptions(results, "test context", params)
+
+    assert [r for r in caplog.records if r.levelno == logging.WARNING] == []
+
+
+def test_uses_its_own_logger_not_the_asyncio_one(caplog):
+    """Was `from asyncio.log import logger` (flagged in review), which
+    misattributed these warnings to the 'asyncio' logging namespace instead
+    of the app's own. Fixed to `logging.getLogger(__name__)`.
+    """
+    params = SerpQuery(queries=["q1"])
+    with caplog.at_level(logging.WARNING):
+        log_gathered_exceptions([ValueError("boom")], "ctx", params)
+
+    assert caplog.records[0].name == "utils"

utils.py

@@ -1,6 +1,9 @@
-from asyncio.log import logger
+import logging
+
 from serp import SerpQuery
 
+logger = logging.getLogger(__name__)
+
 
 def log_gathered_exceptions(results: list, context: str, params: SerpQuery):
     """Logs gathered exceptions with context and parameters."""