tests: fix tap_runner to route diagnostic output to stderr

Agent-Logs-Url: https://github.com/igaw/nvme-cli/sessions/7ae055d8-4c72-40a1-bd86-f23660b6c0bb

Co-authored-by: igaw <1050803+igaw@users.noreply.github.com>

Author: Copilot

tests/nvme_test.py

@@ -36,31 +36,6 @@
 from nvme_test_logger import TestNVMeLogger
 
 
-def find_config_file(filename='config.json'):
-    """ Search for a configuration file by walking up the directory tree.
-
-    Starting from the directory containing this module, walk toward the
-    filesystem root until a file with the given name is found.
-        - Args:
-            - filename: name of the config file to search for
-        - Returns:
-            - Absolute path to the config file
-        - Raises:
-            - FileNotFoundError if the file is not found in any ancestor directory
-    """
-    directory = os.path.dirname(os.path.abspath(__file__))
-    while True:
-        candidate = os.path.join(directory, filename)
-        if os.path.isfile(candidate):
-            return candidate
-        parent = os.path.dirname(directory)
-        if parent == directory:
-            raise FileNotFoundError(
-                f"Config file '{filename}' not found in '{os.path.dirname(os.path.abspath(__file__))}' "
-                f"or any of its parent directories")
-        directory = parent
-
-
 def to_decimal(value):
     """ Wrapper for converting numbers to base 10 decimal
         - Args:
@@ -102,7 +77,7 @@ def setUp(self):
         self.do_validate_pci_device = True
         self.default_nsid = 0x1
         self.flbas = 0
-        self.config_file = find_config_file()
+        self.config_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config.json')
 
         self.load_config()
         if self.do_validate_pci_device:

tests/tap_runner.py

@@ -15,86 +15,18 @@
 import argparse
 import importlib
 import io
-import os
 import sys
-import threading
 import traceback
 import unittest
 
 
-class DiagnosticCapture(io.TextIOBase):
-    """Capture writes and re-emit them as TAP diagnostic lines (# ...)."""
-
-    def __init__(self, real_stdout: io.TextIOBase) -> None:
-        self._real = real_stdout
-        self._buf = ''
-
-    def write(self, text: str) -> int:
-        self._buf += text
-        while '\n' in self._buf:
-            line, self._buf = self._buf.split('\n', 1)
-            self._real.write('# {}\n'.format(line))
-        self._real.flush()
-        return len(text)
-
-    def flush(self) -> None:
-        if self._buf:
-            self._real.write('# {}\n'.format(self._buf))
-            self._buf = ''
-        self._real.flush()
-
-
-class FDCapture:
-    """Redirect a file descriptor at the OS level and re-emit captured output
-    as TAP diagnostic lines.  This intercepts writes from subprocesses which
-    bypass the Python-level sys.stderr redirect."""
-
-    def __init__(self, fd: int, real_stdout: io.TextIOBase) -> None:
-        self._fd = fd
-        self._real = real_stdout
-        self._saved_fd = os.dup(fd)
-        r_fd, w_fd = os.pipe()
-        os.dup2(w_fd, fd)
-        os.close(w_fd)
-        self._thread = threading.Thread(target=self._reader, args=(r_fd,),
-                                        daemon=True)
-        # daemon=True: if restore() is somehow never called (e.g. os._exit()),
-        # the process can still exit rather than hang on a blocking read.
-        self._thread.start()
-
-    def _reader(self, r_fd: int) -> None:
-        buf = b''
-        # Open unbuffered (bufsize=0) so bytes are delivered to the reader
-        # as soon as they are written, without waiting for a buffer to fill.
-        with open(r_fd, 'rb', 0) as f:
-            while True:
-                chunk = f.read(4096)
-                if not chunk:
-                    break
-                buf += chunk
-                while b'\n' in buf:
-                    line, buf = buf.split(b'\n', 1)
-                    self._real.write(
-                        '# {}\n'.format(line.decode('utf-8', errors='replace')))
-                    self._real.flush()
-        if buf:
-            self._real.write(
-                '# {}\n'.format(buf.decode('utf-8', errors='replace')))
-            self._real.flush()
-
-    def restore(self) -> None:
-        """Restore the original file descriptor and wait for the reader to drain."""
-        os.dup2(self._saved_fd, self._fd)
-        os.close(self._saved_fd)
-        self._thread.join()
-
-
 class TAPTestResult(unittest.TestResult):
     """Collect unittest results and render them as TAP version 13."""
 
-    def __init__(self, stream: io.TextIOBase) -> None:
+    def __init__(self, stream: io.TextIOBase, diag_stream: io.TextIOBase) -> None:
         super().__init__()
         self._stream = stream
+        self._diag_stream = diag_stream
         self._test_count = 0
 
     def _description(self, test: unittest.TestCase) -> str:
@@ -112,20 +44,20 @@ def addError(self, test: unittest.TestCase, err: object) -> None:
         self._test_count += 1
         self._stream.write('not ok {} - {}\n'.format(
             self._test_count, self._description(test)))
-        for line in traceback.format_exception(*err):  # type: ignore[misc]
-            for subline in line.splitlines():
-                self._stream.write('# {}\n'.format(subline))
         self._stream.flush()
+        self._diag_stream.write(
+            ''.join(traceback.format_exception(*err)))  # type: ignore[misc]
+        self._diag_stream.flush()
 
     def addFailure(self, test: unittest.TestCase, err: object) -> None:
         super().addFailure(test, err)
         self._test_count += 1
         self._stream.write('not ok {} - {}\n'.format(
             self._test_count, self._description(test)))
-        for line in traceback.format_exception(*err):  # type: ignore[misc]
-            for subline in line.splitlines():
-                self._stream.write('# {}\n'.format(subline))
         self._stream.flush()
+        self._diag_stream.write(
+            ''.join(traceback.format_exception(*err)))  # type: ignore[misc]
+        self._diag_stream.flush()
 
     def addSkip(self, test: unittest.TestCase, reason: str) -> None:
         super().addSkip(test, reason)
@@ -165,23 +97,17 @@ def run_tests(test_module_name: str, start_dir: str | None = None) -> bool:
     real_stdout.write('1..{}\n'.format(suite.countTestCases()))
     real_stdout.flush()
 
-    # Redirect stdout and stderr so any print()/sys.stderr.write() calls from
-    # setUp/tearDown/tests are re-emitted as TAP diagnostic lines and do not
-    # break the TAP stream.
-    sys.stdout = DiagnosticCapture(real_stdout)  # type: ignore[assignment]
-    sys.stderr = DiagnosticCapture(real_stdout)  # type: ignore[assignment]
-    # Also redirect fd 2 at the OS level so that subprocess stderr (which
-    # inherits the raw file descriptor and bypasses sys.stderr) is captured.
-    stderr_fd_capture = FDCapture(2, real_stdout)
+    # Redirect sys.stdout to real_stderr so that print()/sys.stdout.write()
+    # calls from setUp/tearDown/tests do not pollute the TAP stream on stdout.
+    # All diagnostic output (print statements, subprocess stderr via fd 2, and
+    # test failure tracebacks) goes to stderr so that 'meson test -v' shows it
+    # live on the terminal.
+    sys.stdout = real_stderr  # type: ignore[assignment]
     try:
-        result = TAPTestResult(real_stdout)
+        result = TAPTestResult(real_stdout, real_stderr)
         suite.run(result)
     finally:
-        sys.stdout.flush()
         sys.stdout = real_stdout
-        sys.stderr.flush()
-        sys.stderr = real_stderr
-        stderr_fd_capture.restore()
 
     return result.wasSuccessful()