Unify example stdout checking with CheckOutput

Move the hand-rolled StdOutCapture filtering in add_example_test into
CheckOutput, which now accepts optional expected_patterns for its
allow-list.  This gives example tests and unit tests a single mechanism
for catching unexpected stdout.

Add an allowed_output parameter to add_example_test for patterns that
should not trigger failures but are not required to appear (e.g.
solver stats printed only when matplotlib is missing).

Author: adenzler-nvidia

newton/tests/test_examples.py

@@ -25,11 +25,10 @@
 import newton.viewer
 from newton.tests.unittest_utils import (
     USD_AVAILABLE,
-    StdOutCapture,
+    CheckOutput,
     add_function_test,
     get_selected_cuda_test_devices,
     get_test_devices,
-    is_noise_line,
 )
 
 
@@ -69,6 +68,7 @@ def add_example_test(
     test_suffix: str | None = None,
     expected_output: list[str] | None = None,
     expected_output_cpu: list[str] | None = None,
+    allowed_output: list[str] | None = None,
 ):
     """Registers a Newton example to run on ``devices`` as a TestCase.
 
@@ -79,6 +79,9 @@ def add_example_test(
             matched by at least one pattern fails the test.
         expected_output_cpu: Like *expected_output* but only asserted on
             CPU devices.
+        allowed_output: Regex patterns for stdout lines that are allowed
+            but not required.  These prevent unexpected-output failures
+            without requiring a match.
     """
 
     # verify the module exists (use package-relative path so this works from any CWD)
@@ -152,40 +155,33 @@ def run(test, device):
             else:
                 i += 1
 
-        # Create viewer and run example in-process, capturing warnings + stdout
-        viewer = newton.viewer.ViewerNull(num_frames=args.num_frames)
-        factory = getattr(mod, "create_example", None) or mod.Example
-        capture = StdOutCapture()
-        capture.begin()
-        try:
-            with wp.ScopedDevice(device), warnings.catch_warnings(record=True) as caught:
-                warnings.simplefilter("always")
-                example = factory(viewer, args)
-                newton.examples.run(example, args)
-                wp.synchronize()
-
-            stdout = capture.end()
-        except BaseException:
-            error_stdout = capture.end()
-            if error_stdout.strip():
-                print(f"Captured stdout before crash:\n{error_stdout.rstrip()}")
-            raise
-
-        # Build expected pattern list
+        # Build expected pattern list (used for both warning and stdout checks)
         is_cpu = wp.get_device(device).is_cpu
         expected_patterns = list(expected_output or [])
         if is_cpu:
             expected_patterns.extend(expected_output_cpu or [])
 
+        # CheckOutput uses both expected and allowed patterns for filtering;
+        # only expected_patterns are asserted to appear at least once.
+        all_patterns = expected_patterns + list(allowed_output or [])
         compiled_patterns = [re.compile(p) for p in expected_patterns] if expected_patterns else []
 
-        # Combine warnings and stdout for expected-pattern matching
+        # Run example in-process.
+        # CheckOutput captures stdout and fails on unexpected lines.
+        # warnings.catch_warnings captures Python warnings for separate validation.
+        viewer = newton.viewer.ViewerNull(num_frames=args.num_frames)
+        factory = getattr(mod, "create_example", None) or mod.Example
+        with CheckOutput(test, expected_patterns=all_patterns) as check:
+            with wp.ScopedDevice(device), warnings.catch_warnings(record=True) as caught:
+                warnings.simplefilter("always")
+                example = factory(viewer, args)
+                newton.examples.run(example, args)
+
+        # Assert each expected pattern appears in warnings or stdout
         warning_messages = [str(w.message) for w in caught]
         all_text = "\n".join(warning_messages)
-        if stdout.strip():
-            all_text += "\n" + stdout
-
-        # Assert each expected pattern appears at least once
+        if check.output.strip():
+            all_text += "\n" + check.output
         for pattern in compiled_patterns:
             test.assertRegex(all_text, pattern, f"Expected output pattern not found: {pattern.pattern}")
 
@@ -198,16 +194,6 @@ def run(test, device):
             if not any(p.search(msg) for p in compiled_patterns):
                 test.fail(f"Unexpected warning: {msg}")
 
-        # Fail on unexpected stdout (same strictness as warnings above)
-        if stdout.strip():
-            filtered = [
-                line
-                for line in stdout.splitlines()
-                if not is_noise_line(line) and not any(p.search(line) for p in compiled_patterns)
-            ]
-            if filtered:
-                test.fail("Unexpected stdout:\n" + "\n".join(filtered))
-
     test_name = f"test_{name}_{test_suffix}" if test_suffix else f"test_{name}"
     add_function_test(cls, test_name, run, devices=devices, check_output=False)
 
