From a59526c781b48bd83db742a342f20f2452d2ed73 Mon Sep 17 00:00:00 2001 From: Benjamin Sergeant Date: Mon, 2 Nov 2026 13:30:46 -0800 Subject: [PATCH] Add interactive progress reporting for local Autobahn runs --- docs/autobahn.md | 18 ++++++++ makefile.dev | 6 +++ test/autobahn/test_progress.py | 83 ++++++++++++++++++++++++++++++++++ tools/autobahn_progress.py | 83 ++++++++++++++++++++++++++++++++++ tools/run_autobahn.py | 24 ++++++++-- 5 files changed, 210 insertions(+), 4 deletions(-) create mode 100644 test/autobahn/test_progress.py create mode 100644 tools/autobahn_progress.py diff --git a/docs/autobahn.md b/docs/autobahn.md index 8c1dcbcb..9891dd0b 100644 --- a/docs/autobahn.md +++ b/docs/autobahn.md @@ -12,6 +12,24 @@ The workflow calls one make target: make -f makefile.dev test_autobahn ``` +For a live local progress display, run: + +```sh +make -f makefile.dev test_autobahn_interactive +``` + +The display shows the current case ID, cases started, and elapsed time for each +role, followed by a validated PASS/FAIL summary and the report directory. Case +progress is not a pass count: results are checked after each role finishes. +A terminal updates one line in place; redirected output uses plain progress +lines. Ctrl-C stops the run and cleans up its containers. + +To select a smaller run through make: + +```sh +make -f makefile.dev test_autobahn_interactive AUTOBAHN_ARGS="--cases 6.4.* --role client" +``` + Docker and Python 3 are required. The library is built from the current checkout with C++11 and zlib enabled. This checks WebSocket framing, closing, UTF-8, and compression in both directions over plain WebSocket connections; it does not diff --git a/makefile.dev b/makefile.dev index 5ee1c742..0a86f78f 100644 --- a/makefile.dev +++ b/makefile.dev @@ -254,3 +254,9 @@ commit: test_autobahn: python3 -m unittest discover -s test/autobahn -p 'test_*.py' python3 tools/run_autobahn.py + +# Local live reporting; e.g. AUTOBAHN_ARGS="--cases 6.4.* --role client". +AUTOBAHN_ARGS ?= +.PHONY: test_autobahn_interactive +test_autobahn_interactive: + python3 tools/run_autobahn.py --interactive $(AUTOBAHN_ARGS) diff --git a/test/autobahn/test_progress.py b/test/autobahn/test_progress.py new file mode 100644 index 00000000..b67eccc3 --- /dev/null +++ b/test/autobahn/test_progress.py @@ -0,0 +1,83 @@ +"""Progress must reflect observed cases, without claiming compliance results.""" +import importlib.util +import io +from pathlib import Path +import subprocess +import unittest +from unittest.mock import patch + +spec = importlib.util.spec_from_file_location( + "autobahn_progress", Path(__file__).resolve().parents[2] / "tools/autobahn_progress.py") +progress_module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(progress_module) + + +class ProgressTest(unittest.TestCase): + def test_only_known_cases_are_counted_once(self): + output = io.StringIO() + progress = progress_module.Progress("client", ["1.1.1", "6.4.1"], output) + progress.read([ + "Running test case ID 1.1.1 for agent ixwebsocket from peer tcp4:host\n", + "Running test case ID 1.1.1 for agent ixwebsocket from peer tcp4:host\n", + "Running test case ID 9.9.9 for agent ixwebsocket from peer tcp4:host\n", + "Running test case ID 6.4.1 for agent other\n", + ]) + progress.display(final=True) + self.assertIn("client: 1/2 cases started | 1.1.1", output.getvalue()) + self.assertNotIn("PASS", output.getvalue()) + self.assertNotIn("\033", output.getvalue()) + + def test_terminal_updates_in_place(self): + output = io.StringIO() + output.isatty = lambda: True + progress = progress_module.Progress("server", ["1.1.1"], output) + progress.display() + self.assertTrue(output.getvalue().startswith("\r\033[2K")) + self.assertFalse(output.getvalue().endswith("\n")) + progress.display(final=True) + self.assertTrue(output.getvalue().endswith("\n")) + + def test_redirected_output_has_periodic_heartbeat(self): + output = io.StringIO() + with patch.object(progress_module.time, "monotonic", return_value=0) as clock: + progress = progress_module.Progress("client", ["1.1.1"], output) + progress.display() + clock.return_value = 1 + progress.display() + self.assertEqual(len(output.getvalue().splitlines()), 1) + clock.return_value = 30 + progress.display() + self.assertEqual(len(output.getvalue().splitlines()), 2) + + def test_timeout_stops_both_docker_helpers(self): + class Process: + def __init__(self): + self.stdout = io.StringIO("") + self.stderr = io.StringIO("") + self.args = ["docker"] + self.returncode = None + self.terminated = False + + def poll(self): + return self.returncode + + def terminate(self): + self.terminated = True + self.returncode = -15 + + def wait(self, timeout): + return self.returncode + + follower, waiter = Process(), Process() + with patch.object(progress_module.subprocess, "Popen", side_effect=[follower, waiter]), \ + patch.object(progress_module.sys, "stdout", io.StringIO()): + with self.assertRaises(subprocess.TimeoutExpired): + progress_module.wait_with_progress("test", "logs", "client", ["1.1.1"], 0) + self.assertTrue(follower.terminated) + self.assertTrue(waiter.terminated) + self.assertTrue(follower.stdout.closed) + self.assertTrue(waiter.stdout.closed) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/autobahn_progress.py b/tools/autobahn_progress.py new file mode 100644 index 00000000..f475b94d --- /dev/null +++ b/tools/autobahn_progress.py @@ -0,0 +1,83 @@ +"""Display case progress from Autobahn's log stream without interpreting results.""" +import re +import subprocess +import sys +import threading +import time + + +class Progress: + def __init__(self, role, expected, stream=None): + self.role = role + self.expected = set(expected) + self.stream = stream if stream is not None else sys.stdout + self.tty = self.stream.isatty() + self.started = time.monotonic() + self.seen = set() + self.current = "waiting for first case" + self.lock = threading.Lock() + self.last_line = None + self.last_display = 0 + + def read(self, lines): + for line in lines: + match = re.search(r"Running test case ID ([\d.]+) for agent ixwebsocket\b", line) + if match and match.group(1) in self.expected: + with self.lock: + self.current = match.group(1) + self.seen.add(self.current) + + def display(self, final=False): + elapsed = int(time.monotonic() - self.started) + with self.lock: + count, current = len(self.seen), self.current + line = "{}: {}/{} cases started | {} | {:02d}:{:02d} elapsed".format( + self.role, count, len(self.expected), current, elapsed // 60, elapsed % 60) + changed = self.last_line != (count, current) + if final or changed or self.tty or elapsed - self.last_display >= 30: + self.stream.write(("\r\033[2K" if self.tty else "") + line + + ("\n" if final or not self.tty else "")) + self.stream.flush() + self.last_line = (count, current) + self.last_display = elapsed + + +def wait_with_progress(container, log_container, role, expected, timeout): + progress = Progress(role, expected) + follower = subprocess.Popen( + ["docker", "logs", "--follow", log_container], stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, text=True) + reader = threading.Thread(target=progress.read, args=(follower.stdout,), daemon=True) + reader.start() + waiter = None + try: + waiter = subprocess.Popen(["docker", "wait", container], stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + deadline = time.monotonic() + timeout + while waiter.poll() is None: + progress.display() + if time.monotonic() >= deadline: + raise subprocess.TimeoutExpired(waiter.args, timeout) + try: + waiter.wait(timeout=1) + except subprocess.TimeoutExpired: + pass + output, error = waiter.communicate() + if waiter.returncode: + raise subprocess.CalledProcessError(waiter.returncode, waiter.args, output, error) + return int(output.strip()) + finally: + for process in (waiter, follower): + if process is not None and process.poll() is None: + process.terminate() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=3) + reader.join(timeout=3) + follower.stdout.close() + if waiter is not None: + waiter.stdout.close() + waiter.stderr.close() + progress.display(final=True) diff --git a/tools/run_autobahn.py b/tools/run_autobahn.py index 4e10b6db..45d96220 100644 --- a/tools/run_autobahn.py +++ b/tools/run_autobahn.py @@ -11,6 +11,7 @@ import uuid from check_autobahn import check_report +from autobahn_progress import wait_with_progress ROOT = Path(__file__).resolve().parents[1] SUITE = ("crossbario/autobahn-testsuite:25.10.1@sha256:" @@ -23,9 +24,10 @@ def command(args, **kwargs): class Runner: - def __init__(self, output, timeout): + def __init__(self, output, timeout, interactive=False): self.output = output self.timeout = timeout + self.interactive = interactive self.network = "ix-autobahn-" + uuid.uuid4().hex[:12] self.containers = [] self.image = None @@ -40,7 +42,7 @@ def start(self, name, image, args, suite=False, alias=None): if alias: options += ["--network-alias", alias] if suite: - options += ["--platform", "linux/amd64", "--volume", + options += ["--env", "PYTHONUNBUFFERED=1", "--platform", "linux/amd64", "--volume", str(self.output / "config") + ":/config:ro", "--volume", str(ROOT / "test/autobahn") + ":/harness:ro", "--volume", str(self.output / "reports") + ":/reports"] @@ -126,6 +128,7 @@ def run(self, roles, patterns): "patterns": patterns, "expected_count": len(expected)}, indent=2) + "\n") command(["docker", "network", "create", self.network], stdout=subprocess.DEVNULL) failed = False + results = [] try: for role in roles: print("Running {} compliance: {} cases".format(role, len(expected)), flush=True) @@ -146,7 +149,11 @@ def run(self, roles, patterns): "timeout", "--kill-after=10s", str(self.timeout), "wstest", "-m", "fuzzingclient", "-s", "/config/fuzzingclient.json"], suite=True) - status = self.wait(test) + if self.interactive: + status = wait_with_progress(test, peer if role == "client" else test, + role, expected, self.timeout + 30) + else: + status = self.wait(test) if status: print("{} runner failed with exit {}".format(role, status), flush=True) failed = True @@ -158,12 +165,19 @@ def run(self, roles, patterns): for error in errors: print("{}: {}".format(role, error), flush=True) failed = failed or bool(errors) + results.append((role, status == 0 and not errors)) print("{}: {} expected cases, {} report failures".format( role, len(expected), len(errors)), flush=True) self.collect_logs() command(["docker", "stop", "--time", "5", peer], stdout=subprocess.DEVNULL) finally: self.cleanup() + if self.interactive: + print("\nAutobahn summary", flush=True) + for role, passed in results: + print(" {}: {} ({} expected cases)".format( + role, "PASS" if passed else "FAIL", len(expected)), flush=True) + print("Reports: " + str(self.output / "reports"), flush=True) return int(failed) @@ -174,6 +188,8 @@ def interrupted(signum, frame): def main(): signal.signal(signal.SIGTERM, interrupted) parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--interactive", action="store_true", + help="Show live case progress and elapsed time") parser.add_argument("--role", choices=("client", "server", "both"), default="both") parser.add_argument("--cases", default="*", help="Comma-separated patterns for local debugging") parser.add_argument("--timeout", type=int, default=1800, help="Maximum seconds per role") @@ -199,7 +215,7 @@ def main(): parser.error("Another run owns this output directory: " + str(lock)) roles = ["client", "server"] if args.role == "both" else [args.role] try: - return Runner(output, args.timeout).run(roles, args.cases.split(",")) + return Runner(output, args.timeout, args.interactive).run(roles, args.cases.split(",")) except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error: print("Autobahn infrastructure failure: " + str(error), file=sys.stderr) return 1