#!/usr/bin/env python3
"""Tests for the Python side: relayctl's WebSocket codec, and that its key table
has not drifted from the one in the controller page.

    python3 tests/test_relayctl.py

No third-party dependencies, so it runs anywhere the tools do.
"""
from __future__ import annotations

import inspect
import pathlib
import re
import struct
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "tools"))

import relayctl  # noqa: E402

failures = 0
checks = 0


def check(condition, label):
    global checks, failures
    checks += 1
    if not condition:
        failures += 1
        print(f"  FAIL  {label}")


def check_eq(actual, expected, label):
    global checks, failures
    checks += 1
    if actual != expected:
        failures += 1
        print(f"  FAIL  {label}: got {actual!r}, want {expected!r}")


print("handshake")
# RFC 6455 1.3's worked example -- the same vector the C++ suite asserts, so the
# two implementations are pinned to the same truth.
check_eq(relayctl.expect_accept("dGhlIHNhbXBsZSBub25jZQ=="),
         "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=", "Sec-WebSocket-Accept of the RFC example")
check(relayctl.expect_accept("differentKey==") != relayctl.expect_accept("dGhlIQ=="),
      "different keys give different accepts")

print("frame encoding")
frame = relayctl.encode_frame(relayctl.OP_TEXT, b"A1")
check_eq(len(frame), 2 + 4 + 2, "short masked frame length")
check_eq(frame[0], 0x81, "FIN + text opcode")
check_eq(frame[1] & 0x80, 0x80, "mask bit set (clients must mask)")
check_eq(frame[1] & 0x7F, 2, "payload length in the 7-bit field")
mask = frame[2:6]
check_eq(bytes(b ^ mask[i & 3] for i, b in enumerate(frame[6:])), b"A1",
         "payload round-trips through the mask")

long_payload = b"x" * 300
frame = relayctl.encode_frame(relayctl.OP_TEXT, long_payload)
check_eq(frame[1] & 0x7F, 126, "126 signals the 16-bit length")
check_eq(struct.unpack(">H", frame[2:4])[0], 300, "16-bit length value")
mask = frame[4:8]
check_eq(relayctl.mask_payload(frame[8:], mask), long_payload,
         "mask_payload is its own inverse")
check_eq(len(relayctl.encode_frame(relayctl.OP_TEXT, b"")), 6,
         "empty frame still carries a mask key")

print("frame decoding")
receiver = relayctl.Receiver()


def server_frame(opcode, payload):
    """The unmasked frames the board sends."""
    out = bytearray([0x80 | opcode, len(payload)])
    out += payload
    return bytes(out)


status = server_frame(relayctl.OP_TEXT, b'S{"armed":1,"usb":1}')
# Deliver it one byte at a time: the decoder must not invent a frame early.
emitted = []
for i in range(len(status)):
    emitted += list(receiver.feed(status[i:i + 1]))
check_eq(len(emitted), 1, "exactly one frame from byte-at-a-time delivery")
check_eq(emitted[0][0], relayctl.OP_TEXT, "decoded opcode")
check_eq(emitted[0][1], b'S{"armed":1,"usb":1}',
         "decoded payload, including the board's 'S' status prefix")

# Two frames in one read, plus a split across the boundary.
two = server_frame(relayctl.OP_TEXT, b"OK") + server_frame(relayctl.OP_PING, b"hi")
receiver = relayctl.Receiver()
got = list(receiver.feed(two))
check_eq(len(got), 2, "both frames decoded from one read")
check_eq(got[0][1], b"OK", "first frame payload")
check_eq(got[1][0], relayctl.OP_PING, "second frame opcode")

receiver = relayctl.Receiver()
got = list(receiver.feed(two[:5])) + list(receiver.feed(two[5:]))
check_eq(len(got), 2, "a frame split mid-payload is still decoded once")

# 16-bit length path, and the protocol violations the client refuses.
receiver = relayctl.Receiver()
big = bytearray([0x81, 126]) + struct.pack(">H", 130) + b"y" * 130
check_eq(list(receiver.feed(bytes(big)))[0][1], b"y" * 130, "16-bit length decoded")

receiver = relayctl.Receiver()
try:
    list(receiver.feed(bytes([0x81, 0x82]) + b"\x00" * 6))
    check(False, "a masked server frame must be rejected")
except relayctl.RelayError:
    check(True, "masked server frame rejected")

receiver = relayctl.Receiver()
try:
    list(receiver.feed(bytes([0x01, 0x01, b"a"[0]])))
    check(False, "a fragmented frame must be rejected")
except relayctl.RelayError:
    check(True, "fragmented server frame rejected")

print("key table")
check_eq(relayctl.resolve_key("Enter"), 0x28, "Enter is usage 0x28")
check_eq(relayctl.resolve_key("0x28"), 0x28, "numeric usage accepted")
check_eq(relayctl.resolve_key("Delete"), 0x4C, "Delete is Delete-Forward 0x4C")
for bad, why in [("0x00", "usage 0 is not a key"), ("0x1ff", "usage must fit a byte"),
                 ("NotAKey", "unknown name")]:
    try:
        relayctl.resolve_key(bad)
        check(False, why)
    except (relayctl.RelayError, ValueError):
        check(True, why)

# The page and this tool must agree, or `relayctl key Enter` and pressing Enter in
# the browser would be different keys. Compare the JS object in web_page.h.
page = (pathlib.Path(__file__).resolve().parent.parent / "src" / "web_page.h").read_text()
match = re.search(r"const CODE_TO_USAGE = \{(.*?)\};", page, re.S)
check(match is not None, "CODE_TO_USAGE found in src/web_page.h")
if match:
    page_map = {}
    for name, usage in re.findall(r"([A-Za-z][A-Za-z0-9]*)\s*:\s*(0x[0-9a-fA-F]+)",
                                  match.group(1)):
        page_map[name] = int(usage, 16)
    check(len(page_map) > 80, f"page table parsed ({len(page_map)} entries)")
    only_page = sorted(set(page_map) - set(relayctl.CODE_TO_USAGE))
    only_cli = sorted(set(relayctl.CODE_TO_USAGE) - set(page_map))
    check_eq(only_page, [], "keys the page knows but relayctl does not")
    check_eq(only_cli, [], "keys relayctl knows but the page does not")
    mismatched = sorted(k for k in set(page_map) & set(relayctl.CODE_TO_USAGE)
                        if page_map[k] != relayctl.CODE_TO_USAGE[k])
    check_eq(mismatched, [], "same name, different usage id")
    check_eq(page_map.get("Enter"), 0x28, "Enter agrees on 0x28")
    check_eq(page_map.get("KeyA"), 0x04, "KeyA agrees on 0x04")

print("pasted text")
root = pathlib.Path(__file__).resolve().parent.parent
relay_src = (root / "tools" / "relayctl.py").read_text()
proto_src = (root / "src" / "proto.h").read_text()
text_src = (root / "src" / "text.h").read_text()

check_eq(relayctl.normalize_paste("a\r\nb"), "a\nb", "CRLF becomes one Enter")
check_eq(relayctl.normalize_paste("a\rb"), "a\nb", "a lone CR becomes Enter")
check_eq(relayctl.normalize_paste("a\nb"), "a\nb", "LF is left alone")
check_eq(relayctl.text_frame(0, "hi"), "P=0:6869", "the frame relayctl sends")
check_eq(relayctl.text_frame(250, "hi"), "P=250:6869", "the pace travels in decimal")
check_eq(relayctl.text_frame(0, "a"), "P=0:61", "one character")

# What the board's character map can type, read straight out of src/text.h: the
# switch spells out the punctuation and the ranges spell out the alphanumerics.
firmware = set()
for literal in re.findall(r"case '(\\.|[^'\\])':", text_src):
    firmware.add(ord(literal.encode("latin-1").decode("unicode_escape")))
