| """Integration tests for the *grounding* guarantees that make AgAdvisor safe for |
| regulatory (pesticide-label) guidance. These are the behaviours a paper would |
| report in Table 1: abstention on unknown products, page-level citations on every |
| answer, product-scoped retrieval (no cross-product leakage), and determinism. |
| |
| Runs offline over the committed CDMS index via the in-memory-Qdrant fixture |
| (no OpenAI/Tavily/Docker). Skips cleanly if the index isn't present. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import pytest |
|
|
| pytestmark = pytest.mark.integration |
|
|
|
|
| def _available_products(): |
| from src.cdms.product_catalog import get_catalog |
|
|
| return sorted(get_catalog().available_products()) |
|
|
|
|
| def test_abstains_on_unindexed_product(in_memory_rag): |
| """A product we don't have must yield an honest empty result (no substituting |
| another product's label) — the core anti-hallucination guarantee.""" |
| results = in_memory_rag.search( |
| query="application rate", product_name="definitely-not-a-real-product-xyz", |
| limit=5, score_threshold=0.0, |
| ) |
| assert results == [] |
|
|
|
|
| def test_indexed_product_returns_page_citations(in_memory_rag): |
| products = _available_products() |
| if not products: |
| pytest.skip("no indexed products in catalog") |
| target = products[0] |
| results = in_memory_rag.search( |
| query=f"{target} application information", product_name=target, |
| limit=5, score_threshold=0.0, |
| ) |
| assert results, f"expected grounded results for indexed product {target!r}" |
| |
| for r in results: |
| assert r.get("source_file"), "result missing source_file citation" |
| assert int(r.get("page_number", 0)) >= 1, "result missing valid page number" |
|
|
|
|
| def test_product_filter_prevents_cross_product_leakage(in_memory_rag): |
| """When a product is specified, retrieved chunks must belong to that product |
| — the fix for the ISA 'wrong herbicide' bug.""" |
| from src.cdms.product_catalog import normalize_filename |
|
|
| products = _available_products() |
| if not products: |
| pytest.skip("no indexed products in catalog") |
| target = products[0] |
| results = in_memory_rag.search( |
| query="rate and mixing instructions", product_name=target, |
| limit=8, score_threshold=0.0, |
| ) |
| if not results: |
| pytest.skip(f"no results for {target!r} to check scoping") |
| got = {normalize_filename(r.get("source_file", "")) for r in results} |
| assert got == {target}, f"cross-product leakage: expected only {target!r}, got {got}" |
|
|
|
|
| def test_search_is_deterministic(in_memory_rag): |
| products = _available_products() |
| if not products: |
| pytest.skip("no indexed products") |
| target = products[0] |
| kw = dict(query="directions for use", product_name=target, limit=5, score_threshold=0.0) |
| a = in_memory_rag.search(**kw) |
| b = in_memory_rag.search(**kw) |
| assert [r.get("source_file") for r in a] == [r.get("source_file") for r in b] |
| assert [r.get("page_number") for r in a] == [r.get("page_number") for r in b] |
|
|
|
|
| @pytest.mark.parametrize("query", ["", " ", "🌾🌾🌾", "a" * 3000]) |
| def test_edge_case_queries_do_not_crash(in_memory_rag, query): |
| """Empty, whitespace, emoji-only and oversized queries return a list, never raise.""" |
| out = in_memory_rag.search(query=query, limit=5, score_threshold=0.0) |
| assert isinstance(out, list) |
|
|
