File size: 11,186 Bytes
557ab38 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 | """Tests for the evaluation harness.
We test each layer independently plus one end-to-end run with a mocked
extractor. No OpenAI calls are made — the runner accepts any callable that
returns (ExtractionResult, ExtractionMetrics), which is what makes this
testable offline.
Layers covered: comparators, flatten, doc-scoring, aggregation, runner, reports.
Version tag v2 (forces bytecode invalidation on mounted filesystems).
"""
from __future__ import annotations
from datetime import date
from src.data_prep.writer import read_jsonl
from src.eval.comparators import (
compare,
match_date,
match_exact,
match_money,
match_number,
match_text,
)
from src.eval.flatten import flatten_model
from src.eval.metrics import aggregate, micro_macro, score_doc
from src.eval.runner import run_eval
from src.schemas import ExtractionResult, Receipt
from src.utils.cost_tracker import ExtractionMetrics
# --- Comparators -----------------------------------------------------------
class TestComparators:
def test_text_fuzzy_match(self):
assert match_text("TAN WOON YANN", "Tan Woon Yann")[0]
assert match_text("TAN WOON YANN SDN BHD", "TAN WOON YANN")[0]
assert match_text("Starbucks", "McDonalds")[0] is False
def test_text_null_handling(self):
assert match_text(None, None) == (True, 1.0)
assert match_text("x", None)[0] is False
assert match_text(None, "x")[0] is False
def test_money_within_tolerance(self):
# Absolute tolerance: 0.01
assert match_money(72.00, 72.01)[0] is True
# Relative tolerance: 0.5% of 1000 = 5.00, so 4.00 delta passes
assert match_money(1000.00, 1004.00)[0] is True
def test_money_outside_tolerance(self):
# Small values where 0.5% rel is tiny and abs 0.01 is exceeded
assert match_money(1.00, 1.05)[0] is False
# 100 vs 102 -> abs 2 > 0.01, rel 2% > 0.5%
assert match_money(100.00, 102.00)[0] is False
def test_money_null(self):
assert match_money(None, None)[0] is True
assert match_money(0.0, None)[0] is False
def test_number_exact(self):
assert match_number(3, 3)[0] is True
assert match_number(3, 3.0000001)[0] is True
assert match_number(3, 4)[0] is False
def test_date_iso(self):
assert match_date(date(2018, 6, 25), "2018-06-25")[0] is True
assert match_date(date(2018, 6, 25), date(2018, 6, 25))[0] is True
assert match_date(date(2018, 6, 25), "2018-06-26")[0] is False
def test_exact_normalizes(self):
assert match_exact("USD", " usd ")[0] is True
assert match_exact("USD", "EUR")[0] is False
def test_dispatch(self):
assert compare("Hello world", "hello world", "text")[0]
assert compare(1.00, 1.005, "money")[0]
assert compare(date(2020, 1, 1), "2020-01-01", "date")[0]
assert compare("USD", "usd", "exact")[0]
assert compare(3, 3, "number")[0]
# --- Flattener -------------------------------------------------------------
class TestFlatten:
def test_receipt_flatten_basic(self):
r = Receipt(merchant="ACME", total=10.00, currency="USD")
flat = flatten_model(r, Receipt)
assert flat["merchant"] == ("ACME", "text")
assert flat["total"] == (10.00, "money")
assert flat["currency"] == ("USD", "exact")
def test_receipt_flatten_nested_address(self):
r = Receipt(
merchant="X",
total=1.0,
currency="USD",
merchant_address={"line1": "123 Main St", "city": "NYC"},
)
flat = flatten_model(r, Receipt)
assert flat["merchant_address.line1"][0] == "123 Main St"
assert flat["merchant_address.city"] == ("NYC", "text")
assert flat["merchant_address.postal_code"] == (None, "exact")
def test_flatten_line_items(self):
r = Receipt(
merchant="X",
total=5.0,
currency="USD",
line_items=[{"description": "coffee", "quantity": 1, "total": 5.0}],
)
flat = flatten_model(r, Receipt)
assert flat["line_items[]"] == (1, "number")
assert flat["line_items[0].description"] == ("coffee", "text")
assert flat["line_items[0].unit_price"] == (None, "money")
assert flat["line_items[0].total"] == (5.0, "money")
def test_flatten_dict_and_model_symmetric(self):
gt = {
"merchant": "ACME", "total": 10.00, "currency": "USD",
"merchant_address": None, "merchant_phone": None,
"transaction_date": None, "transaction_time": None,
"receipt_number": None, "line_items": [], "subtotal": None,
"tax": None, "tip": None, "payment_method": None,
}
pred = Receipt(merchant="ACME", total=10.00, currency="USD")
assert set(flatten_model(gt, Receipt)) == set(flatten_model(pred, Receipt))
# --- Doc-level scoring -----------------------------------------------------
class TestScoring:
def _pair(self, gt, pred):
return score_doc("doc_1", flatten_model(pred, Receipt), flatten_model(gt, Receipt))
def test_perfect_match(self):
r = Receipt(merchant="ACME", total=10.00, currency="USD")
stat, counts = self._pair(r, r)
assert stat.exact_match
for tp, fp, fn, _tn in counts.values():
assert fp == 0 and fn == 0
def test_wrong_merchant_is_mismatch(self):
gt = Receipt(merchant="ACME", total=10.0, currency="USD")
pred = Receipt(merchant="BETA STORE", total=10.0, currency="USD")
stat, counts = self._pair(gt, pred)
assert stat.exact_match is False
assert counts["merchant"] == (0, 1, 1, 0)
assert counts["total"] == (1, 0, 0, 0)
def test_missing_field_is_fn(self):
gt = Receipt(merchant="ACME", total=10.0, currency="USD", tax=1.0)
pred = Receipt(merchant="ACME", total=10.0, currency="USD")
_stat, counts = self._pair(gt, pred)
assert counts["tax"] == (0, 0, 1, 0)
def test_hallucinated_field_is_fp(self):
gt = Receipt(merchant="ACME", total=10.0, currency="USD")
pred = Receipt(merchant="ACME", total=10.0, currency="USD", tax=1.0)
_stat, counts = self._pair(gt, pred)
assert counts["tax"] == (0, 1, 0, 0)
# --- Aggregation -----------------------------------------------------------
class TestAggregation:
def test_micro_macro_perfect(self):
counts = [
{"merchant": (1, 0, 0, 0), "total": (1, 0, 0, 0)},
{"merchant": (1, 0, 0, 0), "total": (1, 0, 0, 0)},
]
stats = aggregate(counts, {"merchant": "text", "total": "money"})
summary = micro_macro(stats)
assert summary["micro_f1"] == 1.0
assert summary["macro_f1"] == 1.0
def test_micro_macro_partial(self):
counts = [
{"merchant": (1, 0, 0, 0), "total": (1, 0, 0, 0)},
{"merchant": (0, 1, 1, 0), "total": (1, 0, 0, 0)},
]
stats = aggregate(counts, {"merchant": "text", "total": "money"})
assert stats["merchant"].precision == 0.5
assert stats["merchant"].recall == 0.5
assert stats["merchant"].f1 == 0.5
assert stats["total"].f1 == 1.0
assert micro_macro(stats)["macro_f1"] == 0.75
def test_micro_macro_all_wrong(self):
counts = [{"merchant": (0, 1, 1, 0)}]
stats = aggregate(counts, {"merchant": "text"})
assert micro_macro(stats)["micro_f1"] == 0.0
# --- End-to-end runner -----------------------------------------------------
class TestRunner:
def _fake_extractor(self, mutator=None):
def _extract(record):
gt = record["ground_truth"]
data = Receipt.model_validate(gt)
if mutator is not None:
data = mutator(data)
return (
ExtractionResult(
document_type="receipt",
data=data,
field_confidences=[],
overall_confidence=1.0,
warnings=[],
),
ExtractionMetrics(
input_tokens=100, output_tokens=50, latency_ms=250.0, model="fake"
),
)
return _extract
def test_perfect_run_on_samples(self):
records = read_jsonl("data/samples/sroie_sample.jsonl")
report = run_eval(records, self._fake_extractor(), doc_type="receipt")
s = report.summary()
assert s["n_docs"] == len(records)
assert s["errors"] == 0
assert s["micro_f1"] == 1.0
assert s["macro_f1"] == 1.0
assert s["doc_exact_match"] == 1.0
def test_run_with_extractor_error(self):
def broken(_rec):
raise RuntimeError("api down")
records = read_jsonl("data/samples/sroie_sample.jsonl")[:2]
report = run_eval(records, broken, doc_type="receipt")
assert report.n_errors == len(records)
assert report.aggregate["micro_f1"] == 0.0
def test_run_with_wrong_merchant(self):
def wrong_merchant(r: Receipt) -> Receipt:
return r.model_copy(update={"merchant": "TOTALLY WRONG NAME"})
records = read_jsonl("data/samples/sroie_sample.jsonl")
report = run_eval(
records, self._fake_extractor(mutator=wrong_merchant), doc_type="receipt"
)
assert report.field_stats["merchant"].f1 == 0.0
assert 0.0 < report.aggregate["micro_f1"] < 1.0
assert report.doc_exact_match_rate == 0.0
# --- Reports ---------------------------------------------------------------
class TestReports:
def test_write_reports_creates_all_three(self, tmp_path):
from src.eval.report import write_reports
records = [
{
"id": "smoke",
"ground_truth": {
"merchant": "ACME", "total": 1.0, "currency": "USD",
"merchant_address": None, "merchant_phone": None,
"transaction_date": None, "transaction_time": None,
"receipt_number": None, "line_items": [], "subtotal": None,
"tax": None, "tip": None, "payment_method": None,
},
}
]
def extractor(record):
data = Receipt.model_validate(record["ground_truth"])
return (
ExtractionResult(
document_type="receipt",
data=data,
field_confidences=[],
overall_confidence=1.0,
warnings=[],
),
ExtractionMetrics(
input_tokens=1, output_tokens=1, latency_ms=1.0, model="fake"
),
)
report = run_eval(records, extractor, doc_type="receipt", model_label="fake")
paths = write_reports(report, tmp_path)
assert paths["csv"].exists()
assert paths["json"].exists()
assert paths["markdown"].exists()
md = paths["markdown"].read_text()
assert "Micro F1" in md
assert "1.0000" in md
|