check("ch >= 'a' && ch <= 'z'" in text_src, "the lowercase range is still a range")
check("ch >= 'A' && ch <= 'Z'" in text_src, "the uppercase range is still a range")
check("ch >= '0' && ch <= '9'" in text_src, "the digit range is still a range")
firmware |= set(range(ord("a"), ord("z") + 1))
firmware |= set(range(ord("A"), ord("Z") + 1))
firmware |= set(range(ord("0"), ord("9") + 1))

expected = {9, 10, 13} | set(range(32, 127))
check_eq(sorted(firmware), sorted(expected),
         "the board types exactly tab, newline, CR and printable ASCII")

# ... and the tool's validator must agree with it on every ASCII character, once the
# line endings are normalised. A character the tool waves through that the board then
# refuses would fail the whole paste at the board instead of at the terminal.
for code in range(128):
    ch = chr(code)
    tool_says_ok = not relayctl.untypeable(relayctl.normalize_paste(ch))
    if tool_says_ok != (code in firmware):
        check(False, f"code {code:#04x}: tool={tool_says_ok} board={code in firmware}")
check(True, "tool and board agree on all 128 ASCII codes")

for bad in ["\u2019", "\u2014", "\u00e9", "\u00a0", "\x00", "\x07", "\x7f"]:
    check_eq(relayctl.untypeable("ok" + bad), [bad], f"refuses {bad!r}")
check_eq(relayctl.untypeable("line1\nline2\tcol\r"), ["\r"],
         "CR is flagged before normalisation, and never sent")

# The three numbers have to be the board's numbers, or a paste dies on E8 halfway
# through because the tool split the text somewhere the firmware will not accept.
config_src = (root / "src" / "config.h").read_text()


def config_value(name):
    # Decimal or hex: kMouseDeltaMax is 127 and kMouseButtonMask is 0x1F, and both are
    # numbers a client has to agree with the firmware about.
    found = re.search(rf"\b{name}\s*=\s*(0[xX][0-9a-fA-F]+|\d+)\s*;", config_src)
    return int(found.group(1), 0) if found else None


check_eq(relayctl.TEXT_FRAME_MAX, config_value("kTextFrameMax"),
         "relayctl TEXT_FRAME_MAX == kTextFrameMax")
check_eq(relayctl.MAX_PASTE_CHARS, config_value("kTextJobCap"),
         "relayctl MAX_PASTE_CHARS == kTextJobCap")
check_eq(relayctl.DEFAULT_TEXT_PACE_MS, config_value("kTextPeriodMsDefault"),
         "relayctl DEFAULT_TEXT_PACE_MS == kTextPeriodMsDefault")
page_src = (root / "src" / "web_page.h").read_text()
check_eq(int(re.search(r"const TEXT_FRAME_MAX = (\d+);", page_src).group(1)),
         config_value("kTextFrameMax"), "page TEXT_FRAME_MAX == kTextFrameMax")
check_eq(int(re.search(r"const TEXT_JOB_MAX = (\d+);", page_src).group(1)),
         config_value("kTextJobCap"), "page TEXT_JOB_MAX == kTextJobCap")
check(config_value("kTextFrameMax") * 2 + 7 <= config_value("kWsMessageCap"),
      "the largest text frame fits one WebSocket message")

# _drain answered the board's ping with two methods that do not exist, which only
# surfaced on a session longer than the 15 s ping interval. A paste is such a session.
check("_send_frame" not in relay_src, "_drain no longer calls the missing _send_frame")
check("_check_error" not in relay_src, "_drain no longer calls the missing _check_error")
drained = relayctl.Relay._drain
src_lines = inspect.getsource(drained)
check("OP_PONG" in src_lines, "_drain answers a ping")
check("raise RelayError" in src_lines, "_drain still surfaces an error frame")

print("mouse")
check_eq(relayctl.mouse_frame(0x01, 30, -4), "M=01:30:-4:0:0",
         "a mouse frame is the button byte then four movements")
check_eq(relayctl.mouse_frame(0), "M=00:0:0:0:0",
         "and all five fields are always written -- the firmware parses exactly five")
check_eq(relayctl.mouse_frame(0x1f, -127, 127, -127, 127), "M=1f:-127:127:-127:127",
         "the full signed range")

for bad, why in (((0x20, 0, 0, 0, 0), "a sixth button bit has nowhere to go"),
                 ((0, 128, 0, 0, 0), "128 is more than one report"),
                 ((0, -128, 0, 0, 0), "and so is -128"),
                 ((0, 0, 0, 128, 0), "the wheel is one byte too")):
    try:
        relayctl.mouse_frame(*bad)
        check(False, f"mouse_frame should refuse: {why}")
    except relayctl.RelayError:
        check(True, f"mouse_frame refuses: {why}")

# A movement bigger than one report is split here, exactly as the page splits it, and the
# split is exact -- clamping it would move the cursor half as far as was asked and say
# nothing about it.
for dx, dy in ((300, 0), (0, -260), (300, -260), (127, 127), (1, -1), (-500, 500)):
    parts = relayctl.split_moves(dx, dy)
    check_eq(sum(p[0] for p in parts), dx, f"the X split of {dx},{dy} adds up exactly")
    check_eq(sum(p[1] for p in parts), dy, f"and the Y of {dx},{dy}")
    check(all(abs(p[0]) <= relayctl.MOUSE_DELTA_MAX and abs(p[1]) <= relayctl.MOUSE_DELTA_MAX
              for p in parts), f"and no part of {dx},{dy} exceeds one report")

check_eq(relayctl.mouse_frames("click", [], "left"),
         [(0x01, 0, 0, 0, 0), (0, 0, 0, 0, 0)],
         "a click is the button restated down then restated up")
check_eq(relayctl.mouse_frames("release", []), [(0, 0, 0, 0, 0)],
         "release restates the whole button byte as nothing")
drag = relayctl.mouse_frames("drag", ["300", "0"], "right")
check_eq(drag[0], (0x02, 0, 0, 0, 0), "a drag presses first")
check(drag[-1] == (0, 0, 0, 0, 0), "and releases last")
check(all(f[0] == 0x02 for f in drag[1:-1]),
      "and every movement in between carries the button, so it is a drag and not a move")
check_eq(sum(f[1] for f in drag), 300, "and the drag travels the whole distance")
check_eq(relayctl.mouse_frames("wheel", ["-3"]), [(0, 0, 0, -3, 0)],
         "wheel notches go where the report's wheel byte is")
for bad_args, why in ((("move", ["1"], "left"), "move needs dx and dy"),
                      (("wheel", [], "left"), "wheel needs a count"),
                      (("move", ["1", "2"], "thumb"), "an unknown button is refused"),
                      (("wiggle", [], "left"), "an unknown action is refused")):
    try:
        relayctl.mouse_frames(*bad_args)
        check(False, f"mouse_frames should refuse: {why}")
    except (relayctl.RelayError, ValueError):
        check(True, f"mouse_frames refuses: {why}")

# Every frame this tool can build must be one the firmware parses. The shape is asserted
# rather than a reimplementation of the parser, because the parser itself is tested
# against the same frames in tests/test_proto.cpp.
mframe = re.compile(r"^M=[0-9a-f]{2}:-?\d{1,3}:-?\d{1,3}:-?\d{1,3}:-?\d{1,3}$")
built = [relayctl.mouse_frame(*f)
         for f in relayctl.mouse_frames("drag", ["500", "-500", "4", "-4"], "middle")]
check(all(mframe.match(f) for f in built), f"every built frame parses ({len(built)})")
check(all(len(f) <= config_value("kWsMessageCap") for f in built),
      "and every one fits in one WebSocket message")

# The numbers are the board's. If they drift, the board starts answering E6 to frames
# this tool believed were legal.
check_eq(relayctl.MOUSE_DELTA_MAX, config_value("kMouseDeltaMax"),
         "relayctl MOUSE_DELTA_MAX == kMouseDeltaMax")
