#!/usr/bin/env python3 """Stub test: markdown-heading section metadata in ingest_lib -- NO models. Proves: 1. parse_page_sections derives a running h1 > h2 stack from '#' lines and keeps per-page section paths (nested headings, empty-heading lines). 2. One kind='section' record per distinct heading, _summary_record pattern (id/kind/section_path/text/page/page_end/sha/doc_name). 3. _section_for_page gives chunks the section active on their start page. 4. retrieve.py chunk mapping carries section_path through. """ from __future__ import annotations import os import sys import tempfile from pathlib import Path REPO = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO)) os.environ["FACTORY_RAG_DATA"] = tempfile.mkdtemp(prefix="fr-section-") import config # noqa: E402 import ingest_lib # noqa: E402 def main() -> int: failures = [] # --- 1: running section stack over nested headings -------------------- pages_md = { 1: "# 4 Fuel System\n\nsome prose about pumps.\n", 2: "## 4.2 Fuel Filter\nfilter details.\n#tag-not-heading\n## \n## 4.3 Startup Procedure\nsteps here.\n", 3: "plain page with no headings.\n", 4: "# 5 Cooling\nnew chapter.\n", } by_page, recs = ingest_lib.parse_page_sections(pages_md) if by_page.get(1) != "4 Fuel System": failures.append(f"page1 section wrong: {by_page.get(1)!r}") # h2 pops back under the running h1; empty '## ' line must be ignored if by_page.get(2) != "4 Fuel System > 4.3 Startup Procedure": failures.append(f"page2 section wrong: {by_page.get(2)!r}") # heading-free page inherits the running section if by_page.get(3) != "4 Fuel System > 4.3 Startup Procedure": failures.append(f"page3 section wrong: {by_page.get(3)!r}") if by_page.get(4) != "5 Cooling": failures.append(f"page4 section wrong: {by_page.get(4)!r}") # distinct headings only: 4 Fuel System, 4.2 Fuel Filter, 4.3 Startup # Procedure, 5 Cooling (the empty '## ' and '#tag-not-heading' excluded) paths = sorted(r["section_path"] for r in recs) if paths != sorted([ "4 Fuel System", "4 Fuel System > 4.2 Fuel Filter", "4 Fuel System > 4.3 Startup Procedure", "5 Cooling"]): failures.append(f"section records wrong: {paths}") # --- 2: record shape follows the _summary_record pattern -------------- r = next(x for x in recs if x["section_path"] == "4 Fuel System") for key in ("id", "kind", "section_path", "text", "page", "page_end", "sha", "doc_name"): if key not in r: failures.append(f"section record missing key {key}") if r["kind"] != "section" or r["page"] != 1 or r["page_end"] != 1: failures.append(f"section record kind/pages wrong: {r}") if "4 Fuel System" not in r["text"] or "section heading" not in r["text"]: failures.append(f"section record text wrong: {r['text']!r}") # filled in by the caller, same slot usage as doc_summary records filled = dict(r, sha="deadbeef", doc_name="TM_5-811-6.pdf") if filled["sha"] != "deadbeef" or filled["doc_name"] != "TM_5-811-6.pdf": failures.append("section record sha/doc_name fill-in broken") # --- 3: chunk page -> active section ---------------------------------- if ingest_lib._section_for_page(1, by_page) != "4 Fuel System": failures.append("_section_for_page p1 wrong") if ingest_lib._section_for_page(3, by_page) != "4 Fuel System > 4.3 Startup Procedure": failures.append("_section_for_page p3 wrong") if ingest_lib._section_for_page(4, by_page) != "5 Cooling": failures.append("_section_for_page p4 wrong") # a chunk starting on a page BEFORE any heading: '' (nothing active yet) if ingest_lib._section_for_page(0, by_page) != "": failures.append("_section_for_page p0 must be '' before first heading") if ingest_lib._section_for_page(2, {}) != "": failures.append("_section_for_page with empty map must be ''") # no headings at all -> no records, empty paths by2, recs2 = ingest_lib.parse_page_sections({1: "just text\n", 2: "more\n"}) if recs2 or any(by2.values()): failures.append("heading-free doc must yield no sections") # --- 4: chunk mapping carries section_path --------------------------- import retrieve # noqa: E402 import types seam = types.ModuleType("sentence_transformers") class FakeST: def __init__(self, *a, **kw): pass def encode(self, texts, **kw): import numpy as np return np.zeros((len(texts), 4), dtype=np.float32) seam.SentenceTransformer = FakeST sys.modules["sentence_transformers"] = seam fake_points = [ {"text": "t1", "page": 2, "page_end": 2, "sha": "s1", "kind": "text", "doc_name": "d.pdf", "section_path": "4 Fuel System > 4.2 Fuel Filter", "score": 0.9}, {"text": "t2", "page": 3, "sha": "s1", "kind": "text", "doc_name": "d.pdf", "score": 0.4}, # no section_path key ] retr = retrieve.Retriever.__new__(retrieve.Retriever) retr.embedder = FakeST() retr._reranker = "passthrough" orig_rerank = retrieve.Retriever._rerank retrieve.Retriever._rerank = lambda self, q, pts, top_n: pts[:top_n] orig_hybrid = retrieve.Retriever._hybrid_query retrieve.Retriever._hybrid_query = lambda self, dq, sq, top_k: list(fake_points) orig_visual = retrieve.Retriever._visual_rrf retrieve.Retriever._visual_rrf = lambda self, q, pts, top_k: pts orig_gate = config.ENABLE_PRE_REFUSAL_GATE config.ENABLE_PRE_REFUSAL_GATE = False try: res = retr.retrieve("q", top_n=2) c0, c1 = res["chunks"][0], res["chunks"][1] if c0.get("section_path") != "4 Fuel System > 4.2 Fuel Filter": failures.append(f"mapping lost section_path: {c0}") if c1.get("section_path", None) != "": failures.append(f"mapping must default section_path to '': {c1}") finally: retrieve.Retriever._rerank = orig_rerank retrieve.Retriever._hybrid_query = orig_hybrid retrieve.Retriever._visual_rrf = orig_visual config.ENABLE_PRE_REFUSAL_GATE = orig_gate if failures: for f in failures: print(f"FAIL: {f}") return 1 print("PASS test_section_metadata_stub") return 0 if __name__ == "__main__": raise SystemExit(main())