backend/test_core.py

"""
Smallest check that fails if the counting / PCU / id-validation logic breaks.

    python backend/test_core.py

No framework on purpose — these are the branches that silently produce wrong
numbers rather than crashing, so they need a tripwire.
"""
import sys
from pathlib import Path

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

from geometry import _side, _point_to_segment_dist
from pcu import compute_pcu, get_pcu_factor
from speed import estimate_speeds
from config import IMGSZ


def test_side():
    a, b = [0, 0], [10, 0]           # horizontal line
    assert _side((5, 5), a, b) > 0, "above the line must be positive"
    assert _side((5, -5), a, b) < 0, "below the line must be negative"
    assert _side((5, 0), a, b) == 0, "on the line must be zero (skipped by engine)"


def test_point_to_segment():
    assert abs(_point_to_segment_dist(5, 3, 0, 0, 10, 0) - 3.0) < 1e-9
    # Past the end of the segment: clamps to the endpoint, not the infinite line.
    assert abs(_point_to_segment_dist(20, 0, 0, 0, 10, 0) - 10.0) < 1e-9
    # Degenerate segment (both counting-line points identical) must not divide by zero.
    assert abs(_point_to_segment_dist(4, 5, 1, 1, 1, 1) - 5.0) < 1e-9   # 3-4-5


def test_pcu():
    assert get_pcu_factor(4) == 3.0        # Bus
    assert get_pcu_factor(7) == 0.5        # Two-wheeler
    assert get_pcu_factor(999) == 1.0      # unknown class falls back to 1

    # 2 buses in + 4 two-wheelers out = 2*3.0 + 4*0.5 = 8.0
    out = compute_pcu({"4": 2}, {"7": 4})
    assert out["pcu_in"] == 6.0, out
    assert out["pcu_out"] == 2.0, out
    assert out["total_pcu"] == 8.0, out
    assert out["per_class"]["Bus"]["count"] == 2, out

    empty = compute_pcu({}, {})
    assert empty["total_pcu"] == 0.0 and empty["per_class"] == {}


def test_speeds():
    # One track moving 10px/frame, one barely moving, one too short to score.
    tracks = {
        1: [(0, 0, 0), (1, 10, 0), (2, 20, 0)],
        2: [(0, 0, 0), (1, 1, 0), (2, 2, 0)],
        3: [(0, 5, 5)],
    }
    res = estimate_speeds(tracks)
    assert 3 not in res["per_track"], "single-sample tracks have no speed"
    assert res["per_track"][1]["px_per_frame"] == 10.0
    assert res["per_track"][1]["category"] == "fast"
    assert res["per_track"][2]["category"] == "slow"
    assert sum(res["distribution"].values()) == 100.0

    blank = estimate_speeds({})
    assert blank["distribution"] == {"slow": 0, "normal": 0, "fast": 0}


def test_id_validation():
    # Imported lazily: server.py pulls in torch/ultralytics, which is slow and
    # unavailable outside the container. Re-check the regex contract instead.
    import re
    id_re = re.compile(r"^[a-f0-9]{8}$")
    assert id_re.match("a1b2c3d4")
    assert not id_re.match("../../etc"), "traversal must not validate"
    assert not id_re.match("A1B2C3D4"), "uuid4().hex is lowercase"
    assert not id_re.match("a1b2c3d"), "wrong length must not validate"


def test_imgsz_matches_engine():
    # engine.py hardcodes imgsz to match the compiled OpenVINO graph; if these
    # two ever drift, inference silently runs at the wrong input size.
    engine_src = (Path(__file__).parent / "engine.py").read_text(encoding="utf-8")
    assert f"imgsz={IMGSZ}," in engine_src, f"engine.py must call track(imgsz={IMGSZ})"


if __name__ == "__main__":
    for name, fn in sorted(globals().items()):
        if name.startswith("test_") and callable(fn):
            fn()
            print(f"  ok  {name}")
    print("all core checks passed")