check_eq(relayctl.MOUSE_BUTTON_MASK, config_value("kMouseButtonMask"),
         "relayctl MOUSE_BUTTON_MASK == kMouseButtonMask")
check_eq(int(re.search(r"const MOUSE_DELTA_MAX = (\d+);", page_src).group(1)),
         config_value("kMouseDeltaMax"), "page MOUSE_DELTA_MAX == kMouseDeltaMax")
check_eq(int(re.search(r"const MOUSE_BUTTON_MASK = (0x[0-9a-f]+);", page_src).group(1), 16),
         config_value("kMouseButtonMask"), "page MOUSE_BUTTON_MASK == kMouseButtonMask")
check_eq(int(re.search(r"const MOUSE_FRAME_MS = (\d+);", page_src).group(1)),
         relayctl.MOUSE_FRAME_MS, "the page and the CLI pace mouse frames identically")

# Button bits, agreed between the page and this tool even though the page keys them by
# DOM number and this keys them by name. Getting one wrong is a clean frame that is the
# wrong button -- the Alt+Tab failure again, one layer down.
check_eq({name: bits for name, bits in relayctl.MOUSE_BUTTONS.items()},
         {"left": 0x01, "right": 0x02, "middle": 0x04, "back": 0x08, "forward": 0x10},
         "the button bits are the report descriptor's order")
check("0: 0x01, 2: 0x02, 1: 0x04, 3: 0x08, 4: 0x10" in page_src,
      "the page maps DOM button numbers to those same bits (DOM 1 is middle, 2 is right)")

print("protocol strings")

check_eq("S=%02x:%s" % (0x0A, "04,28"), "S=0a:04,28", "state frame format")
check_eq("Y=%02x:%s" % (0x05, "4c"), "Y=05:4c", "chord frame format")
check_eq("T" + "hunter2", "Thunter2", "auth frame format")

# The single-character commands relayctl emits must be ones proto.h accepts, and
# proto.h's own documented list must still mention them: a change on either side
# has to be a deliberate edit in two places.
for command in ("A1", "A0", "R", "H", "K"):
    check(f'"{command}"' in relay_src, f'relayctl sends "{command}"')
for command in ("'A'", "'R'", "'H'", "'K'", "'T'", "'S'", "'Y'", "'C'", "'P'", "'M'"):
    check(command in proto_src, f"proto.h parses {command}")
check("case 'M':" in proto_src, "proto.h routes M= to the mouse parser")

# The two mouse error codes have to be in the README's list. A client that cannot look up
# what E11 means cannot tell the operator why the pointer is not moving, and this suite is
# the only place that notices when the code and the document stop agreeing.
readme_src = (root / "README.md").read_text()
for code, where in (("10 the mouse interface is off", "proto.h"),
                    ("11 the target asked for boot protocol", "proto.h")):
    check(code in proto_src, f"proto.h's error list documents {code}")
    check(code.split()[0] in readme_src,
          f"README's error list has code {code.split()[0]}")
check("mouse interface is off" in readme_src and "boot protocol" in readme_src,
      "README explains what both mouse refusals mean, not just their numbers")
check("`M=<btn>`" in readme_src or "M=<btn>" in readme_src or "`M=`" in readme_src,
      "README's wire protocol table lists the M= command")
check("if (c == 'Y' && n == 0 && out.mods == 0)" in proto_src,
      "proto.h still requires a chord to press something")

print("the fail-safe wiring")
# Source pins, and deliberately honest about being source pins: control_server.cpp and
# main.cpp need a socket and a board, so nothing here runs them. What they were caught
# doing is invisible to a compiler and to every behavioural test on the host, and each of
# these three is a sentence that a later edit can quietly undo.
main_src = (root / "src" / "main.cpp").read_text()
server_src = (root / "src" / "control_server.cpp").read_text()

# The backstop must not ask whether a controller is linked: a controller that has gone
# away is exactly the case where linked() is false, which made the silence timer
# uncallable in the one situation it exists for -- the board sat armed with a key held
# forever, and read-only /status traffic kept it that way.
check("if (armed && control.controllerSeen() &&" in main_src,
      "loop()'s silence disarm runs off controllerSeen(), not linked()")
check("if (armed && control.linked()" not in main_src,
      "nothing gates an armed-and-silent disarm on the link still being there")

# Accepting a TCP connection is not the operator being alive.
accept = server_src[server_src.index("void ControlServer::acceptNew"):
                    server_src.index("size_t ControlServer::readSome")]
# Code lines only -- the function carries a comment explaining why it does NOT stamp the
# clock, and a check that flagged the comment would be worse than no check at all.
accept_code = "\n".join(l for l in accept.splitlines()
                        if not l.strip().startswith(("//", "*", "/*")))
check("_lastInboundMs" not in accept_code,
      "acceptNew() does not refresh the controller-silence clock")

# And closing a read-only /status request must not refresh it either -- the first fix
# stamped it on every close, which held a dead controller's armed state open exactly as
# effectively as the original bug did.
close_fn = server_src[server_src.index("void ControlServer::closeLink"):
                      server_src.index("}", server_src.index("_onLinkChange(_ctx, false)"))]
check("if (wasLinked) _lastInboundMs = millis();" in close_fn,
      "closeLink() stamps the silence clock only when a controller link is what closed")
check("const bool wasLinked = _phase == Phase::WebSocket;" in close_fn,
      "closeLink() treats anything that reached the WebSocket phase as able to leave "
      "keys down, not only what is still marked authenticated")

print("the bridge handover (phase 03)")
# `X` moves the keyboard from one socket to the other, and the safety sentences about it
# are ordering facts in control_server.cpp that no host-runnable test can observe -- they
# need two real sockets and a board. Same trade as bug 10's pins above: these are source
# pins, deliberately, for the three sentences a later edit can undo silently.
server_hdr = (root / "src" / "control_server.h").read_text()
claim_src = (root / "src" / "claim.h").read_text()

hand_at = server_src.index("void ControlServer::handOverClaim")
hand = server_src[hand_at:server_src.index("\n}", hand_at)]
# Code lines only -- the function's comment is full of the words it forbids itself, and
# a check that flagged the comment would be worse than no check at all.
hand_code = "\n".join(l for l in hand.splitlines()
                      if not l.strip().startswith(("//", "*", "/*")))

# (a) The funnel runs BEFORE the newcomer is adopted. closeLink() is the release funnel
#     (it disarms and releases for anything that reached the WebSocket phase -- bug 10's
#     fix -- and runs _onLinkChange(false), which in main.cpp is disarm("controller
#     disconnected") -> releaseEverything()); adopting the claimant is `_authed = true`.
#     If those two ever swap order, the newcomer inherits a held key and an armed
#     bridge, and nothing on a host would ever notice.
check("closeLink(\"taken over\")" in hand_code,
      "the handover runs the existing release funnel, not a copy of it")
check("_authed = true" in hand_code,
      "and it is the code that adopts the claimant as the controller")
check(hand_code.index("closeLink(\"taken over\")") < hand_code.index("_authed = true"),
      "the funnel runs BEFORE the claimant is adopted -- a takeover can never inherit "
      "a held key or an armed bridge")
check("if (!linked) disarm(\"controller disconnected\");" in main_src,
      "and the funnel's link change is the one main.cpp disarms (releases every key "
      "and every mouse button) on")

# (b) No arm is reachable from the `X` handler. The board lands DISARMED by design; the
#     newcomer arms with an ordinary authenticated A1 or never. `arm(`/`setArmed`/"armed =
#     true" must not appear in the handover's code lines, nor in the controller-side `X`
#     case in main.cpp (which is a no-op answered OK).
for tok in ("setArmed", "armed = true", "arm("):
    check(tok not in hand_code,
          f"the handover path contains no {tok!r} -- a takeover cannot produce an "
          f"armed board")
