From f6adf5675d76d15522bcc92e2b431a7e31831adc Mon Sep 17 00:00:00 2001 From: Fernando Gonzalez Date: Tue, 13 Oct 2026 20:18:37 -0400 Subject: [PATCH] test: pin the amp controller's apply path and probe precedence; a front-load test that can fail A review of the suite found one test that passed while showing the opposite of its title, and three paths in the deployed controller that no test reached. test_front_load_drops_once_... asserted only "at least one cap". Its handle climbed 1.2 C a tick whatever was commanded, so after the planned 48 -> 41 A drop the emergency rule fired every tick: 13 caps down to 17 A. It now runs decide() against a handle that follows the cap and asserts exactly one move in two hours, to the largest whole amp that fits. derate_amp_control: - a thermal cap on the tick a probe is due is applied, and the probe does not start (only abandoning a running probe was covered); - main()'s apply path: the charger is written before the state file is saved, and a failed BLE write returns 1, leaves the state file alone and records amp_adjust_failed; - test_probe_one_per_session was held by the 7-day plan interval, never by the one-per-session rule. A second case with an interval shorter than the session leaves only that rule. Each of the four was checked by breaking the behavior it pins. The two discovery tests registered bare lambdas as aiohttp handlers, the source of the suite's 4 deprecation warnings; they are async functions now. Co-Authored-By: Claude Opus 5.5 --- tests/test_derate_amp_control.py | 65 ++++++++++++++++++++++++++++++++ tests/test_discovery_identity.py | 18 ++++++--- tests/test_settle_amp_control.py | 40 ++++++++++++++++---- 3 files changed, 110 insertions(+), 13 deletions(-) diff --git a/tests/test_derate_amp_control.py b/tests/test_derate_amp_control.py index d70b706..e1a5ef7 100644 --- a/tests/test_derate_amp_control.py +++ b/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 diff --git a/tests/test_discovery_identity.py b/tests/test_discovery_identity.py index b482a4c..cd1d08b 100644 --- a/tests/test_discovery_identity.py +++ b/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) diff --git a/tests/test_settle_amp_control.py b/tests/test_settle_amp_control.py index 3803d4e..02e0b76 100644 --- a/tests/test_settle_amp_control.py +++ b/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():