tests/test_causality.py

# test_causality.py — the causality regression test (keep green at all times).
# Feeds shuffled frames to the driver and requires an AssertionError; also checks that
# a correctly-ordered stream runs and that the silent policy emits nothing.
# Dependency-free: run with `python test_causality.py` (exits non-zero on failure).
from humomni.core.streaming_driver import drive
from humomni.phase1.policies import SilentPolicy


def test_shuffled_raises():
    frames = [(0.5, "a"), (1.0, "b"), (2.0, "c"), (1.5, "d")]  # 1.5 after 2.0 -> violation
    try:
        drive(frames, SilentPolicy(), "qid.mp4", "q")
    except AssertionError:
        return
    raise SystemExit("FAIL: shuffled frames did NOT raise AssertionError (causality unguarded)")


def test_ascending_ok():
    frames = [(0.5, "a"), (1.0, "b"), (1.5, "c")]
    rec = drive(frames, SilentPolicy(), "qid.mp4", "q")
    assert rec["question_id"] == "qid.mp4", rec
    assert rec["model_response_list"] == [], rec  # silent policy never speaks


def test_equal_timestamp_raises():
    frames = [(0.5, "a"), (0.5, "b")]  # not strictly increasing
    try:
        drive(frames, SilentPolicy(), "qid.mp4", "q")
    except AssertionError:
        return
    raise SystemExit("FAIL: duplicate timestamp did NOT raise (assert must be strict >)")


if __name__ == "__main__":
    test_shuffled_raises()
    test_ascending_ok()
    test_equal_timestamp_raises()
    print("PASS: causality assert raises on shuffled/duplicate frames; ascending order OK.")