Stream TAP output line by line

9pm.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python3
 
 import argparse
+import errno
 import json
 import os
+import pty
 import yaml
 import subprocess
 import sys
@@ -53,21 +55,43 @@ def rootify_path(path):
     path = os.path.normpath(path)
     return path
 
+def pty_read_lines(fd):
+    # PTY line discipline turns '\n' into '\r\n' on the way out, so trailing
+    # '\r' is stripped from each yielded line.
+    buf = b""
+    while True:
+        try:
+            chunk = os.read(fd, 4096)
+        except OSError as e:
+            if e.errno != errno.EIO:
+                raise
+            chunk = b""
+        if not chunk:
+            if buf:
+                yield buf.rstrip(b'\r')
+            return
+        buf += chunk
+        while b'\n' in buf:
+            line, buf = buf.split(b'\n', 1)
+            yield line.rstrip(b'\r')
+
 def execute(args, test, output_log):
     os.environ["NINEPM_TEST_NAME"] = test['unix_name']
-    proc = subprocess.Popen([test['case']] + args, stdout=subprocess.PIPE)
+
+    # Run the test with stdout connected to a pty so its stdio sees a TTY and
+    # line-buffers, instead of block-buffering through a pipe until exit.
+    master_fd, slave_fd = pty.openpty()
+    proc = subprocess.Popen([test['case']] + args, stdout=slave_fd)
+    os.close(slave_fd)
+
     skip_suite = False
     test_skip = False
     err = False
 
     # Test metadata is now handled in the report generation, not in the log
 
-    while True:
-        line = proc.stdout.readline().decode('utf-8')
-        if line == '':
-            break
-
-        string = line.rstrip()
+    for raw in pty_read_lines(master_fd):
+        string = raw.decode('utf-8', errors='replace').rstrip()
         stamp = time.strftime("%Y-%m-%d %H:%M:%S")
 
         plan = re.search(r'^(\d+)..(\d+)$', string)
@@ -103,7 +127,8 @@ def execute(args, test, output_log):
         else:
             print(f"{stamp} {string}")
 
-    out, error = proc.communicate()
+    os.close(master_fd)
+    proc.wait()
     exitcode = proc.returncode
 
     if exitcode != 0:
@@ -766,6 +791,10 @@ def main():
     global VERBOSE
     global NOEXEC
 
+    # Line-buffer our own stdout so output streams when 9pm itself is piped to
+    # another program (CI, tee, wrappers) instead of buffering until exit.
+    sys.stdout.reconfigure(line_buffering=True)
+
     sha = ""
     if (sha := run_git_cmd(ROOT_PATH, ['rev-parse', 'HEAD'])):
         sha = f"({sha[:10]})"

self_test/cases/line-buffering.sh

@@ -0,0 +1,8 @@
+#!/bin/sh
+
+SLEEP=2
+
+echo "1..2"
+echo "ok 1 - about to sleep ${SLEEP}s"
+sleep "$SLEEP"
+echo "ok 2 - slept ${SLEEP}s"

self_test/run.py

@@ -3,10 +3,12 @@
 import subprocess
 import os
 import json
+import re
 import sys
 import tempfile
 import argparse
 import uuid
+from datetime import datetime
 
 
 VERBOSE = False
@@ -377,6 +379,31 @@ def find_test_spec(obj):
 
         print_green(f"[PASS] Test spec JSON output")
 
+    def test_line_buffering(self):
+        """Verify TAP lines stream as emitted, not buffered until test exit"""
+
+        result = subprocess.run(
+            ["python3", self.ninepm, "cases/line-buffering.sh"],
+            cwd=self.script_dir,
+            text=True,
+            env=self.env,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+
+        assert result.returncode == 0, f"Failed with return code {result.returncode}. stderr: {result.stderr}"
+
+        stamps = re.findall(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) ok \d', result.stdout)
+        assert len(stamps) >= 2, f"Expected 2 stamped ok lines, got: {result.stdout}"
+
+        t1 = datetime.strptime(stamps[0], "%Y-%m-%d %H:%M:%S")
+        t2 = datetime.strptime(stamps[1], "%Y-%m-%d %H:%M:%S")
+        delta = (t2 - t1).total_seconds()
+
+        assert delta >= 1, f"TAP output appears buffered (stamps {stamps[0]!r} and {stamps[1]!r}, delta={delta}s)"
+
+        print_green(f"[PASS] TAP output is line-buffered")
+
     def cleanup(self):
         """Cleanup temp directory after tests"""
         self.temp_dir_base.cleanup()
@@ -407,6 +434,7 @@ def cleanup(self):
         tester.test_repeat_flag()
         tester.test_proj_config()
         tester.test_spec_json_format()
+        tester.test_line_buffering()
         print_green("All tests passed.")
     finally:
         tester.cleanup()