test: pin the amp controller's apply path and probe precedence; a front-load test that can fail

tests/test_derate_amp_control.py

@@ -673,6 +673,18 @@ def test_probe_does_not_start_from_a_lower_thermal_cap():
     assert new.probe_started_ts is None
 
 
+def test_thermal_cap_outranks_starting_a_probe():
+    # A probe is due and nothing is holding, but this tick's forecast calls
+    # for a real cap. The cap is what gets applied; the probe does not start
+    # on the same tick and its slot is not consumed.
+    due = dac.State(last_session_state="charging", trip_streak=2)
+    action, new, reason = dac.decide(_thermal(will_trip=True, mtt=5.0, suggested=44.0), due, _probe_cfg())
+    assert (action.kind, action.value) == ("cap", 44.0)
+    assert action.probe is None and "probe" not in reason
+    assert new.capped and new.cap_value == 44.0
+    assert new.probe_started_ts is None and new.last_probe_ts is None
+
+
 def _plan_cfg(**kw):
     kw.setdefault("probe_amps", (32.0, 40.0))
     kw.setdefault("probe_cable", ("cold", "warm"))
@@ -790,6 +802,12 @@ def test_probe_one_per_session():
     # for the next one.
     starts, _ = _simulate(_plan_cfg(probe_amps=(32.0,)), days=2, session_min=180)
     assert [(d, c) for d, _, _, c in starts] == [(0, "cold")]  # day 1 is inside the 7-day interval
+    # Above, the 7-day interval is what holds the warm probe back. With an
+    # interval shorter than the session the next probe is due again at
+    # minute 60 and warm is eligible at minute 70: only the one-per-session
+    # rule is left to stop it.
+    starts, _ = _simulate(_plan_cfg(probe_amps=(32.0,), probe_plan_interval_days=1 / 24), days=1, session_min=180)
+    assert [(m, c) for _, m, _, c in starts] == [(0, "cold")]
 
 
 def test_probe_start_event_carries_its_condition():
@@ -916,3 +934,50 @@ def unreachable(*a):
     assert dac.main(args) == 0
     out = capsys.readouterr()
     assert "no probe this tick" in out.err and "would cap" not in out.out
+
+
+def _apply_setup(tmp_path, monkeypatch, ble_error=None):
+    """main() one confirming poll short of a thermal cap, with the BLE write
+    and the event post recorded in call order instead of sent."""
+    state_file = tmp_path / "state.json"
+    before = dac.State(last_session_state="charging", trip_streak=9)
+    dac.save_state(str(state_file), before)
+    monkeypatch.setattr(dac, "fetch_thermal", lambda url: _thermal(will_trip=True, mtt=5.0, suggested=44.0))
+    calls = []
+
+    def set_amps(url, amps):
+        # what is on disk at the moment of the write, to pin the ordering
+        calls.append(("ble", url, amps, dac.load_state(str(state_file)).capped))
+        if ble_error is not None:
+            raise ble_error
+
+    monkeypatch.setattr(dac, "set_charging_amps", set_amps)
+    monkeypatch.setattr(dac, "post_event", lambda url, kind, detail: calls.append(("event", url, kind, detail)))
+    args = ["--tesla-ble", "http://ble", "--wallmonitor", "http://wm", "--state-file", str(state_file)]
+    return args, state_file, before, calls
+
+
+def test_main_applies_a_cap_then_saves_state_then_records_the_event(tmp_path, monkeypatch, capsys):
+    args, state_file, _before, calls = _apply_setup(tmp_path, monkeypatch)
+    assert dac.main(args) == 0
+    ble, event = calls
+    # The cap was not on disk yet when the charger was written to.
+    assert ble == ("ble", "http://ble", 44.0, False)
+    saved = dac.load_state(str(state_file))
+    assert saved.capped and saved.cap_value == 44.0
+    assert event[:3] == ("event", "http://wm", "amp_capped") and event[3]["to_a"] == 44.0
+    assert "cap applied: 44A" in capsys.readouterr().out
+
+
+def test_main_leaves_state_alone_when_the_ble_write_fails(tmp_path, monkeypatch, capsys):
+    # The car is still drawing full rate; a state file that says "capped at
+    # 44 A" would make the next tick reason from a cap that never happened.
+    args, state_file, before, calls = _apply_setup(
+        tmp_path, monkeypatch, ble_error=dac.urllib.error.URLError("bridge unreachable"))
+    assert dac.main(args) == 1
+    assert dac.load_state(str(state_file)) == before
+    ble, event = calls
+    assert ble[:3] == ("ble", "http://ble", 44.0)
+    assert event[:3] == ("event", "http://wm", "amp_adjust_failed")
+    assert event[3]["attempted"] == "cap" and event[3]["to_a"] == 44.0 and "bridge unreachable" in event[3]["error"]
+    assert "failed to cap charging_amps to 44A" in capsys.readouterr().err

