BT411/scratchpad/test_mixed_version.py

"""Can a 554 client still play against tonight's build?  (2026-07-26)

The operator's testers are on BT411_4.11.554.  Tonight's build carries Cyd's
cockpit-refit merge, so before telling anyone "you don't need to update", prove
it: one pod from the OLD zip and one from the NEW build, on the same relay,
through a real launch.

Pass = both REGISTER, the relay launches, and NEITHER pod drops during the
round.  Fail = a mismatch that would strand half the players tonight.

Usage:  python scratchpad/test_mixed_version.py [path-to-old-554-root]
Spawns its own relay + pods; kills only that tree.
"""
import os
import re
import shutil
import subprocess
import sys
import time

REPO = r"C:\git\bt411"
CONTENT = os.path.join(REPO, "content")
SCRATCH = os.environ.get("TEMP", ".")
OLD_ROOT = (sys.argv[1] if len(sys.argv) > 1 else
            os.path.join(SCRATCH, "claude", "C--git-bt411",
                         "89ab96e9-6cf0-4555-939d-05bf3708745a", "scratchpad",
                         "old554", "BT411_4.11.554"))

PORT = 1500
SRC_EGG = "FOGDAY.EGG"
EGG = "_mixed_test.EGG"
RELAY_LOG = os.path.join(SCRATCH, "mixed_relay.log")

fails = []
procs = []


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


def relay_text():
    try:
        with open(RELAY_LOG, encoding="utf-8", errors="replace") as f:
            return f.read()
    except OSError:
        return ""


def wait_for(pattern, timeout):
    end = time.time() + timeout
    while time.time() < end:
        if re.search(pattern, relay_text()):
            return True
        time.sleep(0.25)
    return False


def kill_tree():
    ours = {p.pid for p in procs}
    try:
        out = subprocess.run(["wmic", "process", "get",
                              "ProcessId,ParentProcessId"],
                             capture_output=True, text=True).stdout
        pairs = []
        for line in out.splitlines():
            b = line.split()
            if len(b) == 2 and b[0].isdigit() and b[1].isdigit():
                pairs.append((int(b[0]), int(b[1])))
        for _ in range(4):
            for parent, pid in pairs:
                if parent in ours:
                    ours.add(pid)
    except Exception:
        pass
    for p in procs:
        try:
            if p.poll() is None:
                p.kill()
        except Exception:
            pass
    for pid in ours:
        subprocess.run(["taskkill", "/PID", str(pid), "/F", "/T"],
                       capture_output=True)


try:
    old_exe = os.path.join(OLD_ROOT, "build", "Release", "btl4.exe")
    new_exe = os.path.join(REPO, "build", "Release", "btl4.exe")
    old_content = os.path.join(OLD_ROOT, "content")
    for p in (old_exe, new_exe, old_content):
        if not os.path.exists(p):
            sys.exit("missing: %s" % p)

    if "btl4.exe" in subprocess.run(
            ["tasklist", "/FI", "IMAGENAME eq btl4.exe", "/NH"],
            capture_output=True, text=True).stdout:
        sys.exit("ABORT: btl4 already running")

    print("OLD pod: %s" % old_exe)
    print("NEW pod: %s" % new_exe)

    # the SAME egg for both sides (copy into each content dir)
    shutil.copyfile(os.path.join(CONTENT, SRC_EGG),
                    os.path.join(CONTENT, EGG))
    shutil.copyfile(os.path.join(CONTENT, SRC_EGG),
                    os.path.join(old_content, EGG))

    log = open(RELAY_LOG, "w", encoding="utf-8")
    procs.append(subprocess.Popen(
        [sys.executable, "-u", os.path.join(REPO, "tools", "btconsole.py"),
         "--relay", str(PORT), os.path.join(CONTENT, EGG),
         "--bind", "127.0.0.1", "--manual-launch"],
        cwd=CONTENT, stdout=log, stderr=subprocess.STDOUT))
    check("relay up", wait_for(r"console 127\.0\.0\.1:%d" % PORT, 20))

    base = {"BT_START_INSIDE": "1", "BT_DEV_GAUGES": "1", "BT_MATCHLOG": "0",
            "BT_RELAY": "127.0.0.1:%d" % PORT}

    # seat 1 = the OLD 554 pod, seat 2 = tonight's build
    for label, exe, cwd, netport in (
            ("554(old)", old_exe, old_content, 1502),
            ("584(new)", new_exe, CONTENT, 1602)):
        e = dict(os.environ)
        e.update(base)
        e["BT_LOG"] = "mixed_%s.log" % netport
        e["BT_SELF"] = "127.0.0.1:%d" % netport
        procs.append(subprocess.Popen(
            [exe, "-egg", EGG, "-net", str(netport)], cwd=cwd, env=e))
        print("  started %s on :%d" % (label, netport))
        time.sleep(3)

    check("BOTH pods staged (old + new on one relay)",
          wait_for(r"WAITING FOR OPERATOR", 120))
    regs = len(re.findall(r"REGISTERED", relay_text()))
    check("both REGISTERED", regs >= 2, "registered=%d" % regs)

    # launch a real round
    import socket
    with open(os.path.join(CONTENT, "operator_secret.txt")) as f:
        secret = f.read().strip()
    s = socket.create_connection(("127.0.0.1", PORT + 7), timeout=10)
    s.sendall(("AUTH %s\n" % secret).encode())
    time.sleep(0.6)
    s.sendall(b"launch\n")
    time.sleep(1)
    s.close()

    check("relay launched the mixed round",
          wait_for(r"RunMission #2 sent to 2 pod", 60))

    # both must SURVIVE the round -- a wire mismatch shows up as a drop
    print("  holding the round for 30s to watch for drops...")
    time.sleep(30)
    # Match REAL pod drops only.  A bare "dropped" also matches the stats
    # line's own counter -- "(dropped 0, tcp-fallback 0)" -- and the operator's
    # control socket closing normally, which cost one false failure here.
    txt = relay_text()
    drops = re.findall(r"game\[[^\]]*\] dropped|PLAYER \d+ LEFT", txt)
    check("neither pod dropped during the round", len(drops) == 0,
          "drops=%r" % (drops[:3],))
    last = txt.rstrip().splitlines()[-1] if txt.strip() else ""
    m = re.search(r"registered \[([^\]]*)\] udp-known \[([^\]]*)\]", txt[::-1][:0] + txt)
    reg = re.findall(r"registered \[([^\]]*)\] udp-known \[([^\]]*)\]", txt)
    check("both hosts still registered AND on UDP at the end",
          bool(reg) and reg[-1][0].count(",") == 1 and reg[-1][1].count(",") == 1,
          "last stats: registered [%s] udp-known [%s]" % (reg[-1] if reg else ("?", "?")))
    alive = sum(1 for p in procs[1:] if p.poll() is None)
    check("both pod processes still alive", alive == 2, "alive=%d/2" % alive)

finally:
    kill_tree()
    for junk in (os.path.join(CONTENT, EGG), os.path.join(old_content, EGG)):
        try:
            os.remove(junk)
        except OSError:
            pass

print()
print("WIRE-COMPATIBLE with 554" if not fails else "FAILURES: %s" % fails)
sys.exit(1 if fails else 0)