user2595
9647897173
3-signal quality filter (length/alpha-ratio/bib-score) + rule-based section classifier + retroactive CLI pass. 35 new tests, 287 total green.
246 lines
8.8 KiB
Python
246 lines
8.8 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,
|
||
classify_section,
|
||
filter_chunks,
|
||
is_quality_chunk,
|
||
run_quality_pass,
|
||
)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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
|