# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence. """Drive a running fake server through a scenario, and adjudicate the result. Exits non-zero on any mismatch, so scripts/e2e.sh is a real gate rather than a log to read. """ from __future__ import annotations import argparse import asyncio import sys import time from pathlib import Path from typing import Any import httpx from fake_signal.scenario import Scenario, check, load_scenario # Measured, not guessed. A surplus send is *always* detected whatever this is, # because collect() re-reads after the expected count arrives -- injecting a # deliberate double-send proved that at every value tried, six runs, all red. # What this buys is **attribution**: whether the surplus is blamed on the # scenario that caused it or on the innocent one that follows. Measured with a # surplus delayed by hand: # # surplus at +0.05s / +0.25s blamed correctly at 0.35 and at 0.15 # surplus at +0.50s blamed correctly at 0.35, wrongly at 0.15 # # 0.15 halves the replay (35s -> 19s, against a 40s budget) and only # misattributes a surplus arriving half a second late, which took an explicit # asyncio.sleep to construct -- a real double-send comes from the same handler # microseconds apart. Raise it with --settle if a slow machine starts blaming # the wrong scenario; that symptom is what this number is for. SETTLE_SECONDS = 0.15 # How long a scenario may wait for the sends it expects before giving up. # # A single fixed settle cannot cover both a reply that is a dictionary lookup # and one that fetches an image and shells out to exiftool, and the failure # that produced was not "slow" but "wrong": the late send landed after the # NEXT scenario's reset and was counted against that one instead. So two # different scenarios failed per run, never the same two twice -- one with # "captured 0" and one with "captured 2". Waiting for the count is a # guarantee; a bigger number is a guess that fails again on a slower machine. PATIENCE_SECONDS = 15.0 POLL_SECONDS = 0.05 def _emit(line: str) -> None: """Write one report line to stdout.""" sys.stdout.write(f"{line}\n") def resolve_groups(args: argparse.Namespace) -> dict[str, str]: """Map the scenario's symbolic group names onto real ids. Names are supplied as repeated ``--group NAME=ID`` pairs rather than one flag per group, because the bot now watches an arbitrary number of them and a scenario has to be able to say "this reply belongs in the *other* watched group". Args: args: Parsed command line. Returns: Scenario alias to real group id. Raises: SystemExit: A --group argument is not NAME=ID. """ groups: dict[str, str] = {} for pair in args.group: name, separator, value = pair.partition("=") if not separator or not name or not value: msg = f"--group expects NAME=ID, got {pair!r}" raise SystemExit(msg) groups[name] = value return groups async def captured(client: httpx.AsyncClient) -> list[dict[str, Any]]: """Every send the fake server has recorded since the last reset.""" response = await client.get("/_control/sent") response.raise_for_status() sends: list[dict[str, Any]] = response.json() return sends async def collect( client: httpx.AsyncClient, want: int, *, settle: float, patience: float ) -> list[dict[str, Any]]: """Wait for the sends a scenario expects, then for a beat of quiet. The quiet beat is not politeness: returning the moment the expected count arrives would make "expected 1, captured 2" unobservable, because the surplus send is exactly the one that has not landed yet. A scenario that expects nothing skips the poll entirely and waits out the settle, which is the only way to assert that nothing happens. Args: client: A client pointed at the fake server's control plane. want: How many sends the scenario expects. settle: How long "quiet" is. patience: How long to wait for ``want`` sends before giving up. Returns: Every send recorded, which may be more or fewer than expected -- the comparison is somebody else's job. """ deadline = time.monotonic() + patience sends = await captured(client) while len(sends) < want and time.monotonic() < deadline: await asyncio.sleep(POLL_SECONDS) sends = await captured(client) await asyncio.sleep(settle) return await captured(client) async def run_scenario( base_url: str, scenario: Scenario, groups: dict[str, str], *, settle: float, patience: float = PATIENCE_SECONDS, ) -> list[str]: """Inject every frame, wait for the bot to react, then compare.""" async with httpx.AsyncClient(base_url=base_url, timeout=10.0) as client: await client.post("/_control/reset") if scenario.arrive: # Additive: the groups already on the account stay, exactly as # they would when somebody adds the bot to one more. await client.post( "/_control/groups/add", json=[arrival.to_entry(groups) for arrival in scenario.arrive], ) for injection in scenario.inject: await client.post("/_control/inject", json={"raw": injection.to_frame(groups)}) await asyncio.sleep(settle) sent = await collect(client, len(scenario.expect), settle=settle, patience=patience) return check(scenario, sent, groups) def parse_args(argv: list[str] | None = None) -> argparse.Namespace: """Parse the injector's command line.""" parser = argparse.ArgumentParser(description="Replay a scenario against the bot.") parser.add_argument("scenarios", nargs="+", type=Path) parser.add_argument("--base-url", default="http://127.0.0.1:8099") parser.add_argument("--group", action="append", default=[], metavar="NAME=ID") parser.add_argument("--settle", type=float, default=SETTLE_SECONDS) # Overridable so the tests can drive the give-up path without waiting it # out, and so a slow machine can be given longer without editing source. parser.add_argument("--patience", type=float, default=PATIENCE_SECONDS) return parser.parse_args(argv) async def _main_async(args: argparse.Namespace) -> int: """Run every scenario, reporting failures as they happen.""" groups = resolve_groups(args) failed = 0 for path in args.scenarios: scenario = load_scenario(path) failures = await run_scenario( args.base_url, scenario, groups, settle=args.settle, patience=args.patience ) if failures: failed += 1 _emit(f"FAIL {scenario.name}") for line in failures: _emit(f" {line}") else: _emit(f"ok {scenario.name}") return 1 if failed else 0 def main(argv: list[str] | None = None) -> int: """Entry point. Returns a process exit code.""" return asyncio.run(_main_async(parse_args(argv))) if __name__ == "__main__": sys.exit(main())