BT411/scratchpad/test_relay_rearm.py

"""Focused state-machine test for the launch re-arm fix (no sockets, no game).

Reproduces the operator's wedge in-process:
  a round runs (launches_sent=2), only SOME pods rejoin, LAUNCH is pressed.
Before the fix the press was swallowed with no output; after it, the relay
re-arms and serves the launch.
"""
import io
import os
import sys

REPO = r"C:\git\bt411"
sys.path.insert(0, os.path.join(REPO, "tools"))
os.chdir(os.path.join(REPO, "content"))

import btconsole  # noqa: E402

EGG = os.path.join(REPO, "content", "FOGDAY.EGG")

fails = []


def check(label, ok, extra=""):
    print("  %-52s %s %s" % (label, "OK" if ok else "*** FAIL ***", extra))
    if not ok:
        fails.append(label)


def new_relay():
    r = btconsole.Relay(1500, EGG, "127.0.0.1", 0, manual_launch=True)
    return r


class FakeSock:
    """Records what the relay would have put on the wire."""
    def __init__(self):
        self.sent = []
    def settimeout(self, _t):
        pass
    def setblocking(self, _b):
        pass
    def sendall(self, data):
        self.sent.append(data)


class FakeConn:
    """Stands in for a pod's console connection."""
    def __init__(self, acked=True):
        self.egg_sent = True
        self.acked = acked
        self.addr = ("10.0.0.9", 5000)
        self.last_seen = btconsole.time.time()
        self.sock = FakeSock()


print("=== 1. THE REPORTED WEDGE: a round ran, a partial rejoin, LAUNCH pressed ===")
r = new_relay()
print("  roster seats: %d" % len(r.roster))
# A mission ran to completion.
r.eggs_released = True
r.launches_sent = 2
r.mission_started_at = btconsole.time.time() - 600
r.stop_sent = True
# Only SOME of the last round's pods came back (fewer acked conns than seats).
r.console_conns = [FakeConn(acked=True)]
assert r._pods_ready() < len(r.roster), "test needs a PARTIAL rejoin"
# The operator presses LAUNCH.
r.launch_requested = True
r._tick_launch()
check("launches_sent reset from 2", r.launches_sent == 0, "-> %d" % r.launches_sent)
check("round re-opened (eggs_released cleared)", r.eggs_released is False)
check("roster restored to the template", len(r.roster) == len(r.orig_roster))
check("launch request still pending (not eaten)", r.launch_requested is True)
check("stale stop_sent cleared", r.stop_sent is False)

print("\n=== 2. A STALE 'End Mission' MUST NOT KILL THE NEXT MISSION ===")
r = new_relay()
r.launches_sent = 0          # between rounds: nothing is running
r.stop_requested = True      # operator pressed End Mission out of habit
r._tick_stop()
check("stop latch consumed while idle", r.stop_requested is False)
# and it must not have sent anything / marked a stop
check("no StopMission recorded", r.stop_sent is False)

print("\n=== 3. AN UN-ACKABLE SEAT: held, and the operator is told about RE-ARM ===")
r = new_relay()
r.eggs_released = True       # eggs are out ...
r.eggs_done_at = None        # ... but a seat will never ACK, so this stays None
r.launches_sent = 0
r.console_conns = [FakeConn(acked=True)]
r.launch_requested = True
r._tick_launch()
# Deliberately NOT auto-re-armed: during normal staging this state just means
# "pods are still ACKing", and re-releasing eggs there would restart the handout
# under an impatient operator.  The explicit rearm command covers the stuck case.
check("round NOT silently restarted during staging", r.eggs_released is True)
r._rearm_for_new_round("explicit operator re-arm")
check("explicit re-arm DOES re-open the round", r.eggs_released is False)

print("\n=== 3b. MID-PAIR MUST NEVER RE-ARM (the storm caught on the rig) ===")
r = new_relay()
r.console_conns = [FakeConn(acked=True) for _ in r.roster]
r.eggs_released = True
r.eggs_done_at = btconsole.time.time() - 1000
r.launches_sent = 1                                  # RunMission #1 has gone out
r.launch_at = btconsole.time.time() + 5              # #2 is timed and pending
r.launch_requested = True                            # stays set across the pair
r._tick_launch()
check("no re-arm between RunMission #1 and #2", r.launches_sent == 1,
      "-> %d" % r.launches_sent)
check("eggs NOT re-released mid-pair", r.eggs_released is True)
check("pair timer still intact", r.launch_at is not None)

print("\n=== 4. THE HAPPY PATH IS UNCHANGED (full roster, first launch) ===")
r = new_relay()
r.console_conns = [FakeConn(acked=True) for _ in r.roster]
r.eggs_released = True
r.eggs_done_at = btconsole.time.time() - 1000   # every pod ACKed long ago
r.launches_sent = 0
r.launch_requested = True
before_roster = len(r.roster)
r._tick_launch()
check("launch ARMED or fired", r.launch_at is not None or r.launches_sent > 0)
check("roster untouched", len(r.roster) == before_roster)
check("did NOT spuriously re-arm (eggs still released)", r.eggs_released is True)
check("RunMission actually reached the pods",
      all(c.sock.sent for c in r.console_conns),
      "%d pkt(s) to pod 1" % len(r.console_conns[0].sock.sent))

print("\n=== 5. REAPER: a silent peer stops pinning the round open ===")
r = new_relay()
r.launches_sent = 0
dead = FakeConn(acked=True)
dead.last_seen = btconsole.time.time() - (btconsole.DEAD_PEER_SECONDS + 30)
live = FakeConn(acked=True)
r.console_conns = [dead, live]
r.game_conns = []
dropped = []
r._drop_console = lambda c: (dropped.append(c), r.console_conns.remove(c))
r._reap_dead_peers()
check("the silent conn was reaped", dead in dropped)
check("the live conn was kept", live in r.console_conns and live not in dropped)

print("\n=== 6. REAPER IS DISARMED WHILE A MISSION RUNS ===")
r = new_relay()
r.launches_sent = 2                      # mission running
quiet = FakeConn(acked=True)
quiet.last_seen = btconsole.time.time() - 9999
r.console_conns = [quiet]
r.game_conns = []
r._drop_console = lambda c: fails.append("reaped mid-mission!")
r._reap_dead_peers()
check("no reaping during a live mission", "reaped mid-mission!" not in fails)

print("\n%s" % ("ALL CHECKS PASSED" if not fails else "FAILURES: %s" % fails))
sys.exit(1 if fails else 0)