michele
4df04df5d4
Add /api/library datasheet import and component GET/PUT with no exam. Reorganize pytest into datasheet, library, schematic, PCB, and AF+AI. Document Rust criteria (none chosen; no rustup) and coding conformity.
225 lines
8.1 KiB
Python
225 lines
8.1 KiB
Python
"""Tests for get_datasheet_excerpt: safety guards, budget cap, cache."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from tests.paths import SIMPLE_PROJECT, TAXONOMY
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from pypdf import PdfWriter
|
||
|
||
from backend.periscopex.models import DesignGraph
|
||
from backend.periscopex.utils import safe_mpn
|
||
from backend.periscopex.review_tools import (
|
||
EXCERPT_TOPICS,
|
||
ExcerptState,
|
||
execute_tool,
|
||
get_datasheet_excerpt,
|
||
)
|
||
from backend.services.llm.types import PdfBlock
|
||
from backend.services.review_session import _signal_neighbors
|
||
|
||
GRAPH = SIMPLE_PROJECT / "design_graph.json"
|
||
IC_MPNS = {
|
||
"U1": "SPX3819M5-L-3-3/TR",
|
||
"U2": "CH340E",
|
||
"U3": "MSPM0G3507SPTR",
|
||
}
|
||
|
||
|
||
def _blank_pdf(path: Path, pages: int = 3) -> None:
|
||
w = PdfWriter()
|
||
for _ in range(pages):
|
||
w.add_blank_page(width=200, height=200)
|
||
with path.open("wb") as fh:
|
||
w.write(fh)
|
||
|
||
|
||
@pytest.fixture
|
||
def graph_and_pdfs(tmp_path):
|
||
graph = DesignGraph.model_validate(json.loads(GRAPH.read_text()))
|
||
pdf_dir = tmp_path / "datasheets"
|
||
pdf_dir.mkdir()
|
||
for mpn in IC_MPNS.values():
|
||
_blank_pdf(pdf_dir / f"{safe_mpn(mpn)}.pdf")
|
||
return graph, pdf_dir
|
||
|
||
|
||
def _state(graph, current_ic, pdf_dir, **kwargs):
|
||
return ExcerptState(
|
||
current_ic=current_ic,
|
||
connected_designators=_signal_neighbors(graph, current_ic),
|
||
graph=graph,
|
||
pdf_dir=pdf_dir,
|
||
storage=None,
|
||
**kwargs,
|
||
)
|
||
|
||
|
||
def test_rejects_non_neighbor(graph_and_pdfs):
|
||
"""U1's review cannot fetch U2/U3 — they're not signal neighbors."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
state = _state(graph, "U1", pdf_dir)
|
||
text, attachment = get_datasheet_excerpt(
|
||
graph, {}, "U2", "absolute_max", state,
|
||
)
|
||
assert attachment is None
|
||
assert "not a signal neighbor" in text
|
||
assert state.fetch_count == 0
|
||
|
||
|
||
def test_rejects_self(graph_and_pdfs):
|
||
"""The IC under review can't excerpt its own datasheet — it's already in
|
||
context. Steer the model to use the existing PDF."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
state = _state(graph, "U2", pdf_dir)
|
||
text, attachment = get_datasheet_excerpt(
|
||
graph, {}, "U2", "absolute_max", state,
|
||
)
|
||
assert attachment is None
|
||
assert "already reviewing" in text
|
||
|
||
|
||
def test_rejects_unknown_topic(graph_and_pdfs):
|
||
"""Bad topic enum returns a clean error listing valid topics."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
state = _state(graph, "U2", pdf_dir)
|
||
text, attachment = get_datasheet_excerpt(
|
||
graph, {}, "U3", "bogus_topic", state,
|
||
)
|
||
assert attachment is None
|
||
assert "Unknown topic" in text
|
||
assert "absolute_max" in text
|
||
|
||
|
||
def test_returns_pdfblock_for_valid_neighbor(graph_and_pdfs):
|
||
"""U2's review can fetch U3 (a signal neighbor) — returns PdfBlock."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
state = _state(graph, "U2", pdf_dir)
|
||
text, attachment = get_datasheet_excerpt(
|
||
graph, {}, "U3", "pin_voltage_levels", state,
|
||
)
|
||
assert isinstance(attachment, PdfBlock)
|
||
assert attachment.cacheable is True
|
||
assert "U3" in text
|
||
assert "pin_voltage_levels" in text
|
||
assert state.fetch_count == 1
|
||
assert state.page_count > 0
|
||
|
||
|
||
def test_budget_cap_fetch_count(graph_and_pdfs):
|
||
"""After fetch_budget hits, further fetches are rejected with budget message."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
# Tight budget so the test is fast
|
||
state = _state(graph, "U3", pdf_dir, fetch_budget=2, page_budget=100)
|
||
# Two valid fetches succeed
|
||
for topic in ("absolute_max", "pin_voltage_levels"):
|
||
_, att = get_datasheet_excerpt(graph, {}, "U2", topic, state)
|
||
assert att is not None, f"topic {topic} should have succeeded"
|
||
# Third fetch should be budget-blocked
|
||
text, attachment = get_datasheet_excerpt(
|
||
graph, {}, "U2", "electrical_characteristics", state,
|
||
)
|
||
assert attachment is None
|
||
assert "budget exhausted" in text
|
||
|
||
|
||
def test_per_neighbor_page_budget_blocks_third_topic_on_same_neighbor(graph_and_pdfs):
|
||
"""A single neighbor can be fetched up to its per-neighbor page budget,
|
||
then further topics on it are blocked — bounds one interface's cost
|
||
without touching the global budget."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
# Blank PDFs fall back to 3 pages/fetch; budget of 4 admits the first two
|
||
# fetches (0→3, 3→6) and blocks the third once the neighbor is over.
|
||
state = _state(
|
||
graph, "U3", pdf_dir,
|
||
fetch_budget=10, page_budget=1000, per_neighbor_page_budget=4,
|
||
)
|
||
_, a1 = get_datasheet_excerpt(graph, {}, "U2", "absolute_max", state)
|
||
assert a1 is not None
|
||
_, a2 = get_datasheet_excerpt(graph, {}, "U2", "pin_voltage_levels", state)
|
||
assert a2 is not None
|
||
text, a3 = get_datasheet_excerpt(graph, {}, "U2", "electrical_characteristics", state)
|
||
assert a3 is None
|
||
assert "Per-neighbor" in text and "U2" in text
|
||
|
||
|
||
def test_global_page_budget_still_bounds_total_fanout(graph_and_pdfs):
|
||
"""The global page budget is a hard ceiling even when a neighbor is under
|
||
its per-neighbor sub-budget — bounds hub-IC fan-out."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
state = _state(
|
||
graph, "U3", pdf_dir,
|
||
fetch_budget=10, page_budget=2, per_neighbor_page_budget=100,
|
||
)
|
||
_, a1 = get_datasheet_excerpt(graph, {}, "U2", "absolute_max", state)
|
||
assert a1 is not None # capped to 2 pages → page_count hits global budget
|
||
text, a2 = get_datasheet_excerpt(graph, {}, "U2", "pin_voltage_levels", state)
|
||
assert a2 is None
|
||
assert "page budget exhausted" in text
|
||
|
||
|
||
def test_cache_shared_across_states(graph_and_pdfs):
|
||
"""Same (designator, topic, ds_md5) → second fetch reuses trimmed PDF.
|
||
|
||
Mirrors the cross-IC case: U2's review fetches U3@abs_max, then U3's
|
||
review fetches U3@abs_max from a separate ExcerptState sharing the same
|
||
cache dict.
|
||
"""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
shared_cache: dict = {}
|
||
s1 = _state(graph, "U2", pdf_dir, cache=shared_cache)
|
||
s2 = _state(graph, "U3", pdf_dir, cache=shared_cache)
|
||
_, att1 = get_datasheet_excerpt(graph, {}, "U3", "absolute_max", s1)
|
||
assert att1 is not None
|
||
path1 = att1.path
|
||
# U3's review reusing the same neighbor (loopback: U3 fetching itself is
|
||
# blocked, so use a different IC). We assert path identity for the U2
|
||
# case from s1's perspective is preserved across a fresh state with the
|
||
# same cache by re-fetching from s1's neighbor U3 with state s_other.
|
||
s_other = _state(graph, "U2", pdf_dir, cache=shared_cache)
|
||
_, att2 = get_datasheet_excerpt(graph, {}, "U3", "absolute_max", s_other)
|
||
assert att2 is not None
|
||
assert att2.path == path1 # cache hit — same trimmed file
|
||
|
||
|
||
def test_no_state_returns_clean_error(graph_and_pdfs):
|
||
"""Calling without state (regression guard) returns an explicit message."""
|
||
graph, _ = graph_and_pdfs
|
||
text, attachment = get_datasheet_excerpt(graph, {}, "U3", "absolute_max", None)
|
||
assert attachment is None
|
||
assert "without per-review state" in text
|
||
|
||
|
||
def test_execute_tool_dispatcher_routes_correctly(graph_and_pdfs):
|
||
"""The dispatcher recognises the new tool name and forwards state."""
|
||
graph, pdf_dir = graph_and_pdfs
|
||
state = _state(graph, "U2", pdf_dir)
|
||
text, attachment = execute_tool(
|
||
graph, {}, "get_datasheet_excerpt",
|
||
{"designator": "U3", "topic": "absolute_max"},
|
||
state=state,
|
||
)
|
||
assert isinstance(attachment, PdfBlock)
|
||
assert state.fetch_count == 1
|
||
|
||
|
||
def test_existing_tools_return_tuple_with_none_attachment(graph_and_pdfs):
|
||
"""Backward compat: existing tools now return (text, None)."""
|
||
graph, _ = graph_and_pdfs
|
||
text, att = execute_tool(graph, {}, "get_pintable", {"designator": "U2"})
|
||
assert isinstance(text, str) and text
|
||
assert att is None
|
||
|
||
|
||
def test_topics_cover_u2_001_use_case():
|
||
"""The U2-001 false positive needs to verify MSPM0G3507 PA1's 5V-tolerance
|
||
— assert that the pin_voltage_levels topic regex matches typical
|
||
datasheet phrasings for this."""
|
||
pat = EXCERPT_TOPICS["pin_voltage_levels"]
|
||
assert pat.search("This pin is 5V-tolerant under normal operating conditions.")
|
||
assert pat.search("VIH = 0.7 × VDD")
|
||
assert pat.search("Input voltage range: -0.3V to VDD+0.3V")
|