@@ -713,10 +699,8 @@ class TestMPMExamples(unittest.TestCase):
     name="basic.example_basic_plotting",
     devices=test_devices,
     test_options={"num-frames": 200},
-    expected_output=[
-        "Diagnostics plot saved to|Simulation diagnostics summary",
-        r"^\s+(Iterations|Kinetic E|Potential E|Constraints):",
-    ],
+    expected_output=["Diagnostics plot saved to|Simulation diagnostics summary"],
+    allowed_output=[r"^\s+(Iterations|Kinetic E|Potential E|Constraints):"],
 )
 
 

newton/tests/unittest_utils.py

@@ -207,30 +207,50 @@ def is_noise_line(line: str) -> bool:
 
 
 class CheckOutput:
-    def __init__(self, test):
+    """Context manager that captures stdout and fails on unexpected output.
+
+    Args:
+        test: The ``unittest.TestCase`` instance used for assertions.
+        expected_patterns: Optional regex strings.  Lines matching any pattern
+            are considered expected and will not trigger an unexpected-output
+            failure.  The captured text is available as ``self.output`` after
+            the context exits.
+    """
+
+    def __init__(self, test, expected_patterns: list[str] | None = None):
         self.test = test
+        self.output = ""
+        self._compiled = [re.compile(p) for p in expected_patterns] if expected_patterns else []
 
     def __enter__(self):
-        # wp.force_load()
-
         self.capture = StdOutCapture()
         self.capture.begin()
+        return self
 
     def __exit__(self, exc_type, exc_value, traceback):
         # ensure any stdout output is flushed
         wp.synchronize()
 
-        s = self.capture.end()
-        if s != "":
-            print(s.rstrip())
+        self.output = self.capture.end()
+        if exc_type is not None:
+            # An exception is propagating — print captured output for
+            # debugging but skip the unexpected-output assertion.
+            if self.output.strip():
+                print(f"Captured stdout before crash:\n{self.output.rstrip()}")
+            return
+
+        if self.output != "":
+            print(self.output.rstrip())
 
-            # fail if test produces unexpected output (e.g.: from wp.expect_eq() builtins)
-            # we allow strings starting of the form "Module xxx load on device xxx"
-            # for lazy loaded modules
-            filtered_s = "\n".join(line for line in s.splitlines() if not is_noise_line(line))
+            # Fail on unexpected output — filter noise and expected patterns
+            filtered_s = "\n".join(
+                line
+                for line in self.output.splitlines()
+                if not is_noise_line(line) and not any(p.search(line) for p in self._compiled)
+            )
 
             if filtered_s.strip():
-                self.test.fail(f"Unexpected output:\n'{s.rstrip()}'")
+                self.test.fail(f"Unexpected output:\n'{self.output.rstrip()}'")
 
 
 def assert_array_equal(result: wp.array, expect: wp.array):
@@ -285,12 +305,10 @@ def find_nonfinite_members(obj: Any | None) -> list[str]:
     return nonfinite_members
 
 
-# if check_output is True any output to stdout will be treated as an error
-def create_test_func(func, device, check_output, **kwargs):
-    # pass args to func
+def create_test_func(func, device, check_output, expected_patterns=None, **kwargs):
     def test_func(self):
         if check_output:
-            with CheckOutput(self):
+            with CheckOutput(self, expected_patterns=expected_patterns):
                 func(self, device, **kwargs)
         else:
             func(self, device, **kwargs)
@@ -317,9 +335,9 @@ def sanitize_identifier(s):
         return re.sub(r"\W|^(?=\d)", "_", s)
 
 
-def add_function_test(cls, name, func, devices=None, check_output=True, **kwargs):
+def add_function_test(cls, name, func, devices=None, check_output=True, expected_patterns=None, **kwargs):
     if devices is None:
-        setattr(cls, name, create_test_func(func, None, check_output, **kwargs))
+        setattr(cls, name, create_test_func(func, None, check_output, expected_patterns=expected_patterns, **kwargs))
     elif isinstance(devices, list):
         if not devices:
             # No devices to run this test
@@ -329,13 +347,13 @@ def add_function_test(cls, name, func, devices=None, check_output=True, **kwargs
                 setattr(
                     cls,
                     name + "_" + sanitize_identifier(device),
-                    create_test_func(func, device, check_output, **kwargs),
+                    create_test_func(func, device, check_output, expected_patterns=expected_patterns, **kwargs),
                 )
     else:
         setattr(
             cls,
             name + "_" + sanitize_identifier(devices),
-            create_test_func(func, devices, check_output, **kwargs),
+            create_test_func(func, devices, check_output, expected_patterns=expected_patterns, **kwargs),
         )