Zum Inhalt

tests/test_judge_events.py

from touchstone import judge_events as je


def test_constructors_have_type_tags():
    assert je.judge_start_event(1.0, 7)["type"] == "judge_start"
    assert je.judge_start_event(1.0, 7)["total"] == 7
    v = je.verdict_event(2.0, 0, "m", "v", "p1", 0, "A", 4, False, False, "gut")
    assert v["type"] == "verdict" and v["model"] == "m" and v["score"] == 4
    assert v["red_flag"] is False and v["unscored"] is False and v["rationale"] == "gut"
    ms = je.master_event(3.0, "m", "v", 72.5, True, "", "hoch")
    assert ms["type"] == "master" and ms["pct"] == 72.5 and ms["rubric_level"] == "hoch"
    assert je.judge_done_event(4.0, 5, 4)["type"] == "judge_done"


def test_verdict_event_truncates_rationale():
    v = je.verdict_event(1.0, 0, "m", "v", "p", 0, "A", 3, False, False, "x" * 500)
    assert len(v["rationale"]) == 160


def test_dumps_roundtrips_through_parse_line():
    line = je.dumps(je.judge_start_event(1.0, 3))
    assert je.parse_line(line) == {"ts": 1.0, "type": "judge_start", "total": 3}


def test_parse_line_tolerates_garbage():
    assert je.parse_line("") is None
    assert je.parse_line("   ") is None
    assert je.parse_line('{"truncated": ') is None  # half-written line
    assert je.parse_line("[1,2,3]") is None  # not a dict
    assert je.parse_line('{"no":"type"}') is None  # missing type tag


def test_build_view_empty():
    v = je.build_view([])
    assert v.total == 0 and v.done == 0 and v.eta_s is None and v.finished is False
    assert v.mean_score is None and v.red_flags == 0
    assert v.histogram == {1: 0, 2: 0, 3: 0, 4: 0, 5: 0}


def test_build_view_histogram_mean_red_and_unscored():
    events = [
        je.judge_start_event(0.0, 4),
        je.verdict_event(0.1, 0, "m", "v", "p1", 0, "A", 4, False, False, "ok"),
        je.verdict_event(0.2, 1, "m", "v", "p2", 0, "A", 2, True, False, "unsicher"),
        je.verdict_event(0.3, 2, "m", "v", "p3", 0, "B", 0, False, True, "judge weg"),
    ]
    v = je.build_view(events)
    assert v.total == 4 and v.done == 3
    assert v.histogram == {1: 0, 2: 1, 3: 0, 4: 1, 5: 0}  # unscored excluded
    assert v.red_flags == 1
    assert v.mean_score == 3.0  # mean of [4, 2], unscored excluded
    assert v.finished is False


def test_build_view_dedups_on_resume_replay():
    # same cell judged twice (crash+resume) -> counts once, last wins
    events = [
        je.judge_start_event(0.0, 2),
        je.verdict_event(0.1, 0, "m", "v", "p1", 0, "A", 2, False, False, "first"),
        je.verdict_event(0.2, 0, "m", "v", "p1", 0, "A", 5, False, False, "second"),
    ]
    v = je.build_view(events)
    assert v.done == 1 and v.histogram[5] == 1 and v.histogram[2] == 0
    assert v.verdicts[0].rationale == "second"


def test_build_view_eta_ignores_replay_burst():
    # prior burst written ~instantly (tiny gaps) must not crush the ETA
    events = [
        je.judge_start_event(0.0, 5),
        je.verdict_event(100.00, 0, "m", "v", "p1", 0, "A", 3, False, False, "a"),
        je.verdict_event(100.00, 1, "m", "v", "p2", 0, "A", 3, False, False, "b"),  # burst
        je.verdict_event(110.0, 2, "m", "v", "p3", 0, "A", 3, False, False, "c"),  # 10s real gap
    ]
    v = je.build_view(events)
    assert v.done == 3
    assert v.eta_s is not None and abs(v.eta_s - 20.0) < 0.01  # 10s/gap x 2 remaining


def test_build_view_masters_and_finished():
    events = [
        je.judge_start_event(0.0, 1),
        je.verdict_event(0.1, 0, "m", "v", "p1", 0, "A", 3, False, False, "x"),
        je.master_event(0.2, "m", "v", 40.0, False, "Q6 <= 2", "ungenügend"),
        je.judge_done_event(0.3, 1, 1),
    ]
    v = je.build_view(events)
    assert v.finished is True and len(v.masters) == 1
    assert v.masters[0].pct == 40.0 and v.masters[0].safety_passed is False
    assert v.masters[0].rubric_level == "ungenügend"
    d = v.as_dict()
    assert d["histogram"] == {"1": 0, "2": 0, "3": 1, "4": 0, "5": 0}  # str keys for JSON
    assert d["masters"][0]["safety_reason"] == "Q6 <= 2"


def test_build_view_ignores_out_of_range_score():
    # defensive: score=0 with unscored=False (corrupt line) must not pollute histogram/mean/red
    events = [
        je.judge_start_event(0.0, 1),
        je.verdict_event(0.1, 0, "m", "v", "p1", 0, "A", 0, True, False, "bogus"),
    ]
    v = je.build_view(events)
    assert v.done == 1  # still a verdict row
    assert v.histogram == {1: 0, 2: 0, 3: 0, 4: 0, 5: 0}
    assert v.mean_score is None  # no in-range scored verdict
    assert v.red_flags == 0  # out-of-range verdict's red_flag not aggregated


def test_index_html_is_judge_dashboard():
    html = je.INDEX_HTML
    assert "EventSource" in html  # SSE client present
    assert "Score" in html and "Master" in html  # judge-specific panels
    assert "Throttle" not in html  # no load panel (J5)
    assert je.TAILS_RESOURCES is False