e4fb0108d6
is_current() stat'd every source file bare, so a file removed between iterdir() and stat() aborted the whole extract run; a vanished source now just doesn't affect currency. The lazy Zotero snapshot stamped the snapshot with the source's mtime even when shutil.copy2 had failed, which would have passed a stale snapshot off as current on every later run. Copying moved into _copy_snapshot() so both callers know whether it worked. Docs: upsert_status spells out the merge rules (extra_json exempt from empty-keeps-stored, tags/collections union-merged) and `backfill-comments --force` says it bypasses the docket seal only.
277 lines
10 KiB
Python
277 lines
10 KiB
Python
"""llm.source — lazy DocRefs: fingerprints without loads, docket skips,
|
|
lazy Zotero snapshot."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from bib.item import Item, Rule
|
|
from bib.store import Store
|
|
from llm.source import (
|
|
DocRef,
|
|
ZoteroPdfIndex,
|
|
iter_comment_refs,
|
|
iter_corpus_refs,
|
|
iter_rule_refs,
|
|
)
|
|
|
|
DOCKET = "CMS-2019-0111"
|
|
CID = f"{DOCKET}-0042"
|
|
COMBINED = f"---\ncomment_id: {CID}\ndocket_id: {DOCKET}\n---\n\nWe object.\n"
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
s = Store(":memory:", storage_dir=tmp_path / "storage")
|
|
key = s.create(
|
|
Item(
|
|
item_type="report",
|
|
title="A comment",
|
|
url=f"https://www.regulations.gov/comment/{CID}",
|
|
abstract="Inline.",
|
|
date_published="2019-09-27",
|
|
)
|
|
)
|
|
for tag in ("doctype:comment", "year:2019", f"reg-docket:{DOCKET}"):
|
|
s.add_tag(key, tag)
|
|
s._comment_key = key
|
|
return s
|
|
|
|
|
|
@pytest.fixture
|
|
def root(tmp_path):
|
|
d = tmp_path / DOCKET / CID
|
|
d.mkdir(parents=True)
|
|
(d / "combined.md").write_text(COMBINED)
|
|
return tmp_path
|
|
|
|
|
|
class TestCommentRefs:
|
|
def test_ref_has_fingerprint_and_lazy_load(self, store, root, monkeypatch):
|
|
reads = []
|
|
real = Path.read_text
|
|
monkeypatch.setattr(
|
|
Path,
|
|
"read_text",
|
|
lambda self, *a, **k: reads.append(self) or real(self, *a, **k),
|
|
)
|
|
refs = list(iter_comment_refs(store, docket=DOCKET, root=root))
|
|
assert len(refs) == 1
|
|
r = refs[0]
|
|
assert isinstance(r, DocRef)
|
|
assert (r.key, r.collection, r.docket) == (
|
|
store._comment_key,
|
|
"comments",
|
|
DOCKET,
|
|
)
|
|
assert len(r.fingerprint) > 64 and "|" in r.fingerprint
|
|
assert reads == [] # nothing read yet
|
|
doc = r.load()
|
|
assert "We object" in doc.text
|
|
assert doc.metadata["year"] == "2019" and doc.metadata["docket"] == DOCKET
|
|
assert reads # load() read combined.md
|
|
|
|
def test_fingerprint_changes_with_new_attachment(self, store, root):
|
|
f1 = next(iter_comment_refs(store, docket=DOCKET, root=root)).fingerprint
|
|
(root / DOCKET / CID / "attachment_1.pdf").write_bytes(b"%PDF")
|
|
f2 = next(iter_comment_refs(store, docket=DOCKET, root=root)).fingerprint
|
|
assert f1 != f2
|
|
|
|
def test_fingerprint_changes_with_updated_at(self, store, root):
|
|
f1 = next(iter_comment_refs(store, docket=DOCKET, root=root)).fingerprint
|
|
store._con().execute(
|
|
"UPDATE items SET updated_at='2030-01-01T00:00:00Z' WHERE key=?",
|
|
(store._comment_key,),
|
|
)
|
|
f2 = next(iter_comment_refs(store, docket=DOCKET, root=root)).fingerprint
|
|
assert f1 != f2
|
|
|
|
def test_skip_dockets_excludes_rows(self, store, root):
|
|
assert list(iter_comment_refs(store, root=root, skip_dockets={DOCKET})) == []
|
|
|
|
def test_unextracted_load_falls_back_to_abstract(self, store, tmp_path):
|
|
r = next(iter_comment_refs(store, docket=DOCKET, root=tmp_path))
|
|
assert r.load().text == "Inline."
|
|
|
|
def test_empty_comment_load_returns_none(self, store, tmp_path):
|
|
store._con().execute(
|
|
"UPDATE items SET abstract='' WHERE key=?", (store._comment_key,)
|
|
)
|
|
r = next(iter_comment_refs(store, docket=DOCKET, root=tmp_path))
|
|
assert r.load() is None
|
|
|
|
def test_single_query_for_year(self, store, root, monkeypatch):
|
|
"""No per-comment year lookups: exactly one SELECT on items for the listing."""
|
|
calls: list[str] = []
|
|
store._con().set_trace_callback(calls.append)
|
|
list(iter_comment_refs(store, docket=DOCKET, root=root))
|
|
store._con().set_trace_callback(None)
|
|
selects = [c for c in calls if c.lstrip().upper().startswith("SELECT")]
|
|
assert len(selects) == 1
|
|
|
|
|
|
def _anchored_rule(store) -> str:
|
|
"""A rule with one grabbed FR anchor paragraph (sha256 "abc")."""
|
|
key = store.create(Rule(title="R", url="https://fr/1", document_number="2019-1"))
|
|
store._con().execute(
|
|
"INSERT INTO fr_anchor_docs (item_key, document_number, html_url, start_page, end_page, fr_volume, sha256) VALUES (?,?,?,?,?,?,?)",
|
|
(key, "2019-1", "https://fr/1", 1, 2, 84, "abc"),
|
|
)
|
|
store._con().execute(
|
|
"INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)",
|
|
(key, 1, 1, 1, "Para one."),
|
|
)
|
|
return key
|
|
|
|
|
|
class TestRuleRefs:
|
|
def test_fingerprint_from_anchor_sha(self, store):
|
|
key = _anchored_rule(store)
|
|
refs = list(iter_rule_refs(store))
|
|
assert [r.key for r in refs] == [key]
|
|
assert refs[0].fingerprint.startswith("anchors:abc|")
|
|
assert refs[0].collection == "rules" and refs[0].docket is None
|
|
assert refs[0].load().text == "Para one."
|
|
|
|
def test_anchor_fingerprint_changes_with_updated_at(self, store):
|
|
"""The sha covers the FR body, not the item's own metadata."""
|
|
key = _anchored_rule(store)
|
|
f1 = next(iter_rule_refs(store)).fingerprint
|
|
store._con().execute(
|
|
"UPDATE items SET updated_at='2030-01-01T00:00:00Z' WHERE key=?", (key,)
|
|
)
|
|
assert next(iter_rule_refs(store)).fingerprint != f1
|
|
|
|
def test_tag_filter_selects_only_tagged_rules(self, store):
|
|
tagged = store.create(Rule(title="Tagged", document_number="2019-2"))
|
|
store.add_tag(tagged, "project:pfs")
|
|
store.create(Rule(title="Untagged", document_number="2019-3"))
|
|
assert [r.key for r in iter_rule_refs(store, tag="project:pfs")] == [tagged]
|
|
|
|
|
|
class TestCorpusRefs:
|
|
def test_excludes_comments_and_is_lazy(self, store):
|
|
key = store.create(
|
|
Item(
|
|
item_type="report", title="Report", url="https://x/r", abstract="Body."
|
|
)
|
|
)
|
|
refs = list(iter_corpus_refs(store))
|
|
assert [r.key for r in refs] == [key]
|
|
assert refs[0].collection == "corpus"
|
|
assert refs[0].load().metadata["kind"] == "corpus"
|
|
|
|
def test_zotero_not_consulted_until_load(self, store, tmp_path):
|
|
store.create(
|
|
Item(
|
|
item_type="report", title="Report", url="https://x/r", abstract="Body."
|
|
)
|
|
)
|
|
calls = []
|
|
|
|
class Z(ZoteroPdfIndex):
|
|
def pdfs_for(self, key):
|
|
calls.append(key)
|
|
return []
|
|
|
|
refs = list(iter_corpus_refs(store, zotero=Z({})))
|
|
assert calls == []
|
|
refs[0].load()
|
|
assert calls
|
|
|
|
def test_tag_filter_selects_only_tagged_items(self, store):
|
|
tagged = store.create(Item(item_type="report", title="T", abstract="Body."))
|
|
store.add_tag(tagged, "project:pfs")
|
|
store.create(Item(item_type="report", title="U", abstract="Body."))
|
|
assert [r.key for r in iter_corpus_refs(store, tag="project:pfs")] == [tagged]
|
|
|
|
|
|
class TestLazyZotero:
|
|
def test_no_copy_until_first_lookup(self, tmp_path):
|
|
src = tmp_path / "zotero.sqlite"
|
|
import sqlite3
|
|
|
|
con = sqlite3.connect(src)
|
|
con.executescript(
|
|
"CREATE TABLE items(itemID INTEGER, key TEXT); CREATE TABLE itemAttachments(itemID INTEGER, parentItemID INTEGER, path TEXT); CREATE TABLE deletedItems(itemID INTEGER);"
|
|
)
|
|
con.close()
|
|
snap_dir = tmp_path / "snap"
|
|
z = ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir)
|
|
assert not (snap_dir / "zotero.sqlite").exists()
|
|
assert z.pdfs_for("ABCD1234") == []
|
|
assert (snap_dir / "zotero.sqlite").exists()
|
|
m1 = (snap_dir / "zotero.sqlite").stat().st_mtime_ns
|
|
z2 = ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir)
|
|
z2.pdfs_for("ABCD1234")
|
|
assert (
|
|
snap_dir / "zotero.sqlite"
|
|
).stat().st_mtime_ns == m1 # source unchanged → no recopy
|
|
bump = src.stat().st_mtime_ns + 10**9
|
|
os.utime(src, ns=(bump, bump)) # source now newer than the snapshot
|
|
z3 = ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir)
|
|
z3.pdfs_for("ABCD1234")
|
|
assert (snap_dir / "zotero.sqlite").stat().st_mtime_ns != m1
|
|
|
|
def test_newer_wal_forces_a_recopy(self, tmp_path):
|
|
"""Zotero commits into zotero.sqlite-wal and only touches the main
|
|
file at a checkpoint — the WAL's mtime has to count too."""
|
|
src = tmp_path / "zotero.sqlite"
|
|
import sqlite3
|
|
|
|
con = sqlite3.connect(src)
|
|
con.executescript(
|
|
"CREATE TABLE items(itemID INTEGER, key TEXT); CREATE TABLE itemAttachments(itemID INTEGER, parentItemID INTEGER, path TEXT); CREATE TABLE deletedItems(itemID INTEGER);"
|
|
)
|
|
con.close()
|
|
snap_dir = tmp_path / "snap"
|
|
ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir).pdfs_for("ABCD1234")
|
|
snap = snap_dir / "zotero.sqlite"
|
|
m1 = snap.stat().st_mtime_ns
|
|
|
|
wal = tmp_path / "zotero.sqlite-wal"
|
|
wal.write_bytes(b"wal")
|
|
bump = m1 + 10**9
|
|
os.utime(wal, ns=(bump, bump)) # main file untouched, WAL is newer
|
|
|
|
ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir).pdfs_for("ABCD1234")
|
|
m2 = snap.stat().st_mtime_ns
|
|
assert m2 != m1 # re-copied
|
|
# and the snapshot now records what it captured, so the next run
|
|
# doesn't re-copy 1.95 GB for the same unchanged WAL
|
|
ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir).pdfs_for("ABCD1234")
|
|
assert snap.stat().st_mtime_ns == m2
|
|
|
|
def test_failed_copy_does_not_stamp_the_stale_snapshot(self, tmp_path, monkeypatch):
|
|
"""Stamping after a failed copy would pass a stale snapshot off as
|
|
current forever."""
|
|
import shutil
|
|
import sqlite3
|
|
|
|
src = tmp_path / "zotero.sqlite"
|
|
con = sqlite3.connect(src)
|
|
con.executescript(
|
|
"CREATE TABLE items(itemID INTEGER, key TEXT); CREATE TABLE itemAttachments(itemID INTEGER, parentItemID INTEGER, path TEXT); CREATE TABLE deletedItems(itemID INTEGER);"
|
|
)
|
|
con.close()
|
|
snap_dir = tmp_path / "snap"
|
|
ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir).pdfs_for("ABCD1234")
|
|
snap = snap_dir / "zotero.sqlite"
|
|
m1 = snap.stat().st_mtime_ns
|
|
|
|
bump = m1 + 10**9
|
|
os.utime(src, ns=(bump, bump)) # source now newer → a copy is due
|
|
monkeypatch.setattr(
|
|
shutil, "copy2", lambda *a, **k: (_ for _ in ()).throw(OSError("no space"))
|
|
)
|
|
assert (
|
|
ZoteroPdfIndex.lazy(src, tmp_path / "storage", snap_dir).pdfs_for(
|
|
"ABCD1234"
|
|
)
|
|
== []
|
|
)
|
|
assert snap.stat().st_mtime_ns == m1 # untouched, so the next run retries
|