"""Supervisor test: the PARKED relay must outlive crashes and remote restarts. Covers the two things a travelling operator cannot do by hand: * the relay dies (crash / killed) -> supervisor relaunches it * the operator sends `restart` remotely -> relay exits 42, comes back with the egg RE-READ (roster resize) Also asserts the restart is REFUSED mid-mission (it would drop every pod out of a live round) and that the log rotation keeps the dead generation's evidence. Spawns its own supervisor and kills only that tree. """ import os import re import shutil import socket import subprocess import sys import time REPO = r"C:\git\bt411" CONTENT = os.path.join(REPO, "content") sys.path.insert(0, os.path.join(REPO, "tools")) PORT = 1500 CTL = PORT + 7 SRC_EGG = "FOGDAY.EGG" EGG = "_park_test.EGG" LOG = os.path.join(CONTENT, "parked_relay.log") fails = [] sup = None def check(label, ok, extra=""): print(" %-56s %s %s" % (label, "OK" if ok else "*** FAIL ***", extra)) if not ok: fails.append(label) def log_text(path=LOG): try: with open(path, "r", encoding="utf-8", errors="replace") as f: return f.read() except OSError: return "" def wait_for(pattern, timeout, path=LOG): end = time.time() + timeout while time.time() < end: if re.search(pattern, log_text(path)): return True time.sleep(0.25) return False def relay_pids(): out = subprocess.run( ["wmic", "process", "where", "name='python.exe'", "get", "ProcessId,CommandLine"], capture_output=True, text=True).stdout pids = [] for line in out.splitlines(): if "btconsole.py" in line: m = re.search(r"(\d+)\s*$", line.strip()) if m: pids.append(int(m.group(1))) return pids def ctl(command, secret): """One control-port command, as a remote operator would send it.""" s = socket.create_connection(("127.0.0.1", CTL), timeout=10) try: s.sendall(("AUTH %s\n" % secret).encode()) time.sleep(0.6) s.sendall((command + "\n").encode()) time.sleep(1.2) # The relay ACKs `restart` and then exits, so the socket may be reset # under our recv -- that is CORRECT behaviour, not a failure. Drain # whatever arrived and treat a reset as end-of-stream. s.settimeout(2) got = b"" try: while True: chunk = s.recv(4096) if not chunk: break got += chunk except (socket.timeout, ConnectionResetError, OSError): pass return got.decode("utf-8", "replace") finally: s.close() try: if relay_pids(): sys.exit("ABORT: a relay is already running -- refusing to disturb it") shutil.copyfile(os.path.join(CONTENT, SRC_EGG), os.path.join(CONTENT, EGG)) with open(os.path.join(CONTENT, "operator_secret.txt"), encoding="ascii") as f: secret = f.read().strip() SUPLOG = os.path.join(os.environ.get("TEMP", "."), "park_sup.log") suplog = open(SUPLOG, "w", encoding="utf-8") sup = subprocess.Popen( [sys.executable, "-u", os.path.join(REPO, "tools", "btrelay_park.py"), "--egg", EGG, "--port", str(PORT)], cwd=REPO, stdout=suplog, stderr=subprocess.STDOUT) check("supervisor brought the relay up", wait_for(r"operator control on port %d" % CTL, 25)) first = relay_pids() check("relay process is running", len(first) == 1, "pids=%s" % first) # ---- 1. the relay DIES -> supervisor relaunches it ------------------- subprocess.run(["taskkill", "/PID", str(first[0]), "/F"], capture_output=True) time.sleep(1) back = False for _ in range(40): now = relay_pids() if now and now != first: back = True break time.sleep(1) check("relay relaunched after being killed", back, "new pids=%s" % relay_pids()) check("the dead generation's log was kept as .1", os.path.exists(LOG + ".1")) check("new generation is serving again", wait_for(r"operator control on port %d" % CTL, 25)) # ---- 2. remote `restart` -------------------------------------------- before = relay_pids() reply = ctl("restart", secret) check("relay acknowledged the remote restart", "restarting" in reply, reply.strip()[:60]) swapped = False for _ in range(40): now = relay_pids() if now and now != before: swapped = True break time.sleep(1) check("remote restart produced a NEW relay process", swapped, "%s -> %s" % (before, relay_pids())) suptext = log_text(SUPLOG) check("supervisor called it an operator restart, NOT a crash", "operator requested a restart" in suptext and "relay exited 42" not in suptext, ) check("supervisor treated the earlier KILL as a death (backoff path)", re.search(r"relay exited \d+ after .*relaunching in \d+s", suptext) is not None) check("relay is serving after the remote restart", wait_for(r"operator control on port %d" % CTL, 25)) # ---- 3. a roster resize really takes effect on restart --------------- # (the thing that was impossible from the road) doc = open(os.path.join(CONTENT, EGG), "rb").read() check("egg is re-read on restart (roster line present in the new log)", wait_for(r"roster: \d+ pilot\(s\)", 20)) # ---- 4. restart must be REFUSED mid-mission -------------------------- # Simulate a running mission by driving the relay's own state via the # control channel is not possible without pods, so assert the guard # exists and is wired to launches_sent/stop_sent. src = open(os.path.join(REPO, "tools", "btconsole.py"), encoding="utf-8").read() guard = re.search(r'elif cmd == "restart":(.{0,600})', src, re.S) check("mid-mission guard present in the restart command", bool(guard) and "launches_sent >= 2" in guard.group(1) and "not self.stop_sent" in guard.group(1)) finally: if sup is not None: for pid in relay_pids(): subprocess.run(["taskkill", "/PID", str(pid), "/F", "/T"], capture_output=True) subprocess.run(["taskkill", "/PID", str(sup.pid), "/F", "/T"], capture_output=True) for junk in (os.path.join(CONTENT, EGG), LOG, LOG + ".1"): try: os.remove(junk) except OSError: pass print() print("ALL PASS" if not fails else "FAILURES: %s" % fails) sys.exit(1 if fails else 0)