mt_at = main_src.index("case proto::Cmd::Takeover:")
mt_case_raw = main_src[mt_at:main_src.index("return;", mt_at)]
# Comment lines stripped for the same reason as hand_code: the case's own comment
# explains what it must not do, in words that contain the words being searched for.
mt_case = "\n".join(l for l in mt_case_raw.splitlines()
                    if not l.strip().startswith(("//", "*", "/*")))
check("control.sendOk();" in mt_case,
      "`X` from the incumbent controller is a no-op answered OK, not E6")
check("setArmed" not in mt_case and "arm(" not in mt_case,
      "and it cannot arm either -- you already hold the bridge, that is all this says")

# (c) A claim cannot be promoted without `_claimAuthed`. Two halves: the whitelist is
#     consulted before the Takeover branch in handleClaimFrame, and claim.h admits `X`
#     only in the authenticated branch, so the ordering has something correct to check.
hcf_at = server_src.index("bool ControlServer::handleClaimFrame")
hcf = server_src[hcf_at:server_src.index("\n}", hcf_at)]
check("claim::honours(_viewerRole, _claimAuthed, cmd)" in hcf,
      "handleClaimFrame asks the one whitelist, with _claimAuthed in the question")
check(hcf.index("claim::honours(") < hcf.index("proto::Cmd::Takeover"),
      "and it asks BEFORE the Takeover branch -- `X` cannot reach the promotion "
      "unauthenticated")
check_eq(server_src.count("handOverClaim(nowMs);"), 1,
         "handOverClaim is called from exactly that one place, behind the whitelist")
hon_at = claim_src.index("inline bool honours(")
hon = claim_src[hon_at:claim_src.index("\n}", hon_at)]
check("if (!authed) return cmd == proto::Cmd::Auth;" in hon,
      "claim.h admits only T before authentication -- an unauthenticated `X` is refused")
check(hon.index("if (!authed)") < hon.index("case proto::Cmd::Takeover"),
      "and `X` lives in the authenticated branch only, after that refusal")

# E12 is its own sentence: "you were dropped", not E3's "you were refused". The named
# code and the incumbent-side send are both pinned so the two can never merge back.
check("kErrTakenOver = 12" in server_hdr,
      "the error code is named at 12, the number the incumbent's page will read")
check('sendError(kErrTakenOver, "session taken over")' in hand_code,
      "and it is that named code the incumbent is told with, before the funnel closes "
      "its link")

print("one routing path, two transports")
# Phase 02 task 04 moved the TLS session underneath the routing instead of beside it. The
# claims are structural, so they are pinned as source sentences: what they defend is a
# second path that can disagree, and nothing that runs on a host can observe two paths
# agreeing until one of them is wrong.
conn_src = (root / "src" / "conn.h").read_text()
ws_src = (root / "src" / "ws_codec.h").read_text()

# readSome()'s rule, carried over from the TLS task and now the only clock-stamping read:
# bytes count as "the operator is alive" only from an authenticated controller on the
# WebSocket. Over TLS a handshake, a /status poll and a viewer refresh all look identical
# to the transport, so if this condition ever loosens to "bytes arrived", anything on the
# network can hold a dead controller's armed state open -- bug 10's shape again.
read_fn = server_src[server_src.index("size_t ControlServer::readSome"):
                     server_src.index("\n}", server_src.index("size_t ControlServer::readSome"))]
check("if (_phase == Phase::WebSocket && _authed) _lastInboundMs = millis();" in read_fn,
      "readSome() stamps the silence clock only for an authenticated controller on the "
      "WebSocket, whatever transport the bytes came over")
check("_lastInboundMs" in read_fn, "readSome() is still the code that notices inbound bytes")
# ... and it is the ONLY inbound path that may. Every writer of that clock is a route
# that can keep the board armed, so the count is pinned: readSome, the authentication,
# closeLink stamping on the way out, and the bridge handover ADOPTING an already-
# authenticated claim (phase 03 -- the claim was checked against the same passphrase,
# so from that instant a controller exists and is speaking; the adoption stamps exactly
# the stamp an auth would). A fifth writer needs its sentence written here first.
stamps = [l.strip() for l in server_src.splitlines() if "_lastInboundMs = millis()" in l]
check_eq(len(stamps), 4,
         "exactly four writers of the controller-silence clock: auth, readSome, "
         "closeLink, handover adoption")
check(any("the controller exists from here" in l for l in stamps),
      "one of them is the authentication itself")
# The spare slot must not be able to reach that clock even by accident: it reads the Conn
# directly, never through readSome().
viewer_fn = server_src[server_src.index("void ControlServer::serveViewer"):
                       server_src.index("void ControlServer::stopViewer")]
# Code lines only: the function carries a comment explaining why it does NOT go through
# readSome(), and a check that flagged the comment would be worse than no check at all.
viewer_code = "\n".join(l for l in viewer_fn.splitlines()
                        if not l.strip().startswith(("//", "*", "/*")))
check("readSome" not in viewer_code,
      "the read-only slot reads its Conn directly, so it cannot refresh the controller clock")

# One HTTP response writer. The task's completion criterion is a grep that comes up empty,
# so this is that grep, as a check: exactly one formatter in src/, and the only literal
# response status line outside it is the 101 protocol switch (which is not a response).
writers = sorted(p.name for p in (root / "src").glob("*")
                 if '"HTTP/1.1 ' in p.read_text(errors="replace"))
check_eq(writers, ["conn.h", "control_server.cpp"],
         "only conn.h's formatter and the 101 upgrade line write an HTTP status line")
check_eq(server_src.count('"HTTP/1.1 '), 1,
         "control_server.cpp holds exactly one, the WebSocket 101 -- not a second response writer")
check("respondAndClose(rsp)" in server_src and "respondViewer(rsp)" in server_src,
      "both slots queue a built response through the same pair of writers")
# serveReadonly is the single read-only answer, reachable from both listeners/slots.
check_eq(server_src.count("serveReadonly(req, rsp,"), 2,
         "the page/status routing is reached from the controller slot and the spare, once each")
check("void ControlServer::serveReadonly" in server_src,
      "and it is defined once")
check_eq(server_src.count("plaintextRoute("), 1,
         "the :80 answer table is consulted in exactly one place, inside serveReadonly")
# Both slots go through the transport view; nothing in the routing consults the port.
check("_ctl.read(" in conn_src or "_ctl.read(" in server_src or "_plain->read" in conn_src,
      "the controller slot reads through kr::Conn")
check("_view.read(" in server_src, "the spare slot reads through kr::Conn too")
check("_server.hasClient()" in server_src and "::accept(_tlsListen" in server_src,
      "both listeners still feed one accept decision")

# The :80 policy itself is pure (kr::plaintextRoute, walked in tests/test_conn.cpp); these
# pins keep the firmware from re-deciding it locally.
check("PlainRoute::UpgradeRequired" in conn_src and "PlainRoute::Redirect" in conn_src,
      "conn.h spells out the redirect and the 426 as data, not as scattered ifs")
check("namespace kr" in conn_src, "conn.h is in namespace kr, like every other module")
check("token=" in conn_src and "The value is never copied" in conn_src,
      "the token-in-URL test looks for the key without copying the passphrase")
# The Host header had to be parsed to build a Location that keeps the operator's address.
check('ciEquals(ls, "host:", 5)' in ws_src, "ws_codec parses Host: for the redirect")
check("char host[" in ws_src, "and carries it on Request")

# What the terminal client says when the board wants TLS. A refusal that reads like a
# passphrase problem sends the operator re-typing a correct secret until the board locks it.
check("board refuses plaintext" in relayctl.handshake_refusal("HTTP/1.1 426 Upgrade Required"),
      "426 is named as a TLS refusal")