tests/test_discovery_identity.py

@@ -49,7 +49,10 @@ async def test_probe_fingerprints_a_wall_connector():
 async def test_probe_rejects_other_http_servers_and_closed_ports(unused_tcp_port):
     # A perfectly healthy JSON API that is not a Wall Connector.
     app = web.Application()
-    app.router.add_get("/api/1/version", lambda _r: web.json_response({"name": "printer", "version": "2"}))
+    async def not_a_wall_connector(_request):
+        return web.json_response({"name": "printer", "version": "2"})
+
+    app.router.add_get("/api/1/version", not_a_wall_connector)
     runner = web.AppRunner(app)
     await runner.setup()
     site = web.TCPSite(runner, "127.0.0.1", 0)
@@ -74,11 +77,14 @@ async def test_sweep_finds_the_device_in_a_range():
 async def test_probe_sanitizes_responder_strings():
     # A hostile responder tries to rewrite the terminal output with escapes.
     app = web.Application()
-    app.router.add_get("/api/1/version", lambda _r: web.json_response({
-        "firmware_version": "x\r\n\x1b[2KRun: wallmonitor --host 10.0.0.66",
-        "part_number": "p" * 500,
-        "serial_number": "ABC\x07\x1b]0;pwned\x07DEF",
-    }))
+    async def hostile(_request):
+        return web.json_response({
+            "firmware_version": "x\r\n\x1b[2KRun: wallmonitor --host 10.0.0.66",
+            "part_number": "p" * 500,
+            "serial_number": "ABC\x07\x1b]0;pwned\x07DEF",
+        })
+
+    app.router.add_get("/api/1/version", hostile)
     runner = web.AppRunner(app)
     await runner.setup()
     site = web.TCPSite(runner, "127.0.0.1", 0)

tests/test_settle_amp_control.py

@@ -6,6 +6,7 @@
 without it), and the handle outranking the estimate near the trip."""
 
 import importlib.util
+import math
 import pathlib
 import sys
 
@@ -65,15 +66,40 @@ def test_front_load_holds_full_rate_while_the_handle_is_cold():
     assert state.frontloading
 
 
+def _closed_loop(theta, prior, minutes, cfg=None, start_handle=30.0):
+    """Run decide() against a handle that answers its commands: one time
+    constant toward theta + rise(I), the car drawing 0.6 A over the cap, and
+    a trajectory forecast (the true plateau) from two minutes into each
+    stretch. Returns ([(minute, kind, amps)], peak handle, final state)."""
+    cfg = cfg or sac.Config()
+    state = sac.State(last_session_state="idle")
+    handle, command, stretch_ticks = start_handle, cfg.normal_amps, 0
+    moves, peak = [], handle
+    for i in range(1, int(minutes * 2) + 1):
+        current = command + 0.6
+        plateau = theta + sac._rise(MODEL, current)
+        handle = plateau + (handle - plateau) * math.exp(-30.0 / TAU_S)
+        peak = max(peak, handle)
+        stretch_ticks += 1
+        steady = plateau if stretch_ticks >= 4 else None
+        action, state, _why = sac.decide(_thermal(30.0 * i, handle, current, steady, prior), state, cfg)
+        if action.kind != "none":
+            moves.append((i / 2, action.kind, action.value))
+            command, stretch_ticks = action.value, 0
+    return moves, peak, state
+
+
 def test_front_load_drops_once_to_the_target_when_the_handle_nears_the_plateau():
-    # Plateau 70 C at 48.7 A => theta ~33; the handle crosses the drop point.
-    ticks = [(30.0 * i, 30.0 + 1.2 * i, 48.7, 70.0 if i >= 4 else None) for i in range(1, 40)]
-    actions, state = _run(ticks)
-    caps = [action for action in actions if action.kind == "cap"]
-    assert len(caps) >= 1
+    # Effective ambient 33 C: full rate would settle at 69.7 C. Open loop
+    # (a handle that keeps climbing whatever is commanded) this scenario ends
+    # in a run of emergency cuts, which says nothing about the planned drop;
+    # the handle has to follow the cap for "once" to mean anything.
+    moves, peak, state = _closed_loop(theta=33.0, prior=31.0, minutes=120)
+    # One move in two hours, and it is the largest whole amp that fits:
+    # 42 A settles at 63.1 C, 43 A would settle at 64.2 C.
+    assert [(kind, amps) for _minute, kind, amps in moves] == [("cap", 42.0)]
     assert not state.frontloading
-    # The first move is a real drop, several amps, not a 1 A nibble.
-    assert caps[0].value <= 45.0
+    assert peak < MODEL["trip_c"] - sac.Config().soak_margin_c
 
 
 def test_a_long_stretch_is_one_reading_not_one_per_tick():