EvalHarness/tests/test_eval.py

"""Regression tests for the eval layer.

Anchored to OFFICIAL benchmark grading behavior:
- DROP: allennlp/simple-evals get_drop_metrics test cases
- math: PRM800K grader semantics (sympy equivalence)
- MCQ/gsm8k/BBH: extraction dispatch conventions

Run:  .venv/bin/python -m pytest tests/ -q   (or python tests/test_eval.py)
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent))

from evalharness.data.sample import Sample  # noqa: E402
from evalharness.eval import evaluate, get_eval, list_evals  # noqa: E402
from evalharness.eval.extractor import make_extractor  # noqa: E402
from evalharness.eval.scorer import _drop_metrics  # noqa: E402


def _mk(input_text='q', target='', **kw):
    return Sample(input=input_text, target=target, **kw)


def test_drop_official_cases():
    cases = [
        (['test'], ['test'], (1.0, 100.0)),
        (['test'], ['testing'], (0.0, 0.0)),  # token bags disjoint -> f1 0 (official)
        (['test test'], ['test'], (0.0, 100.0)),
        (['a'], ['b'], (0.0, 0.0)),
        (['1'], ['1.0'], (1.0, 100.0)),
        (['1'], ['2'], (0.0, 0.0)),
        (['ted', 'dan'], ['dan', 'ted'], (1.0, 100.0)),
        (['x'], ['y', 'z'], (0.0, 0.0)),
    ]
    for pred, gold, want in cases:
        got = _drop_metrics(pred, gold)
        assert got == want, f'{pred} vs {gold}: {got} != official {want}'


def test_math_grader():
    from evalharness.eval.math_grader import grade_answer

    assert grade_answer('0.5', '\\dfrac{1}{2}')
    assert not grade_answer('0.6', '\\dfrac{1}{2}')
    assert grade_answer('70000', '70,\\!000')
    assert not grade_answer('x=2', '2')
    assert grade_answer('1/2', '0.5') is True or grade_answer('1/2', '0.5') is False  # deterministic


def test_extractors():
    box = make_extractor('math_boxed')
    assert box('\\boxed{42} done', _mk())[0] == '42'
    assert box('no box here', _mk())[1] is False
    cascade = make_extractor(['math_boxed', 'answer_phrase', 'last_number'])
    assert cascade('The answer is 7.', _mk())[0] == '7'
    assert cascade('total 3 apples and 5 pears', _mk())[0] == '5'
    letter = make_extractor('mcq_letter')
    assert letter('So the answer is (B).', _mk())[0] == 'B'
    assert letter('答案是C', _mk())[0] == 'C'
    spans = make_extractor('answer_spans')
    got, ok, _ = spans('Answer: Chaz Schilens\nAnswer: JaMarcus Russell', _mk())
    assert ok and got == 'Chaz Schilens\nJaMarcus Russell'


def test_gsm8k_end_to_end():
    samples = [_mk(str(i), str(10 + i)) for i in range(4)]
    preds = ['\\boxed{10}', '#### 11', 'The answer is 12.', 'no idea']
    rep = evaluate(samples, preds, get_eval('gsm8k'))
    assert rep.metrics['acc'] == 0.75
    assert rep.num_failed_extractions == 1


def test_bbh_dispatch_by_target_format():
    mc = _mk('q?', '(B)', metadata={'subset': 'date_understanding'})
    ff = _mk('q?', 'True')
    ex = get_eval('bbh').resolve_extract()
    assert ex('So the answer is (B).', mc)[0] == 'B'
    assert ex('the answer is True', ff)[0] == 'True'


def test_mcq_end_to_end():
    samples = [_mk('q', 'B', choices=['a', 'b'])] * 2
    rep = evaluate(samples, ['(B)', '答案是B'], get_eval('mmlu'))
    assert rep.metrics['acc'] == 1.0


def test_all_recipes_resolve():
    from evalharness.eval.aggregator import get_aggregator
    from evalharness.eval.scorer import get_scorer

    for name in list_evals():
        recipe = get_eval(name)
        recipe.resolve_extract()
        scorers = recipe.resolve_scorers()
        assert scorers
        recipe.resolve_aggregators()
        for spec in recipe.aggregators.values():
            get_aggregator(spec[0] if isinstance(spec, tuple) else (spec or 'mean'))


def test_judge_default_label():
    from evalharness.eval.scorer import ScoreContext, get_scorer

    scorer = get_scorer('llm_judge')
    ctx = ScoreContext(judge=lambda msgs: 'zzz', params={  # no A/B/C anywhere
        'prompt_template': '{prediction}',
        'labels': {'A': {'is_correct': 1.0}, 'B': {'is_correct': 0.0}, 'C': {'is_correct': 0.0}},
        'default_label': 'C', 'primary': 'is_correct'})
    scores, details = scorer('x', 'y', _mk('q'), ctx)
    assert scores['is_correct'] == 0.0 and details['is_correct']['judge_label'] == 'C'


def test_aggregators():
    from evalharness.eval.aggregator import grouped_avg, mean, unbiased_pass_at_k
    from evalharness.eval.record import SampleResult

    rs = [SampleResult(raw_prediction='', scores={'acc': 1.0}, group_key='x'),
          SampleResult(raw_prediction='', scores={'acc': 0.0}, group_key='x'),
          SampleResult(raw_prediction='', scores={'acc': 1.0}, group_key='y')]
    assert mean(rs, 'acc') == 2 / 3
    assert grouped_avg(rs, 'acc') == {'x': 0.5, 'y': 1.0}
    assert unbiased_pass_at_k(10, 10, 5) == 1.0
    assert abs(unbiased_pass_at_k(10, 5, 5) - (1 - 1 / 252)) < 1e-9  # C(5,5)/C(10,5)=1/252


def test_viz_renders(tmp_path=None):
    from evalharness.viz import render

    samples = [_mk('q', '1')] * 3
    rep = evaluate(samples, ['\\boxed{1}'] * 3, get_eval('gsm8k'), model='m1')
    text = render(rep, style='text')
    assert 'acc' in text and '100.0%' in text
    md = render(rep, style='md')
    assert '| metric |' in md
    errs = render(rep, style='errors')
    assert 'Errors' in errs or errs == ''  # no errors -> empty-ish


if __name__ == '__main__':
    fails = 0
    for name, fn in sorted({k: v for k, v in globals().items()
                            if k.startswith('test_') and callable(v)}.items()):
        try:
            fn()
            print(f'PASS {name}')
        except AssertionError as e:
            fails += 1
            print(f'FAIL {name}: {e}')
        except Exception as e:
            fails += 1
            print(f'ERROR {name}: {type(e).__name__}: {e}')
    sys.exit(1 if fails else 0)