Add interactive reporting for local Autobahn runs

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

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)

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()

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)

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