diff --git a/tests/nvme_test.py b/tests/nvme_test.py index b34d720243..dc72aaf7ef 100644 --- a/tests/nvme_test.py +++ b/tests/nvme_test.py @@ -77,7 +77,7 @@ def setUp(self): self.do_validate_pci_device = True self.default_nsid = 0x1 self.flbas = 0 - self.config_file = 'tests/config.json' + self.config_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'config.json') self.load_config() if self.do_validate_pci_device: diff --git a/tests/tap_runner.py b/tests/tap_runner.py index e04ac9374a..28a038809c 100644 --- a/tests/tap_runner.py +++ b/tests/tap_runner.py @@ -20,35 +20,42 @@ import unittest -class DiagnosticCapture(io.TextIOBase): - """Capture writes and re-emit them as TAP diagnostic lines (# ...).""" +class TAPDiagnosticStream(io.TextIOBase): + """Wrap a stream and prefix every line with '# ' for TAP diagnostics. - def __init__(self, real_stdout: io.TextIOBase) -> None: - self._real = real_stdout - self._buf = '' + This lets print()/sys.stdout.write() calls from setUp/tearDown/tests + appear on stdout as TAP-compliant diagnostic lines instead of being + mixed into stderr. + """ - 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 __init__(self, stream: io.TextIOBase) -> None: + super().__init__() + self._stream = stream + self._pending = '' + + def write(self, s: str) -> int: + self._pending += s + while '\n' in self._pending: + line, self._pending = self._pending.split('\n', 1) + self._stream.write('# {}\n'.format(line)) + self._stream.flush() + return len(s) def flush(self) -> None: - if self._buf: - self._real.write('# {}\n'.format(self._buf)) - self._buf = '' - self._real.flush() + if self._pending: + self._stream.write('# {}\n'.format(self._pending)) + self._pending = '' + self._stream.flush() class TAPTestResult(unittest.TestResult): """Collect unittest results and render them as TAP version 13.""" - def __init__(self) -> 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 - self._lines: list[str] = [] def _description(self, test: unittest.TestCase) -> str: return '{} ({})'.format(test._testMethodName, type(test).__name__) @@ -56,50 +63,50 @@ def _description(self, test: unittest.TestCase) -> str: def addSuccess(self, test: unittest.TestCase) -> None: super().addSuccess(test) self._test_count += 1 - self._lines.append('ok {} - {}\n'.format( + self._stream.write('ok {} - {}\n'.format( self._test_count, self._description(test))) + self._stream.flush() def addError(self, test: unittest.TestCase, err: object) -> None: super().addError(test, err) self._test_count += 1 - self._lines.append('not ok {} - {}\n'.format( + 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._lines.append('# {}\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._lines.append('not ok {} - {}\n'.format( + 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._lines.append('# {}\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) self._test_count += 1 - self._lines.append('ok {} - {} # SKIP {}\n'.format( + self._stream.write('ok {} - {} # SKIP {}\n'.format( self._test_count, self._description(test), reason)) + self._stream.flush() def addExpectedFailure(self, test: unittest.TestCase, err: object) -> None: super().addExpectedFailure(test, err) self._test_count += 1 - self._lines.append('ok {} - {} # TODO expected failure\n'.format( + self._stream.write('ok {} - {} # TODO expected failure\n'.format( self._test_count, self._description(test))) + self._stream.flush() def addUnexpectedSuccess(self, test: unittest.TestCase) -> None: super().addUnexpectedSuccess(test) self._test_count += 1 - self._lines.append('not ok {} - {} # TODO unexpected success\n'.format( + self._stream.write('not ok {} - {} # TODO unexpected success\n'.format( self._test_count, self._description(test))) - - def print_tap(self, stream: io.TextIOBase) -> None: - stream.write('1..{}\n'.format(self._test_count)) - for line in self._lines: - stream.write(line) - stream.flush() + self._stream.flush() def run_tests(test_module_name: str, start_dir: str | None = None) -> bool: @@ -112,21 +119,23 @@ def run_tests(test_module_name: str, start_dir: str | None = None) -> bool: suite = loader.loadTestsFromModule(module) real_stdout = sys.stdout - # TAP version header must be the very first line on stdout. + real_stderr = sys.stderr + # TAP version header and plan must appear before any test output. real_stdout.write('TAP version 13\n') + real_stdout.write('1..{}\n'.format(suite.countTestCases())) real_stdout.flush() - # Redirect stdout so any print() 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] + # Redirect sys.stdout to a TAP diagnostic stream so that + # print()/sys.stdout.write() calls from setUp/tearDown/tests appear on + # stdout as '# ...' diagnostic lines rather than being sent to stderr. + # Error tracebacks (genuine failures) still go to stderr via diag_stream. + sys.stdout = TAPDiagnosticStream(real_stdout) # type: ignore[assignment] try: - result = TAPTestResult() + result = TAPTestResult(real_stdout, real_stderr) suite.run(result) finally: - sys.stdout.flush() sys.stdout = real_stdout - result.print_tap(real_stdout) return result.wasSuccessful()