check("passphrase" not in relayctl.handshake_refusal("HTTP/1.1 426 Upgrade Required").lower(),
      "and the 426 wording does not blame the passphrase")
check("passphrase" in relayctl.handshake_refusal("HTTP/1.1 401 Unauthorized"),
      "401 still says what it always said")
check(relayctl.handshake_refusal("HTTP/1.1 503 Service Unavailable")
      != relayctl.handshake_refusal("HTTP/1.1 426 Upgrade Required"),
      "'busy' and 'this board wants TLS' are never the same sentence")
check("https" in relayctl.handshake_refusal("HTTP/1.1 301 Moved Permanently"),
      "a redirect on :80 is reported as the port policy it is")
for line, want in (("HTTP/1.1 418 Teapot", "refused"), ("garbage", "refused")):
    check_eq(relayctl.handshake_refusal(line), want, f"unknown status {line!r}")

# The page the transport sends: the byte count tests/test_conn.cpp drains through the
# chunk arithmetic has to be the page that is actually in the tree. The page is a
# chain of raw-string chunks joined by one #if KR_TLS build marker (task 05), so
# "the page" is the chunks plus exactly one marker branch -- and both branches are
# asserted to be the same length, which is what keeps every pinned count (this one,
# the gate's, the device's `wc -c`) build-independent. That is a test, not a hope.
def served_page(src, tls=True):
    """The bytes one build serves: the raw-string chunks concatenated around the
    marker branch that build takes. None when the layout is not chunks joined by
    markers, so a restructured page fails loudly instead of counting half of it."""
    chunks = re.findall(r'R"HTML\((.*?)\)HTML"', src, re.S)
    marks = re.findall(r'#if KR_TLS\n\s*"([^"]*)"\n#else\n\s*"([^"]*)"\n#endif', src)
    if len(chunks) < 1 or len(marks) != len(chunks) - 1:
        return None
    out = ""
    for i, chunk in enumerate(chunks):
        out += chunk
        if i < len(marks):
            out += marks[i][0] if tls else marks[i][1]
    return out


page_tls = served_page(page_src, True)
page_plain = served_page(page_src, False)
check(page_tls is not None and page_plain is not None,
      "every join between the page's raw-string chunks is one #if KR_TLS marker")
check(all(len(a) == len(b) for a, b in
          re.findall(r'#if KR_TLS\n\s*"([^"]*)"\n#else\n\s*"([^"]*)"\n#endif',
                     page_src)),
      "both branches of every build marker are the same length, so the served page "
      "is the same size whichever build flashes")
check(page_tls is not None and "var KR_TLS_BUILD=1;" in page_tls,
      "the TLS build's page carries its marker")
check("var KR_TLS_BUILD=0;" in page_plain,
      "and the -D KR_TLS=0 rollback page carries its own")
check_eq(len(page_tls.encode("utf-8")), len(page_plain.encode("utf-8")),
         "the two builds serve byte-count-equal pages")
pinned_page_bytes = int(re.search(r"kServedPageBytes = (\d+)",
                                  (root / "tests" / "test_conn.cpp").read_text()).group(1))
check_eq(len(page_tls.encode("utf-8")), pinned_page_bytes,
         "the page the chunk arithmetic is tested against is the page in web_page.h")

# The page's only scheme choice is its own origin; a TLS board never has an http: origin
# because :80 answers a redirect (pinned above, walked in test_conn).
check(page_src.count('"wss://"') == 1 and page_src.count('"ws://"') == 1,
      "the page derives the socket scheme from location.protocol, in exactly one place")
check('location.protocol === "https:" ? "wss://" : "ws://"' in page_src,
      "https: origin gives wss://, and ws:// is reachable only on a KR_TLS=0 build")
# The guard the other way round (task 05): a TLS build's page on an http: origin
# refuses to connect and refuses the passphrase. The behaviour is walked in
# tests/test_page.js; these pins keep the refusal from being edited into a warning.
check("function servedInClear()" in page_src and "if (servedInClear())" in page_src,
      "the page asks whether it arrived in the clear, in connect() and at load")
check("this page arrived in the clear; reload over https" in page_src,
      "and it says why, in words the operator can act on")
check('$("token").disabled = true' in page_src,
      "a clear-served page will not even take passphrase input")
check("paintChannel(s)" in page_src and 'id="certFp"' in page_src
      and 'id="pTls"' in page_src,
      "the status frame's tls/fp pair is rendered: a pill and a certificate line")

print("the client side of pinning")
# The half of phase 02 the operator types. None of this may guess: no TLS argument
# means no connection, a wrong fingerprint means nothing was sent, and the only
# path to the plaintext port is the flag that says it knows what it is doing.
import contextlib                                          # noqa: E402
import hashlib                                             # noqa: E402
import io                                                  # noqa: E402
import os                                                  # noqa: E402
import socket as socketmod                                 # noqa: E402

# --- the decision functions, walked directly ---------------------------------
check_eq(relayctl.select_transport("a" * 64, "", False, False), "pin",
         "a fingerprint pins")
check_eq(relayctl.select_transport("", "/tmp/ca.pem", False, False), "ca",
         "a CA file verifies a chain")
check_eq(relayctl.select_transport("", "", True, False), "insecure",
         "--insecure is a decision, not a default")
check_eq(relayctl.select_transport("", "", False, True), "plain",
         "--plain is the rollback path")
for bad, why in ((("", "", False, False), "no trust argument at all"),
                 (("a" * 64, "/tmp/ca.pem", False, False), "a pin and a CA at once"),
                 (("", "", True, True), "--insecure together with --plain"),
                 (("a" * 64, "", False, True), "a pin on the rollback path")):
    try:
        relayctl.select_transport(*bad)
        check(False, f"select_transport should refuse: {why}")
    except relayctl.RelayError as exc:
        check(True, f"select_transport refuses: {why}")
        if why == "no trust argument at all":
            for option in ("--fingerprint", "--ca", "--insecure", "--plain"):
                check(option in str(exc),
                      f"the refusal names {option} as a way forward")

check_eq(relayctl.resolve_port("pin", None), 443, "a pinned run defaults to :443")
check_eq(relayctl.resolve_port("ca", None), 443, "a verified run defaults to :443")
check_eq(relayctl.resolve_port("insecure", None), 443,
         "even an unverified TLS run defaults to the TLS port")
check_eq(relayctl.resolve_port("plain", None), 80,
         "--plain is the only path to :80")
check_eq(relayctl.resolve_port("pin", 8443), 8443, "an explicit --port wins")

check_eq(relayctl.normalize_fingerprint("A" * 64), "a" * 64,
         "a bare hex fingerprint normalises to lowercase")
check_eq(relayctl.normalize_fingerprint(":".join("a1" for _ in range(32))), "a1" * 32,
         "OpenSSL's colon pairs are accepted")
check_eq(relayctl.normalize_fingerprint("  " + "ab" * 32 + "\n"), "ab" * 32,
         "and so is the whitespace a serial console wraps in")
for bad in ("", "deadbeef", "zz" * 32, "ab" * 31, "ab" * 32 + "0"):
    try:
        relayctl.normalize_fingerprint(bad)
        check(False, f"a malformed fingerprint is refused: {bad[:12]!r}")
    except relayctl.RelayError:
        check(True, f"malformed fingerprint refused: {bad[:12]!r}")

# --- the connection itself, against a fake board ------------------------------
FAKE_DER = b"-----BEGIN FAKE CERTIFICATE----- der-bytes pinning never sees"


# The status an adopted controller gets back to back with its `OK` -- the frame
# whose ABSENCE after `OK` is how a pending claim learns the board is busy. The
# values are the honest post-handover ones: disarmed, busy (with itself), one
# completed takeover.
FAKE_STATUS = b'S{"v":"1.0.0","armed":0,"usb":1,"typing":0,"busy":1,"tkov":1}'


