github-featured/scrapegraphai--scrapegraph-ai
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(fetch): surface HTTP errors and missing content instead of answering NA

Browse Source 
A page that could not be scraped as intended was indistinguishable from one
that could. FetchNode's default path (ChromiumLoader -> ascrape_playwright)
dropped the Response returned by page.goto(), so a 404, 403, 500, captcha wall
or login redirect reached the LLM as ordinary content and the model answered
"NA" with nothing in the logs to explain why. Reported in #1102, where
en.wikipedia.org/wiki/Timpson_(company) 404s (the article is at
Timpson_(retailer)) and the run still looked clean.

Two deterministic, LLM-free guards, both warnings so existing behaviour is
unchanged for anyone deliberately scraping error pages:

- ChromiumLoader keeps the Response from every page.goto() call site
  (ascrape_playwright, ascrape_playwright_scroll, ascrape_with_js_support) and
  warns on status >= 400. This mirrors what the opt-in use_soup=True path in
  FetchNode has always done.
- ParseNode warns when the parsed content contains none of the terms the user
  asked about — schema field names plus the significant words of the prompt.
  A 200 response can still reach the LLM without the requested data: content
  behind JavaScript that never rendered, a field inside a <script> blob the
  parser drops, or a document truncated beyond the model window. Zero matches
  is a deliberately conservative bar, so the warning stays quiet when the page
  simply phrases the answer differently. The graphs now pass their schema to
  ParseNode so it has the field names available.

Verified against the URLs from the issue: the 404 now logs "Received HTTP 404
for .../Timpson_(company); the scraped content is likely an error page" before
returning NA, while the corrected URL stays silent and answers 1865.

Also drops three dead imports from smart_scraper_multi_batch_graph.py, which
ruff blocks on now that the file is touched.

