← Commit history

Update 1 file(s)

Colby Knox ·10980b33c6 ·13d ago ·parent b42a805
1 file changed +37
tests/run_verb_tests.py+37
@@ -666,6 +666,42 @@ def section_of(name):     return "bridge-lifecycle"  +def other_batteries(target):+    """Other lanes' verb batteries (any run_verb_tests.py) driving the same machine right now: every+    lane runs in this distro, so /proc sees them. Ours and our parents are not counted."""+    mine = {os.getpid(), os.getppid()}+    found = []+    for d in os.listdir("/proc"):+        if not d.isdigit() or int(d) in mine:+            continue+        try:+            argv = open("/proc/%s/cmdline" % d, "rb").read().split(b"\0")+        except OSError:+            continue+        argv = [x.decode("utf-8", "replace") for x in argv if x]+        if not any(x.endswith("run_verb_tests.py") for x in argv):+            continue+        t = next((argv[i + 1] for i, x in enumerate(argv[:-1]) if x == "--target"), None) or next((x.split("=", 1)[1] for x in argv if x.startswith("--target=")), None)+        if t == target:+            found.append("%s (pid %s)" % (next(x for x in argv if x.endswith("run_verb_tests.py")), d))+    return found+++def wait_for_the_machine(target, limit=1800, step=20):+    """kicad-lane's rule for the shared laptop (2026-10-29): never overlap another lane's battery on+    the same machine, since apps raising windows steal each other's focus. Wait up to 30 min, then refuse."""+    t0 = time.time()+    while True:+        busy = other_batteries(target)+        if not busy:+            return+        if time.time() - t0 >= limit:+            print("REFUSED: another lane's battery is still driving %s after %d min: %s" % (target, limit // 60, "; ".join(busy)))+            sys.exit(3)+        print("waiting: another lane's battery is driving %s (%s)" % (target, "; ".join(busy)), flush=True)+        time.sleep(step)++ def main():     ap = argparse.ArgumentParser()     ap.add_argument("--target", required=True)@@ -675,6 +711,7 @@ def main():     tiers = {"daily": {"daily", "negative"}, "weekly": {"daily", "weekly", "negative"},              "all": {"daily", "weekly", "negative"}}[a.tier]     only = set(x.strip() for x in a.only.split(",") if x.strip())+    wait_for_the_machine(a.target)      print("warmup: waking the bridge (unscored)...")     call(a.target, "fusion_get_app_state", {}, timeout=45)