class FakeBoardSock:
    """Records every byte this side sends; answers the upgrade and `T<token>`.

    The recorded frame list is the point: "the pin aborted before T<token>" can
    only be shown by what did NOT go out.

    `claim_slot=True` makes the board a BUSY one: `T` authenticates into the spare
    slot as a pending claim (`OK`, and deliberately no adoption status -- the status
    frame belongs to the controller alone), and the board adopts the client only
    when it sends the one deliberate `X`. That asymmetry is the whole of phase 03
    from this side of the socket, so the fake spells it out instead of collapsing
    both answers into "OK".
    """

    def __init__(self, host, port, close_early=False, bad_accept=False, flood=False,
                 claim_slot=False):
        self.host, self.port = host, port
        self.bad_accept = bad_accept   # answer the upgrade with a wrong accept hash
        self.flood = flood             # answer with headers that never terminate
        self.claim_slot = claim_slot   # busy board: park the newcomer as a claim
        self.sent = []                 # every sendall payload, in order
        self.inbox = bytearray()
        self.upgraded = False
        self.post = bytearray()        # bytes after the 101 (client WS frames)
        self.closed = False
        self.timeout = None
        self.close_early = close_early   # answer the upgrade with a clean EOF

    def settimeout(self, t):
        self.timeout = t

    def sendall(self, data):
        data = bytes(data)
        self.sent.append(data)
        if self.close_early:
            return                     # received, but the board answers nothing
        if not self.upgraded:
            blob = b"".join(self.sent)
            if b"\r\n\r\n" in blob:
                m = re.search(rb"Sec-WebSocket-Key: (\S+)", blob)
                if self.flood:
                    self.inbox += b"HTTP/1.1 101 " + b"x" * 20000
                    self.upgraded = True
                    return
                accept = (b"not-the-accept" if self.bad_accept
                          else relayctl.expect_accept(m.group(1).decode()).encode())
                self.inbox += (b"HTTP/1.1 101 Switching Protocols\r\n"
                               b"Sec-WebSocket-Accept: " + accept + b"\r\n\r\n")
                self.upgraded = True
            return
        self.post += data
        while len(self.post) >= 2:
            length = self.post[1] & 0x7F
            key = self.post[2:6]
            payload = bytes(b ^ key[i & 3] for i, b in enumerate(self.post[6:6 + length]))
            del self.post[:6 + length]
            if payload.startswith(b"T"):
                self.inbox += relayctl_server_frame(relayctl.OP_TEXT, b"OK")
                if not self.claim_slot:
                    # The adoption status a real controller gets back to back with
                    # the OK -- the auth path sends both on one write.
                    self.inbox += relayctl_server_frame(relayctl.OP_TEXT, FAKE_STATUS)
            elif payload == b"X":
                # Only a claim has anything to take; this fake only ever sees X
                # from one. The handover lands disarmed, and the adopted session
                # gets exactly the greeting a fresh authentication gets.
                if self.claim_slot:
                    self.claim_slot = False
                    self.inbox += relayctl_server_frame(relayctl.OP_TEXT, b"OK")
                    self.inbox += relayctl_server_frame(relayctl.OP_TEXT, FAKE_STATUS)
            elif payload == b"H":
                self.inbox += relayctl_server_frame(relayctl.OP_TEXT, FAKE_STATUS)

    def recv(self, n):
        if self.inbox:
            data = bytes(self.inbox[:n])
            del self.inbox[:n]
            return data
        if self.close_early:
            return b""                 # clean EOF: the board hung up mid-handshake
        raise socketmod.timeout("fake board: no more bytes right now")

    def close(self):
        self.closed = True


def relayctl_server_frame(opcode, payload):
    out = bytearray([0x80 | opcode, len(payload)])
    out += payload
    return bytes(out)


class FakeSsl:
    """Stand-in for the ssl module: real policy code runs against these knobs, and
    wrap_socket returns the same socket with a getpeercert of our choosing."""

    PROTOCOL_TLS_CLIENT = "tls-client"
    CERT_NONE = 0
    CERT_REQUIRED = 2

    class SSLError(Exception):
        pass

    def __init__(self, der=FAKE_DER, wrap_error=None, empty_der=False):
        self.der = der
        self.wrap_error = wrap_error   # raised by wrap_socket, simulating a failed TLS
        self.empty_der = empty_der     # peer completes TLS but shows no certificate
        self.contexts = []             # (kind, ctx) for every context built

    def SSLContext(self, protocol):
        ctx = FakeSsl.Ctx(self, protocol)
        self.contexts.append(("manual", ctx))
        return ctx

    def create_default_context(self, cafile=None):
        ctx = FakeSsl.Ctx(self, "default")
        ctx.cafile = cafile
        self.contexts.append(("default", ctx))
        return ctx

    class Ctx:
        def __init__(self, owner, protocol):
            self.owner, self.protocol = owner, protocol
            self.check_hostname = None
            self.verify_mode = None
            self.cafile = None
            self.server_hostname = None

        def wrap_socket(self, sock, server_hostname=None):
            self.server_hostname = server_hostname
            if self.owner.wrap_error is not None:
                raise self.owner.wrap_error
            der = b"" if self.owner.empty_der else self.owner.der
            sock.getpeercert = lambda binary_form=False: der if binary_form else {}
            return sock


def connect_with(fingerprint="", ca="", insecure=False, plain=False, port=None,
                 der=FAKE_DER, token="sekrit", wrap_error=None, close_early=False,
                 empty_der=False, bad_accept=False, flood=False, claim_slot=False,
                 take_over=False,
                 real_create=socketmod.create_connection, real_ssl=relayctl.ssl):
    fake = FakeSsl(der, wrap_error=wrap_error, empty_der=empty_der)
    socks = []

    def fake_create(addr, timeout):
        s = FakeBoardSock(addr[0], addr[1], close_early=close_early,
                          bad_accept=bad_accept, flood=flood, claim_slot=claim_slot)
        socks.append(s)
        return s

    relayctl.ssl = fake
    relayctl.socket.create_connection = fake_create
    stderr = io.StringIO()
    relay, err = None, None
    try:
        with contextlib.redirect_stderr(stderr):
            relay = relayctl.Relay("keyrelay.local", port, token, 1.0,
                                   fingerprint=fingerprint, ca=ca,
                                   insecure=insecure, plain=plain,
                                   take_over=take_over)
    except relayctl.RelayError as exc:
        err = exc
    finally:
        relayctl.ssl = real_ssl
        relayctl.socket.create_connection = real_create
    return relay, err, socks, fake, stderr.getvalue()


def client_frame_payloads(sock):
    """Decode the masked client frames this side sent, skipping the HTTP upgrade."""
    blob = b"".join(sock.sent)
    tail = blob.split(b"\r\n\r\n", 1)[1] if b"\r\n\r\n" in blob else b""
    out = []
    while len(tail) >= 2:
        length = tail[1] & 0x7F
        key = tail[2:6]
        out.append(bytes(b ^ key[i & 3] for i, b in enumerate(tail[6:6 + length])))
        tail = tail[6 + length:]
    return out

# A wrong fingerprint aborts BEFORE a single frame -- above all before T<token>.
_wrong = "0" * 64
relay, err, socks, fake, warns = connect_with(fingerprint=_wrong)
check(isinstance(err, relayctl.RelayError), "a wrong fingerprint raises")
check(err is not None and "fingerprint mismatch" in str(err),
      "and says exactly what mismatched")
check(socks and socks[0].sent == [],
      "nothing at all went out -- not the upgrade, and certainly not T<token>")

# The right fingerprint (in any sane spelling) connects, and the passphrase frame
# is the first thing on the encrypted wire.
_right = hashlib.sha256(FAKE_DER).hexdigest()
relay, err, socks, fake, warns = connect_with(fingerprint=_right)
check(err is None and relay is not None, f"the pinned fingerprint connects ({err})")
check(relay is not None and relay.mode == "pin", "the mode is the pin")
if socks:
    frames = client_frame_payloads(socks[0])
    check(frames and frames[0] == b"Tsekrit",
          "the first frame on a pinned session is the authentication")
    check(b"Tsekrit" not in b"".join(socks[0].sent),
          "the token never travels as plaintext bytes -- it is only in masked frames")
