MAINT-21: tests read and patch the Controller through tools/main_files.py

Prepares the split of app/main.py's Controller into files by job, with no app
change. tools/main_files.py gives a test the text and the AST of main.py plus
the files split from it, and `live`, which sets a global in every file whose
functions read it (and raises when none does). Tests that read main.py's
source, and tests and replay tools that patch a main global (datetime,
compute, resolve, _mono, WRITE_BUDGET_S, ...), go through it, so they pass on
this layout and on the split one. test_split_patches.py fails when a test or
tool patches an app.main attribute that main.py no longer reads, or that
another file also reads. tools/maint21_same_ast.py proves every function,
method and module-level assignment of main.py at the base is found unchanged
in its new file.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
glenn schrooyen
2026-11-09 00:27:07 +01:00
co-authored by Claude
parent 56fc41fffa
commit 0d49027daa
50 changed files with 707 additions and 179 deletions
+8 -7
View File
@@ -38,6 +38,7 @@ logging.disable(logging.WARNING)
import opt11_identity as rig # noqa: E402
import safety79_identity as s79 # noqa: E402
import safety84_identity as s84 # noqa: E402
import main_files # noqa: E402
import app.main as app_main # noqa: E402
from app import clamp, messages # noqa: E402
from app.main import Controller # noqa: E402
@@ -202,7 +203,7 @@ check("⭐ N=2, A in its HOLD and B, which the loop drives, gives nothing: the "
check("...it starts no earlier than the saturation cycles allow (3 x 5 s)",
all(e.split(" ", 1)[0] >= "11:10:10" for c in (_c1, _c2)
for e in c.events if "OUTPUT_CLAMP" in e))
_quiet = patched(app_main, "delivery_short", lambda *a: False)(
_quiet = patched(main_files.live, "delivery_short", lambda *a: False)(
lambda: run(1, "drain", 1000.0, 20000, dur=1200.0, cut=gone(600, 1200)))
check("non-vacuous: without the delivery rule the N=1 drain that gives "
"nothing is silent (the loop's clamp never reaches its rail here)",
@@ -214,7 +215,7 @@ check("a dip of two cycles is not an alarm", warned(_dip[1]) == (0, 0, 0))
print("\n3. control coming back from maintenance does not jump")
END = {"maintenance_hold_min": 10}
_bounds = []
_real_compute = app_main.compute
_real_compute = main_files.live.compute
def _spy_compute(*a, **kw):
@@ -222,11 +223,11 @@ def _spy_compute(*a, **kw):
return _real_compute(*a, **kw)
app_main.compute = _spy_compute
main_files.live.compute = _spy_compute
try:
_h2 = run(2, "hold", 4500.0, 20000, dur=900.0, extra=END)
finally:
app_main.compute = _real_compute
main_files.live.compute = _real_compute
_h2_old = no_bound(lambda: run(2, "hold", 4500.0, 20000, dur=900.0,
extra=END))
_h1 = run(1, "hold", 4500.0, 20000, dur=900.0, extra=END)
@@ -380,8 +381,8 @@ def lag_runs():
_lag = lagging(lag_runs)
_lag_old = patched(app_main, "SHORT_CYCLES", 1)(
lambda: patched(app_main, "SHORT_SPAN_S", 0.0)(
_lag_old = patched(main_files.live, "SHORT_CYCLES", 1)(
lambda: patched(main_files.live, "SHORT_SPAN_S", 0.0)(
lambda: lagging(lag_runs)))
_lag_dead = lagging(lambda: run(1, "drain", 1000.0, 20000, dur=900.0,
cut=gone(600, 900)))
@@ -432,7 +433,7 @@ def both_ways(asked, given, margin):
_tp = taper(charge_runs)
_tp_old = patched(app_main, "delivery_short", both_ways)(
_tp_old = patched(main_files.live, "delivery_short", both_ways)(
lambda: taper(charge_runs))
print(f" a charge that tapers: {[warned(c) for _r, c in _tp]}, grading "
f"charges too {[warned(c) for _r, c in _tp_old]}")