b2f01ce010
#1 ingest: normalize the pinned caller id (_norm_cited_id) so a non-bare DOI caller cannot store a URL-form papers.id that defeats DQ-5 idempotency / the startswith(10.) recovery gates. #2 quality.section_label: collapse to a controlled bucket only for an EXACT canonical heading; descriptive titles ('Abstract Nonsense...') keep their real title instead of being mislabelled. #4 ra_grobid_backfill: release the read connection before the slow GROBID network loop, fresh connection for the write (no idle-in-transaction across the loop over the flaky tunnel). #5/#10 tex: flatten_inputs strips unresolved input/include at the depth cap (no literal leak on cycles); _norm_texkey strips only a single leading ./ . #6/#7 arxiv.fetch_source: keep non-.tex members resolvable for input; pick primary on an UN-commented documentclass line. #13 is_arxiv_id: also exclude http:// and arXiv-DOI forms. Tests added/updated for each. Left as deliberate decisions: #3 (pre-section text drop is pre-existing in extract_sections; abstract stored separately), #8 (script normalizer is intentionally self-contained, already documented), #9/#11/#12 (no current trigger / tightening the gzip heuristic would reject valid old LaTeX like documentstyle / title truncation is by design on a write-only column). ruff + mypy clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
303 lines
11 KiB
Python
303 lines
11 KiB
Python
"""Tests for codex.quality — chunk quality gate and section classification (F-16)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import MagicMock
|
||
|
||
from codex.quality import (
|
||
_bib_score,
|
||
_clean_title,
|
||
classify_section,
|
||
filter_chunks,
|
||
is_quality_chunk,
|
||
run_quality_pass,
|
||
section_label,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _settings(min_chars: int = 60, min_alpha: float = 0.40, max_bib: float = 0.70):
|
||
s = MagicMock()
|
||
s.chunk_min_chars = min_chars
|
||
s.chunk_min_alpha_ratio = min_alpha
|
||
s.chunk_max_bib_score = max_bib
|
||
return s
|
||
|
||
|
||
GOOD_TEXT = (
|
||
"In differential geometry, the discrete Laplace–Beltrami operator "
|
||
"generalises the smooth operator to polyhedral surfaces by using cotangent "
|
||
"weights. This discretisation preserves key spectral properties."
|
||
)
|
||
|
||
BIB_TEXT = (
|
||
"Bobenko, A. and Pinkall, U. (2015) Discrete differential geometry. "
|
||
"doi:10.1145/2010324.1964978 Smith et al. (2018) Surface remeshing. "
|
||
"doi:10.1007/s00211-019-01082-w Jones et al. (2020) doi:10.1145/1234567.890123"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _bib_score
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestBibScore:
|
||
def test_empty_string(self):
|
||
assert _bib_score("") == 0.0
|
||
|
||
def test_clean_prose_is_low(self):
|
||
assert _bib_score(GOOD_TEXT) < 0.10
|
||
|
||
def test_bib_like_text_is_high(self):
|
||
assert _bib_score(BIB_TEXT) > 0.70
|
||
|
||
def test_single_doi_short_text_spikes(self):
|
||
text = "See doi:10.1145/2010324.1964978 for details."
|
||
# 8 words, 1 DOI (weight 3) → raw = 3 / max(0.8, 1) = 3.0 → capped 1.0
|
||
assert _bib_score(text) == 1.0
|
||
|
||
def test_score_capped_at_one(self):
|
||
# _DOI_RE requires \d{4,} — use a valid registrant length
|
||
heavy = "10.1145/abcdef " * 20
|
||
assert _bib_score(heavy) == 1.0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# is_quality_chunk
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestIsQualityChunk:
|
||
def test_passes_good_text(self):
|
||
assert is_quality_chunk(GOOD_TEXT, settings=_settings()) is True
|
||
|
||
def test_fails_too_short(self):
|
||
assert is_quality_chunk("Short.", settings=_settings(min_chars=60)) is False
|
||
|
||
def test_fails_low_alpha_ratio(self):
|
||
# Mostly digits and symbols
|
||
junk = "1234 5678 9012 3456 7890 |||| ==== ---- ____ .... ;;;; 0000 9999 8888 7777"
|
||
assert is_quality_chunk(junk, settings=_settings(min_alpha=0.40)) is False
|
||
|
||
def test_fails_high_bib_score(self):
|
||
assert is_quality_chunk(BIB_TEXT, settings=_settings(max_bib=0.70)) is False
|
||
|
||
def test_bib_threshold_is_strict_gt(self):
|
||
# A text that scores exactly at the threshold should still pass (> not >=)
|
||
s = _settings(max_bib=1.0)
|
||
assert is_quality_chunk(GOOD_TEXT, settings=s) is True
|
||
|
||
def test_exactly_min_chars_passes(self):
|
||
# check is `len < min_chars`, so exactly at the threshold passes
|
||
text = "a" * 60
|
||
assert is_quality_chunk(text, settings=_settings(min_chars=60)) is True
|
||
|
||
def test_one_below_min_chars_fails(self):
|
||
text = "a" * 59
|
||
assert is_quality_chunk(text, settings=_settings(min_chars=60)) is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# classify_section
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestClassifySection:
|
||
def test_abstract_header(self):
|
||
assert classify_section("Abstract\n\nThis paper presents...") == "abstract"
|
||
|
||
def test_abstract_german(self):
|
||
assert classify_section("Zusammenfassung\nDiese Arbeit...") == "abstract"
|
||
|
||
def test_intro_header(self):
|
||
assert classify_section("Introduction\nOver the past decade...") == "intro"
|
||
|
||
def test_intro_numbered(self):
|
||
assert classify_section("1. Introduction\nWe study...") == "intro"
|
||
|
||
def test_theorem_header(self):
|
||
assert classify_section("Theorem 3.1 (Existence). Let M be...") == "theorem"
|
||
|
||
def test_lemma_header(self):
|
||
assert classify_section("Lemma 2. If f is smooth then...") == "theorem"
|
||
|
||
def test_proof_header(self):
|
||
assert classify_section("Proof. We proceed by induction...") == "proof"
|
||
|
||
def test_bibliography_header(self):
|
||
assert classify_section("References\n[1] Bobenko, A....") == "bibliography"
|
||
|
||
def test_bibliography_german(self):
|
||
assert classify_section("Literatur\n[1] Müller, H....") == "bibliography"
|
||
|
||
def test_body_fallback(self):
|
||
assert classify_section("The cotangent weight for edge (i,j) is...") == "body"
|
||
|
||
def test_doi_density_fallback_to_bibliography(self):
|
||
# No "References" header but heavy DOI density → bibliography
|
||
dense = BIB_TEXT
|
||
assert classify_section(dense) == "bibliography"
|
||
|
||
def test_only_first_200_chars_checked(self):
|
||
# Pattern in position >200 should NOT be matched as section header
|
||
prefix = "x" * 201
|
||
text = prefix + "\nAbstract"
|
||
assert classify_section(text) == "body"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# filter_chunks
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFilterChunks:
|
||
def test_empty_list(self):
|
||
assert filter_chunks([], settings=_settings()) == []
|
||
|
||
def test_all_good(self):
|
||
chunks = [GOOD_TEXT, GOOD_TEXT + " More content here."]
|
||
assert filter_chunks(chunks, settings=_settings()) == chunks
|
||
|
||
def test_all_bad(self):
|
||
chunks = ["x", "y"]
|
||
assert filter_chunks(chunks, settings=_settings()) == []
|
||
|
||
def test_mixed(self):
|
||
chunks = [GOOD_TEXT, "Short.", GOOD_TEXT]
|
||
result = filter_chunks(chunks, settings=_settings())
|
||
assert len(result) == 2
|
||
assert all(c == GOOD_TEXT for c in result)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# run_quality_pass
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_mock_conn(rows: list[dict]) -> MagicMock:
|
||
"""Return a mock psycopg connection whose execute().fetchall() returns rows."""
|
||
conn = MagicMock()
|
||
conn.execute.return_value.fetchall.return_value = rows
|
||
return conn
|
||
|
||
|
||
class TestRunQualityPass:
|
||
def test_removes_bad_chunk(self):
|
||
conn = _make_mock_conn([{"id": 1, "content": "x"}])
|
||
stats = run_quality_pass(conn=conn, settings=_settings())
|
||
assert stats["removed"] == 1
|
||
assert stats["kept"] == 0
|
||
# The DELETE must have been called with chunk id=1
|
||
delete_calls = [c for c in conn.execute.call_args_list if "DELETE" in str(c)]
|
||
assert len(delete_calls) == 1
|
||
|
||
def test_tags_good_chunk(self):
|
||
conn = _make_mock_conn([{"id": 2, "content": GOOD_TEXT}])
|
||
stats = run_quality_pass(conn=conn, settings=_settings())
|
||
assert stats["kept"] == 1
|
||
assert stats["tagged"] == 1
|
||
assert stats["removed"] == 0
|
||
# UPDATE must have been called
|
||
update_calls = [c for c in conn.execute.call_args_list if "UPDATE" in str(c)]
|
||
assert len(update_calls) == 1
|
||
|
||
def test_mixed_batch(self):
|
||
rows = [
|
||
{"id": 1, "content": "x"},
|
||
{"id": 2, "content": GOOD_TEXT},
|
||
{"id": 3, "content": "y"},
|
||
{"id": 4, "content": GOOD_TEXT + " extra words here."},
|
||
]
|
||
conn = _make_mock_conn(rows)
|
||
stats = run_quality_pass(conn=conn, settings=_settings())
|
||
assert stats["removed"] == 2
|
||
assert stats["kept"] == 2
|
||
assert stats["tagged"] == 2
|
||
|
||
def test_paper_id_scopes_query(self):
|
||
conn = _make_mock_conn([])
|
||
run_quality_pass(paper_id="2301.07041", conn=conn, settings=_settings())
|
||
first_call = conn.execute.call_args_list[0]
|
||
sql = first_call[0][0]
|
||
assert "paper_id" in sql
|
||
|
||
def test_all_papers_query_has_no_filter(self):
|
||
conn = _make_mock_conn([])
|
||
run_quality_pass(paper_id=None, conn=conn, settings=_settings())
|
||
first_call = conn.execute.call_args_list[0]
|
||
sql = first_call[0][0]
|
||
assert "paper_id" not in sql
|
||
|
||
def test_commit_is_called(self):
|
||
conn = _make_mock_conn([])
|
||
run_quality_pass(conn=conn, settings=_settings())
|
||
conn.commit.assert_called_once()
|
||
|
||
def test_section_label_stored_correctly(self):
|
||
conn = _make_mock_conn([{"id": 5, "content": "Abstract\n\n" + GOOD_TEXT}])
|
||
run_quality_pass(conn=conn, settings=_settings())
|
||
update_calls = [c for c in conn.execute.call_args_list if "UPDATE" in str(c)]
|
||
assert len(update_calls) == 1
|
||
kwargs = update_calls[0][0][1]
|
||
assert kwargs["section"] == "abstract"
|
||
assert kwargs["id"] == 5
|
||
|
||
|
||
class TestCleanTitle:
|
||
def test_strips_leading_numbering(self):
|
||
assert _clean_title("3.2 Variational Principles") == "variational principles"
|
||
|
||
def test_strips_latex_and_braces(self):
|
||
assert _clean_title(r"The \emph{Discrete} Laplacian") == "the discrete laplacian"
|
||
|
||
def test_collapses_and_lowercases(self):
|
||
assert _clean_title(" Rigidity Results ") == "rigidity results"
|
||
|
||
def test_strips_accent_macros(self):
|
||
assert _clean_title(r"M\"obius Invariance") == "mobius invariance"
|
||
|
||
def test_strips_ties_and_accents(self):
|
||
assert _clean_title(r"Colin~de~Verdi\`ere") == "colin de verdiere"
|
||
|
||
def test_truncates_at_word_boundary(self):
|
||
long = "alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu nu omega"
|
||
out = _clean_title(long)
|
||
assert len(out) <= 60
|
||
assert not out.endswith(" ")
|
||
assert long.lower().startswith(out) # whole words from the start, no partial tail
|
||
|
||
|
||
class TestSectionLabel:
|
||
def test_bare_canonical_heading_maps_to_bucket(self):
|
||
# Only an EXACT canonical heading collapses to its controlled bucket.
|
||
assert section_label("Introduction", "body text") == "intro"
|
||
assert section_label("Proof", "body text") == "proof"
|
||
assert section_label("References", "body text") == "bibliography"
|
||
assert section_label("Abstract", "body text") == "abstract"
|
||
|
||
def test_descriptive_title_stored_verbatim(self):
|
||
# R-F: a heading that merely STARTS with a keyword keeps its real title,
|
||
# not the bucket — fixes classify_section prefix-match over-bucketing.
|
||
assert section_label("Preliminaries", "body text") == "preliminaries"
|
||
assert section_label("3.1 Discrete Conformal Maps", "x") == "discrete conformal maps"
|
||
assert section_label("Proof of Theorem 3", "x") == "proof of theorem 3"
|
||
assert section_label("Abstract Nonsense and Categories", "x") == (
|
||
"abstract nonsense and categories"
|
||
)
|
||
assert section_label("Introduction to Operator Algebras", "x") == (
|
||
"introduction to operator algebras"
|
||
)
|
||
|
||
def test_none_or_blank_title_falls_back_to_content(self):
|
||
# .pdf/.txt path: unchanged content-based classification.
|
||
assert section_label(None, "Theorem 1. Let X be ...") == "theorem"
|
||
assert section_label("", "ordinary body prose here") == "body"
|
||
|
||
def test_title_cleaning_to_empty_falls_back_to_content(self):
|
||
# A pure-numbering heading cleans to "" → classify by content instead.
|
||
assert section_label("3.2", "Lemma 2. Suppose ...") == "theorem"
|