check(socks and socks[0].port == 443, "pinning dials :443")
relay2, err2, socks2, _, _ = connect_with(
    fingerprint=":".join(_right[i:i + 2] for i in range(0, 64, 2)).upper())
check(err2 is None, "a colon-separated uppercase pin is the same pin")

# TLS failures, in the two shapes the operator can actually act on.
relay, err, socks, fake, warns = connect_with(
    fingerprint=_right, wrap_error=FakeSsl.SSLError("handshake refused"))
check(err is not None and "TLS handshake with keyrelay.local failed" in str(err),
      "a failed TLS handshake is named as that")
check(err is not None and "426" in str(err),
      "and it names the plaintext-port mistake without ever proposing to make it")
check(socks and socks[0].closed, "and the socket is closed on the way out")

relay, err, socks, fake, warns = connect_with(
    ca="/tmp/ca.pem", wrap_error=FakeSsl.SSLError("certificate verify failed"))
check(err is not None and "failed verification" in str(err),
      "a --ca verification failure is not collapsed into 'handshake failed'")
check(err is not None and "nvs erase" in str(err),
      "and it says the two things a verify failure can mean")

relay, err, socks, fake, warns = connect_with(fingerprint=_right, empty_der=True)
check(err is not None and "no certificate" in str(err),
      "a TLS peer that shows no certificate is refused, not hashed as empty")
check(socks and socks[0].closed, "and that socket is closed too")

# --plain against a board that hangs up instead of speaking WebSocket.
relay, err, socks, fake, warns = connect_with(plain=True, close_early=True)
check(err is not None and "server closed during the handshake" in str(err),
      "a closed handshake is reported for what it is")

relay, err, socks, fake, warns = connect_with(plain=True, bad_accept=True)
check(err is not None and "Sec-WebSocket-Accept" in str(err),
      "a wrong accept hash refuses -- the RFC 6455 check is not decoration")

relay, err, socks, fake, warns = connect_with(plain=True, flood=True)
check(err is not None and "handshake response too large" in str(err),
      "and a header block that never ends is bounded, not buffered forever")

# --ca verifies with OpenSSL's own machinery: hostname checking ON, chain REQUIRED.
relay, err, socks, fake, warns = connect_with(ca="/tmp/board-ca.pem")
check(err is None, f"--ca connects ({err})")
kinds = [k for k, _ in fake.contexts]
check(kinds == ["default"], "--ca builds its context with create_default_context")
ctx = fake.contexts[0][1]
check_eq(ctx.cafile, "/tmp/board-ca.pem", "and loads the file the operator named")
check_eq(ctx.check_hostname, True, "--ca keeps hostname checking on")
check_eq(ctx.verify_mode, 2, "and requires the chain")
check(ctx.server_hostname == "keyrelay.local",
      "the hostname travels -- SNI and the check share one value")

# --insecure: connects, warns, and warns on EVERY run (it is stderr, not a memory).
relay, err, socks, fake, warns = connect_with(insecure=True)
check(err is None, "--insecure connects")
check("--insecure" in warns and "not verifying" in warns,
      "--insecure is loud about what it skipped")
check_eq(socks[0].port, 443, "and still uses the TLS port")
_, _, _, fake_p, warns_p = connect_with(plain=True)
check(fake_p.contexts == [], "--plain builds no TLS context at all")
check(warns_p == "", "--plain warns about nothing it did not skip")

# --plain is the only way to :80, and it is explicit every time.
relay, err, socks, fake, warns = connect_with(plain=True)
check(err is None and socks[0].port == 80, "--plain reaches :80")
frames = client_frame_payloads(socks[0])
check(frames and frames[0] == b"Tsekrit",
      "the rollback session authenticates the same way")

print("the busy board and --take-over")
# Phase 03, from the terminal side. A busy board authenticates this tool into its
# one spare slot as a pending claim -- `OK` and then the silence that means `busy`,
# because the status frame belongs to the controller alone. The default is a
# refusal that names the flag; the flag is one deliberate `X`; a free board is a
# no-op. The frames that DO go out are asserted, in order, because "takeover can
# never arm" is a sentence about what a session looks like at the moment it takes
# over: T, X, and nothing that types.

# Default against a busy board: refuse, name --take-over, send no X, free the slot.
relay, err, socks, fake, warns = connect_with(plain=True, claim_slot=True)
check(isinstance(err, relayctl.RelayError), "a busy board refuses by default")
check(err is not None and "--take-over" in str(err), "and the refusal names the flag")
check(err is not None and "DISARMED" in str(err),
      "and says what the handover lands on, before anyone takes it")
check_eq(client_frame_payloads(socks[0]), [b"Tsekrit"],
         "the refusal sends no X -- the act needs the flag, not just the passphrase")
check(socks and socks[0].closed, "and the claim slot is freed rather than parked on")

# With the flag: the claim's one deliberate frame, in the right order, alone.
relay, err, socks, fake, warns = connect_with(plain=True, claim_slot=True,
                                              take_over=True)
check(err is None and relay is not None, f"--take-over takes a busy board ({err})")
check(relay is not None and relay.took_over, "and the session knows it took the bridge")
check_eq(client_frame_payloads(socks[0]), [b"Tsekrit", b"X"],
         "T then X, in that order, and nothing else -- one intent, no arm, no state")

# --take-over against a free board is a no-op, not an error, and never wastes an X
# on a board that has nothing to take.
relay, err, socks, fake, warns = connect_with(plain=True, take_over=True)
check(err is None and relay is not None, f"--take-over on a free board succeeds ({err})")
check(relay is not None and relay.busy is False and not relay.took_over,
      "and it reports the board was free")
check_eq(client_frame_payloads(socks[0]), [b"Tsekrit"],
         "a free board is never sent an X it did not need")

# --- main(): the refusals that need no board -----------------------------------


def run_main(argv, fp_env=None, token_env="hunter2secret", claim_slot=False,
             socks_out=None):
    """Run main() with the transport faked: create_connection records the address
    instead of dialing, and ssl is the fake from above. Any credential that
    reaches the returned text is a test failure, so the caller never has to
    trust this helper to scrub it."""
    calls = []

    def record_create(addr, timeout):
        calls.append(addr)
        s = FakeBoardSock(addr[0], addr[1], claim_slot=claim_slot)
        if socks_out is not None:
            socks_out.append(s)
        return s

    old_argv = sys.argv
    old_env = os.environ.get("KEYRELAY_TLS_FP"), os.environ.get("KEYRELAY_TOKEN")
    sys.argv = ["relayctl.py"] + argv
    if fp_env is None:
        os.environ.pop("KEYRELAY_TLS_FP", None)
    else:
        os.environ["KEYRELAY_TLS_FP"] = fp_env
    if token_env is None:
        os.environ.pop("KEYRELAY_TOKEN", None)
    else:
        os.environ["KEYRELAY_TOKEN"] = token_env
    real_ssl_main = relayctl.ssl
    relayctl.ssl = FakeSsl()
    relayctl.socket.create_connection = record_create
    out, err_stream = io.StringIO(), io.StringIO()
    code, exc = None, None
    try:
        with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err_stream):
            code = relayctl.main()
    except BaseException as e:          # noqa: BLE001 -- reported, then re-raised below
        exc = e
    finally:
        sys.argv = old_argv
        for name, value in zip(("KEYRELAY_TLS_FP", "KEYRELAY_TOKEN"), old_env):
            if value is None:
                os.environ.pop(name, None)
            else:
                os.environ[name] = value
        relayctl.socket.create_connection = real_create_main
        relayctl.ssl = real_ssl_main
    if exc is not None:
        raise exc
    return code, out.getvalue(), err_stream.getvalue(), calls


