from __future__ import annotations import pytest from smokescreen.models import ProbeReading from smokescreen.router import ProbeRouter def reading(probe_id: str, tip_c: float = 60.0) -> ProbeReading: return ProbeReading(probe_id=probe_id, tip_c=tip_c, source="test") def test_pinned_probe_drives_its_jack(): router = ProbeRouter(jacks=[1, 2], pinned={2: "meater:aa"}) assert router.observe(reading("meater:aa")) == 2 def test_auto_assignment_fills_free_jacks_in_order(): router = ProbeRouter(jacks=[1, 2]) assert router.observe(reading("probe:a")) == 1 assert router.observe(reading("probe:b")) == 2 def test_auto_assignment_skips_pinned_jacks(): router = ProbeRouter(jacks=[1, 2], pinned={1: "meater:aa"}) assert router.observe(reading("probe:b")) == 2 def test_binding_is_sticky_across_a_dropout(): """A probe that reconnects must not land on a different jack mid-cook.""" router = ProbeRouter(jacks=[1, 2]) assert router.observe(reading("probe:a")) == 1 assert router.observe(reading("probe:b")) == 2 assert router.observe(reading("probe:a", 70.0)) == 1 def test_extra_probes_have_nowhere_to_go(): router = ProbeRouter(jacks=[1, 2]) router.observe(reading("probe:a")) router.observe(reading("probe:b")) assert router.observe(reading("probe:c")) is None assert [r.probe_id for r in router.unassigned()] == ["probe:c"] def test_reading_for_returns_latest_sample(): router = ProbeRouter(jacks=[1]) router.observe(reading("probe:a", 50.0)) router.observe(reading("probe:a", 75.0)) assert router.reading_for(1).tip_c == 75.0 def test_reading_for_unbound_jack_is_none(): assert ProbeRouter(jacks=[1, 2]).reading_for(2) is None def test_pin_to_unknown_jack_is_rejected(): with pytest.raises(ValueError, match="unknown jacks"): ProbeRouter(jacks=[1, 2], pinned={3: "probe:a"}) def test_probe_pinned_twice_is_rejected(): with pytest.raises(ValueError, match="more than one jack"): ProbeRouter(jacks=[1, 2], pinned={1: "probe:a", 2: "probe:a"})