Fixes #1102

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Marco Vinciguerra
2026-08-23 15:11:48 +02:00
parent 083c54fcb5
commit adc92f7eff
12 changed files with 640 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/test-suite.yml
+1
View File
@@ -42,6 +42,7 @@ jobs:
tests/test_batch_api.py
tests/test_csv_scraper_multi_graph.py
tests/test_depth_search_graph.py
tests/test_error_page_detection.py
tests/test_json_scraper_graph.py
tests/test_minimax_models.py
tests/test_scrape_do.py
scrapegraphai/docloaders/chromium.py
+32 -3
View File
@@ -10,6 +10,32 @@ from ..utils import Proxy, dynamic_import, get_logger, parse_or_search_proxy
logger = get_logger("web-loader")
def _warn_on_error_status(response: Any, url: str) -> None:
"""Log a warning when a navigation returned an HTTP error status.
Playwright's ``page.goto()`` returns the main-frame ``Response``, but the
scrapers only keep ``page.content()``. Without this check an error page
(404, 403, 500, a captcha wall, a login redirect) is indistinguishable
from the intended document once it reaches the LLM, which then produces a
confidently wrong answer with no signal that anything went wrong.
This mirrors the behaviour of the ``use_soup=True`` path in ``FetchNode``:
it warns rather than raising, so scraping error pages on purpose keeps
working.
Args:
response: The ``Response`` returned by ``page.goto()``; may be ``None``
(for example on a same-document navigation) or lack a usable status.
url: The URL that was requested, used in the warning message.
"""
status = getattr(response, "status", None)
if isinstance(status, int) and status >= 400:
logger.warning(
f"Received HTTP {status} for {url}; the scraped content is likely "
"an error page, not the intended document."
)
class ChromiumLoader:
"""Scrapes HTML pages from URLs using a (headless) instance of the
Chromium web driver with proxy protection.
@@ -251,7 +277,8 @@ class ChromiumLoader:
context = await browser.new_context()
await Malenia.apply_stealth(context)
page = await context.new_page()
await page.goto(url, wait_until="domcontentloaded")
response = await page.goto(url, wait_until="domcontentloaded")
_warn_on_error_status(response, url)
await page.wait_for_load_state(self.load_state)
previous_height = None
@@ -364,7 +391,8 @@ class ChromiumLoader:
)
await Malenia.apply_stealth(context)
page = await context.new_page()
await page.goto(url, wait_until="domcontentloaded")
response = await page.goto(url, wait_until="domcontentloaded")
_warn_on_error_status(response, url)
await page.wait_for_load_state(self.load_state)
results = await page.content()
logger.info("Content scraped")
@@ -421,7 +449,8 @@ class ChromiumLoader:
storage_state=self.storage_state
)
page = await context.new_page()
await page.goto(url, wait_until="networkidle")
response = await page.goto(url, wait_until="networkidle")
_warn_on_error_status(response, url)
results = await page.content()
logger.info("Content scraped after JavaScript rendering")
return results
scrapegraphai/graphs/code_generator_graph.py
+5 -1
View File
@@ -93,7 +93,11 @@ class CodeGeneratorGraph(AbstractGraph):
parse_node = ParseNode(
input="doc",
output=["parsed_doc"],
node_config={"llm_model": self.llm_model, "chunk_size": self.model_token},
node_config={
"llm_model": self.llm_model,
"chunk_size": self.model_token,
"schema": self.schema,
},
)
generate_validation_answer_node = GenerateAnswerNode(
scrapegraphai/graphs/document_scraper_graph.py
+1
View File
@@ -76,6 +76,7 @@ class DocumentScraperGraph(AbstractGraph):
"parse_html": False,
"chunk_size": self.model_token,
"llm_model": self.llm_model,
"schema": self.schema,
},
)
generate_answer_node = GenerateAnswerNode(
scrapegraphai/graphs/omni_scraper_graph.py
+1
View File
@@ -83,6 +83,7 @@ class OmniScraperGraph(AbstractGraph):
"chunk_size": self.model_token,
"parse_urls": True,
"llm_model": self.llm_model,
"schema": self.schema,
},
)
scrapegraphai/graphs/script_creator_graph.py
+1
View File
@@ -82,6 +82,7 @@ class ScriptCreatorGraph(AbstractGraph):
"chunk_size": self.model_token,
"parse_html": False,
"llm_model": self.llm_model,
"schema": self.schema,
},
)
scrapegraphai/graphs/smart_scraper_graph.py
+6 -1
View File
@@ -110,7 +110,11 @@ class SmartScraperGraph(AbstractGraph):
parse_node = ParseNode(
input="doc",
output=["parsed_doc"],
node_config={"llm_model": self.llm_model, "chunk_size": self.model_token},
node_config={
"llm_model": self.llm_model,
"chunk_size": self.model_token,
"schema": self.schema,
},
)
generate_answer_node = GenerateAnswerNode(
@@ -152,6 +156,7 @@ class SmartScraperGraph(AbstractGraph):
node_config={
"llm_model": self.llm_model,
"chunk_size": self.model_token,
"schema": self.schema,
},
)
scrapegraphai/graphs/smart_scraper_lite_graph.py
+5 -1
View File
@@ -74,7 +74,11 @@ class SmartScraperLiteGraph(AbstractGraph):
parse_node = ParseNode(
input="doc",
output=["parsed_doc"],
node_config={"llm_model": self.llm_model, "chunk_size": self.model_token},
node_config={
"llm_model": self.llm_model,
"chunk_size": self.model_token,
"schema": self.schema,
},
)
return BaseGraph(
scrapegraphai/graphs/smart_scraper_multi_batch_graph.py
+2 -3
View File
@@ -5,9 +5,8 @@ A scraping pipeline that uses the OpenAI Batch API for LLM calls,
providing 50% cost savings compared to real-time API calls.
"""
import asyncio
from copy import deepcopy
from typing import Dict, List, Optional, Type
from typing import List, Optional, Type
from pydantic import BaseModel
@@ -17,7 +16,6 @@ from ..nodes.merge_answers_node import MergeAnswersNode
from ..utils.copy import safe_deepcopy
from .abstract_graph import AbstractGraph
from .base_graph import BaseGraph
from .smart_scraper_graph import SmartScraperGraph
class _FetchParseOnlyGraph(AbstractGraph):
@@ -57,6 +55,7 @@ class _FetchParseOnlyGraph(AbstractGraph):
node_config={
"llm_model": self.llm_model,
"chunk_size": self.model_token,
"schema": self.schema,
},
)
scrapegraphai/graphs/speech_graph.py
+5 -1
View File
@@ -70,7 +70,11 @@ class SpeechGraph(AbstractGraph):
parse_node = ParseNode(
input="doc",
output=["parsed_doc"],
node_config={"chunk_size": self.model_token, "llm_model": self.llm_model},
node_config={
"chunk_size": self.model_token,
"llm_model": self.llm_model,
"schema": self.schema,
},
)
generate_answer_node = GenerateAnswerNode(
scrapegraphai/nodes/parse_node.py
+255 -1
View File
@@ -3,7 +3,7 @@ ParseNode Module
"""
import re
from typing import List, Optional, Tuple
from typing import List, Optional, Set, Tuple, get_args
from urllib.parse import urljoin
from langchain_community.document_transformers import Html2TextTransformer
@@ -37,6 +37,93 @@ class ParseNode(BaseNode):
)
relative_url_pattern = re.compile(r"[\(](/[^\(\)\s]*)")
# Words carrying no discriminative signal, dropped before checking whether the
# parsed document contains any evidence of what the user asked for.
prompt_stopwords = frozenset(
{
"about",
"after",
"all",
"also",
"and",
"any",
"are",
"been",
"both",
"but",
"can",
"content",
"data",
"does",
"each",
"every",
"extract",
"find",
"for",
"from",
"get",
"give",
"has",
"have",
"here",
"how",
"html",
"info",
"information",
"into",
"its",
"json",
"list",
"many",
"more",
"much",
"must",
"not",
"only",
"out",
"page",
"please",
"provide",
"retrieve",
"return",
"scrape",
"site",
"some",
"such",
"text",
"that",
"the",
"their",
"them",
"then",
"there",
"these",
"they",
"this",
"those",
"url",
"was",
"web",
"webpage",
"website",
"were",
"what",
"when",
"where",
"which",
"who",
"why",
"will",
"with",
"would",
"you",
"your",
}
)
word_pattern = re.compile(r"[a-zA-Z][a-zA-Z0-9]{2,}")
camel_case_pattern = re.compile(r"(?<=[a-z0-9])(?=[A-Z])")
def __init__(
self,
input: str,
@@ -58,6 +145,7 @@ class ParseNode(BaseNode):
self.llm_model = node_config.get("llm_model")
self.chunk_size = node_config.get("chunk_size")
self.schema = node_config.get("schema")
def execute(self, state: dict) -> dict:
"""
@@ -119,6 +207,8 @@ class ParseNode(BaseNode):
text=docs_transformed, chunk_size=chunk_size
)
self._warn_if_content_lacks_requested_fields(chunks, state.get("user_prompt"))
state.update({self.output[0]: chunks})
state.update({"parsed_doc": chunks})
@@ -128,6 +218,170 @@ class ParseNode(BaseNode):
return state
def _warn_if_content_lacks_requested_fields(
self, chunks: List[str], user_prompt: Optional[str]
) -> None:
"""
Warns when the parsed content holds no trace of what the user asked for.
An HTTP status check catches error pages, but a perfectly valid 200 page
can still reach the LLM without the requested data: content behind
JavaScript that never rendered, a field living in a ``<script>`` blob the
parser drops, or a document truncated beyond the model window. In each
case the LLM answers ``NA`` and the run looks clean.
This is a deterministic, LLM-free check: it collects the terms the user
asked about (schema field names and the significant words of the prompt)
and warns only when *none* of them appear in the parsed text. Zero
matches is a deliberately conservative bar, so the warning stays quiet
for legitimate runs where the answer is phrased differently from the
question.
Args:
chunks (List[str]): The parsed content chunks about to be handed downstream.
user_prompt (Optional[str]): The user's request, when available in the state.
"""
texts = [chunk for chunk in chunks if isinstance(chunk, str)]
total_length = sum(len(text) for text in texts)
if not any(text.strip() for text in texts):
self.logger.warning(
"The parsed content is empty; the model will be asked to answer "
"from nothing. Check that the source was fetched correctly."
)
return
expected_terms = self._collect_expected_terms(user_prompt)
if not expected_terms:
return
# Chunks overlap, so a term split across a boundary is still found in one
# of them; searching chunk by chunk avoids rebuilding the whole document.
for text in texts:
lowered = text.lower()
if any(term in lowered for term in expected_terms):
return
self.logger.warning(
f"None of the requested terms {sorted(expected_terms)} appear in the "
f"parsed content ({total_length} chars). The source may be an error "
"page, may render its content with JavaScript, or the relevant "
"section may have been dropped while parsing; the model will most "
"likely answer NA."
)
def _collect_expected_terms(self, user_prompt: Optional[str]) -> Set[str]:
"""
Builds the set of lowercase terms that evidence the user's request.
Args:
user_prompt (Optional[str]): The user's request, when available.
Returns:
Set[str]: Significant terms drawn from the schema field names and the
prompt; empty when nothing discriminative could be derived.
"""
terms = self._schema_field_terms(self.schema)
if isinstance(user_prompt, str):
terms |= self._significant_words(user_prompt)
return terms
@classmethod
def _significant_words(cls, text: str) -> Set[str]:
"""
Extracts the discriminative words of a piece of text.
Args:
text (str): The text to tokenize.
Returns:
Set[str]: Lowercase words at least three characters long that are not
generic scraping vocabulary.
"""
words = cls.word_pattern.findall(cls.camel_case_pattern.sub(" ", text))
return {
word.lower() for word in words if word.lower() not in cls.prompt_stopwords
}
@classmethod
def _schema_field_terms(cls, schema, _depth: int = 0) -> Set[str]:
"""
Extracts the field names of an output schema, recursing into nested ones.
Supports the schema flavours the library accepts: Pydantic models, plain
JSON Schema dictionaries, and lists of either.
Args:
schema: The output schema, in any of the supported forms; may be None.
_depth (int): Internal recursion guard for deeply nested schemas.
Returns:
Set[str]: Lowercase terms derived from the field names, with
``snake_case`` and ``camelCase`` names split into their parts.
"""
if schema is None or _depth > 5:
return set()
names: List[str] = []
terms: Set[str] = set()
if isinstance(schema, dict):
properties = schema.get("properties")
if isinstance(properties, dict):
names.extend(str(key) for key in properties)
for value in properties.values():
terms |= cls._schema_field_terms(value, _depth + 1)
items = schema.get("items")
if items is not None:
terms |= cls._schema_field_terms(items, _depth + 1)
elif isinstance(schema, (list, tuple, set)):
for entry in schema:
terms |= cls._schema_field_terms(entry, _depth + 1)
else:
model_fields = getattr(schema, "model_fields", None) # pydantic v2
if model_fields is None:
model_fields = getattr(schema, "__fields__", None) # pydantic v1
if isinstance(model_fields, dict):
names.extend(str(key) for key in model_fields)
for field in model_fields.values():
annotation = getattr(field, "annotation", None)
for nested in cls._nested_models(annotation):
terms |= cls._schema_field_terms(nested, _depth + 1)
for name in names:
terms |= cls._significant_words(name.replace("_", " "))
return terms
@staticmethod
def _nested_models(annotation) -> List:
"""
Finds the Pydantic models reachable from a field annotation.
Unwraps the typing containers schemas commonly use — ``List[Item]``,
``Optional[Item]``, ``Dict[str, Item]`` — so nested field names are not
lost.
Args:
annotation: The annotation of a Pydantic field; may be None.
Returns:
List: The Pydantic model classes found in the annotation.
"""
if annotation is None:
return []
if hasattr(annotation, "model_fields") or hasattr(annotation, "__fields__"):
return [annotation]
return [
arg
for arg in get_args(annotation)
if hasattr(arg, "model_fields") or hasattr(arg, "__fields__")
]
def _extract_urls(self, text: str, source: str) -> Tuple[List[str], List[str]]:
"""
Extracts URLs from the given text.
tests/test_error_page_detection.py
+326
View File
@@ -0,0 +1,326 @@
"""
Tests for the silent-error-page guards added for issue #1102.
Two independent signals are covered:
1. ``ChromiumLoader`` warns when ``page.goto()`` reports an HTTP error status,
so a 404/403/500 page is no longer handed to the LLM as if it were the
intended document.
2. ``ParseNode`` warns when the parsed content contains no trace of what the
user asked for, which also catches 200 pages whose content never rendered.
"""
import asyncio
from typing import List
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from langchain_core.documents import Document
from pydantic import BaseModel
from scrapegraphai.docloaders.chromium import ChromiumLoader, _warn_on_error_status
from scrapegraphai.nodes import ParseNode
from scrapegraphai.utils.logging import set_propagation, unset_propagation
# --------------------------------------------------------------------------- #
# ChromiumLoader: HTTP status awareness on every page.goto() call site
# --------------------------------------------------------------------------- #
class _MockPage:
def __init__(self, status):
response = MagicMock()
response.status = status
self.goto = AsyncMock(return_value=response)
self.wait_for_load_state = AsyncMock()
self.content = AsyncMock(return_value="<html>error page</html>")
self.evaluate = AsyncMock(return_value=1000)
self.mouse = MagicMock()
self.mouse.wheel = AsyncMock()
@pytest.fixture
def playwright_with_status():
"""Patch playwright so page.goto() returns a response with a given status."""
def _factory(status):
page = _MockPage(status)
context = MagicMock()
context.new_page = AsyncMock(return_value=page)
browser = MagicMock()
browser.new_context = AsyncMock(return_value=context)
browser.close = AsyncMock()
pw = MagicMock()
pw.chromium.launch = AsyncMock(return_value=browser)
pw.firefox.launch = AsyncMock(return_value=browser)
cm = MagicMock()
cm.__aenter__ = AsyncMock(return_value=pw)
cm.__aexit__ = AsyncMock(return_value=False)
return cm, page
return _factory
@pytest.mark.parametrize("status", [404, 403, 500, 503])
def test_ascrape_playwright_warns_on_error_status(
playwright_with_status, status, caplog
):
cm, _ = playwright_with_status(status)
loader = ChromiumLoader(["https://example.com/missing"], backend="playwright")
with (
patch("playwright.async_api.async_playwright", return_value=cm),
patch("undetected_playwright.Malenia.apply_stealth", new=AsyncMock()),
caplog.at_level("WARNING"),
):
asyncio.run(loader.ascrape_playwright("https://example.com/missing"))
assert f"Received HTTP {status}" in caplog.text
assert "likely an error page" in caplog.text
def test_ascrape_playwright_silent_on_success(playwright_with_status, caplog):
cm, _ = playwright_with_status(200)
loader = ChromiumLoader(["https://example.com"], backend="playwright")
with (
patch("playwright.async_api.async_playwright", return_value=cm),
patch("undetected_playwright.Malenia.apply_stealth", new=AsyncMock()),
caplog.at_level("WARNING"),
):
asyncio.run(loader.ascrape_playwright("https://example.com"))
assert "Received HTTP" not in caplog.text
def test_ascrape_with_js_support_warns_on_error_status(playwright_with_status, caplog):
cm, _ = playwright_with_status(404)
loader = ChromiumLoader(
["https://example.com/missing"], backend="playwright", requires_js_support=True
)
with (
patch("playwright.async_api.async_playwright", return_value=cm),
caplog.at_level("WARNING"),
):
asyncio.run(loader.ascrape_with_js_support("https://example.com/missing"))
assert "Received HTTP 404" in caplog.text
def test_ascrape_playwright_scroll_warns_on_error_status(
playwright_with_status, caplog
):
cm, page = playwright_with_status(404)
# Stop the scroll loop immediately: same height twice means "bottom reached".
page.evaluate = AsyncMock(return_value=1000)
loader = ChromiumLoader(["https://example.com/missing"], backend="playwright")
with (
patch("playwright.async_api.async_playwright", return_value=cm),
patch("undetected_playwright.Malenia.apply_stealth", new=AsyncMock()),
caplog.at_level("WARNING"),
):
asyncio.run(
loader.ascrape_playwright_scroll(
"https://example.com/missing", scroll=5000, sleep=0.01, timeout=1
)
)
assert "Received HTTP 404" in caplog.text
def test_warn_on_error_status_tolerates_missing_response(caplog):
"""page.goto() returns None for same-document navigations; that is not an error."""
with caplog.at_level("WARNING"):
_warn_on_error_status(None, "https://example.com")
_warn_on_error_status(MagicMock(status=None), "https://example.com")
assert "Received HTTP" not in caplog.text
# --------------------------------------------------------------------------- #
# ParseNode: warn when the parsed content holds no trace of the request
# --------------------------------------------------------------------------- #
@pytest.fixture
def library_logs_propagate():
"""Let caplog see records from the library root logger.
``scrapegraphai`` disables propagation by default so it does not pollute the
host application's logging; the nodes log through that root logger.
"""
set_propagation()
yield
unset_propagation()
class Company(BaseModel):
company_name: str
foundingYear: int
class Employee(BaseModel):
employee_name: str
salary: str
class Payroll(BaseModel):
employees: List[Employee]
def _parse_node(**node_config):
config = {"chunk_size": 4096, "verbose": False}
config.update(node_config)
return ParseNode(input="doc", output=["parsed_doc"], node_config=config)
def _run(node, html, user_prompt):
state = {"doc": [Document(page_content=html)], "user_prompt": user_prompt}
return node.execute(state)
WIKIPEDIA_404 = (
"<html><body><p>Jump to content. Main menu. Navigation. "
"Wikipedia does not have an article with this exact name.</p></body></html>"
)
TIMPSON_PAGE = (
"<html><body><p>Timpson is a British retailer founded in 1865 "
"by William Timpson.</p></body></html>"
)
# A shell page whose real content is rendered client-side: HTTP 200, no error,
# and nothing for the LLM to work with.
JS_SHELL_PAGE = "<html><body><div id='root'>Loading...</div></body></html>"
STRUCTURED_PAGE = (
"<html><body><p>Company name: Timpson</p>"
"<p>Founding year: 1865</p></body></html>"
)
def test_warns_when_no_requested_term_is_present(library_logs_propagate, caplog):
node = _parse_node()
with caplog.at_level("WARNING"):
_run(node, WIKIPEDIA_404, "What is the founding year of Timpson?")
assert "None of the requested terms" in caplog.text
def test_silent_when_the_content_holds_the_answer(library_logs_propagate, caplog):
node = _parse_node()
with caplog.at_level("WARNING"):
_run(node, TIMPSON_PAGE, "What is the founding year of Timpson?")
assert "None of the requested terms" not in caplog.text
def test_schema_field_names_count_as_requested_terms(library_logs_propagate, caplog):
"""A page that never rendered is a 200, so only the schema can flag it."""
node = _parse_node(schema=Company)
with caplog.at_level("WARNING"):
_run(node, JS_SHELL_PAGE, None)
assert "None of the requested terms" in caplog.text
# snake_case and camelCase names are split into their parts
assert "founding" in caplog.text
assert "company" in caplog.text
assert "year" in caplog.text
def test_schema_match_keeps_the_check_quiet(library_logs_propagate, caplog):
node = _parse_node(schema=Company)
with caplog.at_level("WARNING"):
_run(node, STRUCTURED_PAGE, None)
assert "None of the requested terms" not in caplog.text
def test_schema_terms_can_rescue_an_unspecific_prompt(library_logs_propagate, caplog):
"""The union of prompt and schema terms only ever makes the check quieter."""
node = _parse_node(schema=Company)
with caplog.at_level("WARNING"):
_run(node, STRUCTURED_PAGE, "Extract everything you can find")
assert "None of the requested terms" not in caplog.text
def test_nested_pydantic_models_are_unwrapped(library_logs_propagate, caplog):
"""List[Item] and friends must not hide the nested field names."""
node = _parse_node(schema=Payroll)
with caplog.at_level("WARNING"):
_run(node, JS_SHELL_PAGE, None)
assert "salary" in caplog.text
assert "employee" in caplog.text
def test_json_schema_dict_is_supported(library_logs_propagate, caplog):
schema = {
"type": "object",
"properties": {
"founding_year": {"type": "integer"},
"locations": {
"type": "array",
"items": {
"type": "object",
"properties": {"postcode": {"type": "string"}},
},
},
},
}
node = _parse_node(schema=schema)
with caplog.at_level("WARNING"):
_run(node, JS_SHELL_PAGE, None)
assert "postcode" in caplog.text
assert "locations" in caplog.text
def test_no_prompt_and_no_schema_produces_no_warning(library_logs_propagate, caplog):
node = _parse_node()
with caplog.at_level("WARNING"):
_run(node, WIKIPEDIA_404, None)
assert "None of the requested terms" not in caplog.text
def test_generic_prompt_words_alone_do_not_trigger_the_warning(
library_logs_propagate, caplog
):
"""A prompt made only of scraping vocabulary carries no signal to check."""
node = _parse_node()
with caplog.at_level("WARNING"):
_run(node, WIKIPEDIA_404, "Extract all the information from this webpage")
assert "None of the requested terms" not in caplog.text
def test_empty_parsed_content_is_reported(library_logs_propagate, caplog):
node = _parse_node()
with caplog.at_level("WARNING"):
_run(node, "", "What is the founding year of Timpson?")
assert "parsed content is empty" in caplog.text
def test_state_is_unchanged_by_the_guard():
"""The guard only logs; the parsed chunks must reach the state as before."""
node = _parse_node(schema=Company)
state = _run(node, TIMPSON_PAGE, "What is the founding year of Timpson?")
assert state["parsed_doc"]
assert "1865" in "".join(state["parsed_doc"])