real_create_main = socketmod.create_connection

code, out, err_text, calls = run_main(["--host", "keyrelay.local", "status"])
check_eq(code, 2, "no trust argument: main refuses with a usage error")
check(calls == [], "and refuses without opening a socket")
for option in ("--fingerprint", "--ca", "--insecure", "--plain"):
    check(option in err_text, f"the refusal names {option}")
check("hunter2secret" not in err_text and "hunter2secret" not in out,
      "the refusal prints no credential")

code, out, err_text, calls = run_main(["--fingerprint", "nothex", "status"])
check_eq(code, 2, "a malformed pin is refused before the network too")
check(calls == [] and "SHA-256" in err_text, "with the shape of what a pin must be")

# $KEYRELAY_TLS_FP is honoured like $KEYRELAY_TOKEN: the gate passes and the
# socket that follows is the board's. The fake's DER does not match the pin,
# so the run ends at the mismatch -- proof the pin is wired through main, and
# the token (never echoed) stays in the terminal.
code, out, err_text, calls = run_main(["--timeout", "0.1", "status"],
                                      fp_env="dead" * 16)
check(calls == [("keyrelay.local", 443)],
      "$KEYRELAY_TLS_FP satisfied the gate and selected :443")
check(code == 1 and "fingerprint mismatch" in err_text,
      "and a pin that does not match still aborts the run, through main()")
check("hunter2secret" not in err_text and "hunter2secret" not in out,
      "the passphrase is not printed on the abort path")

code, out, err_text, calls = run_main(["paste"], fp_env="ab" * 32, token_env=None)
check_eq(code, 2, "the token pre-flight still runs before anything else")
check("passphrase" in err_text and calls == [],
      "no token: refused in the terminal, exactly as before TLS")
code, out, err_text, calls = run_main(["paste"], fp_env="ab" * 32)
check_eq(code, 4, "paste still needs --yes before any connection")
check(calls == [], "and that check still does not require a reachable board")

# The whole takeover gesture through main(): take the held bridge, then read the
# board. T, X, H -- claim, handover, status -- with the terminal told which path
# ran, and no credential in anything it prints.
board = []
code, out, err_text, calls = run_main(["--plain", "--take-over", "status"],
                                      token_env="sekrit", claim_slot=True,
                                      socks_out=board)
check_eq(code, 0, f"--take-over status against a busy board runs ({err_text})")
session = client_frame_payloads(board[0]) if board else []
check_eq([f for f in session if f in (b"Tsekrit", b"X", b"H")],
         [b"Tsekrit", b"X", b"H"],
         "the busy session's frames are T, then X, then H -- claim, handover, status")
check("took the bridge" in out and "DISARMED" in out,
      "and the terminal is told what the handover did before anything else runs")
check("sekrit" not in out and "sekrit" not in err_text,
      "and the passphrase stays out of the terminal")

board = []
code, out, err_text, calls = run_main(["--plain", "--take-over", "status"],
                                      token_env="sekrit", socks_out=board)
check_eq(code, 0, f"--take-over against a free board still runs ({err_text})")
check("nothing to take" in out, "and says it was a no-op, in those words")
check_eq([f for f in (client_frame_payloads(board[0]) if board else []) if f == b"X"],
         [], "with no X on the wire at all")

# The default against a busy board, through main(): exit 1, the refusal names
# --take-over, and the wire never carried an X.
code, out, err_text, calls = run_main(["--plain", "status"], token_env="sekrit",
                                      claim_slot=True, socks_out=board)
check_eq(code, 1, "a busy board is a refusal through main() too")
check("--take-over" in err_text, "and the refusal names the flag in the terminal")
check_eq([f for f in (client_frame_payloads(board[-1]) if board else [])
          if f == b"X"], [], "and refused means refused: no X went out anyway")

print("the target-side verifier")
# Twice now the verifier itself was the bug -- mislabelled LEDs, unnamed key codes that
# looked like lost keystrokes -- so the thing that decides what "it arrived" means gets
# tested too, and the mouse half of it especially: it is new, and a watcher that misread
# REL_X would either cry wolf or miss a mouse that worked.
import watch_target as wt                                    # noqa: E402

# EV_REL is what makes a node a pointing device. Nothing here trusts a device name.
check_eq(wt.classify(1 << wt.EV_REL | 1 << wt.EV_KEY), "mouse", "EV_REL means mouse")
check_eq(wt.classify(1 << wt.EV_KEY | 1 << wt.EV_LED), "keyboard",
         "EV_KEY without EV_REL means keyboard")
check_eq(wt.classify(1 << wt.EV_ABS), None,
         "an absolute-only device is neither, and is not watched as a mouse")
check_eq(wt.capability_bits("/nonexistent/does/not/exist"), 0,
         "an unreadable sysfs file is 0 rather than an exception")

m = wt.MouseWatch(False)
check(m.event(0, 0, 20) is None and m.event(0, 1, -6) is None,
      "a raw axis event is not a line -- one report is one line")
check(m.event(0, 0, 15) is None, "and more movement in the same report accumulates")
line = m.flush(0.1)
check(line is not None and "+35" in line and "-6" in line,
      f"the report comes out as the sum of its axes ({line!r})")
check(m.flush(0.2) is None, "a report with no movement prints nothing")
check_eq(m.x, 35, "net X is the exact sum of what the target received")
check_eq(m.y, -6, "and so is net Y")
check_eq(m.travel, 41, "and the path length counts both, not the diagonal")

# REL_WHEEL_HI_RES is 1/240 of a notch. Counting it as a notch would report 240x the
# scroll that happened, which is the same class of error as the mislabelled LEDs.
# The axis numbers, pinned to the kernel's, not to the HID report's byte order. Getting
# this wrong made a working wheel look dead -- see the comment on REL_NAMES.
check_eq((wt.REL_X, wt.REL_Y, wt.REL_WHEEL, wt.REL_HWHEEL), (0, 1, 8, 6),
         "axis codes are the kernel's: REL_WHEEL is 8, not 2")
check_eq((wt.REL_WHEEL_HI_RES, wt.REL_HWHEEL_HI_RES), (11, 12),
         "and the hi-res wheel codes are 11/12, not 6/7")

m2 = wt.MouseWatch(False)
check(m2.event(0, wt.REL_WHEEL, -3) is not None,
      "a wheel notch is printed as it happens, not coalesced with movement")
m2.event(0, wt.REL_WHEEL_HI_RES, 720)   # the same notches again, finer
check_eq(m2.wheel, -3, "hi-res wheel events are not counted as extra notches")
check("REL_WHEEL" in "\n".join(m2.report()),
      "and the summary names the axis it saw, so a wheel that arrived is visible")
m3 = wt.MouseWatch(False)
m3.event(0, 40, 5)                       # an axis this tool does not know
check(any("rel40" in l for l in m3.report()),
      "an unnamed axis code is counted and shown, never dropped in silence")
m2.event(0, 272, 1)
check("still held" in "\n".join(m2.report()),
      "a button the target still believes is down is called out, in those words")
m2.event(0, 272, 0)
check("still held" not in "\n".join(m2.report()),
      "and the warning goes away when it is released")

k = wt.KeyboardWatch(False)
k.event(0, 30, 1)
k.event(0, 30, 0)
check("a" in "".join(k.typed), "the keyboard watcher still reassembles typed text")
kc = wt.KeyboardWatch(False)
for code in (29, 56, 111):
    kc.event(0, code, 1)
check(any("Ctrl+Alt+Del" in l for l in kc.report()),
      "and still names Ctrl+Alt+Del as one combination, not three keys")

print(f"relayctl              {checks} checks, {failures} failures")
sys.exit(1 if failures else 0)