Improve Pyxsim tester output

CHANGELOG.rst

@@ -9,9 +9,19 @@ UNRELEASED
   * ADDED:     Methods in Xsi class for getting the xsim tick frequency
   * CHANGED:   Pyxsim CMake build uses XCommon CMake
   * CHANGED:   The way time is incremented by time_step for better floating point precision
+  * CHANGED:   ComparisonChecker only prints expected output when verbosity is 2 or higher (i.e.
+    -vv)
+  * CHANGED:   Pyxsim prints captured simulator output when verbosity is enabled while still
+    preserving output capture for tester comparisons
+  * CHANGED:   ComparisonChecker verbose output uses colour to highlight expected and missing
+    output
+  * CHANGED:   ComparisonChecker filters suppressed output from verbose Pyxsim logs and reports
+    colourised suppression counts for multidrive and ignored lines
   * FIXED:     Resolved issues with stdout/stderr capture in Pyxsim
   * FIXED:     Subprocess exit code checking in Pyxsim to properly report errors from
     failed commands
+  * FIXED:     Pyxsim now joins and terminates simulator/subprocess workers on timeout to avoid
+    leaking child processes
 
 2.0.0
 -----
@@ -27,4 +37,3 @@ UNRELEASED
 -----
 
   * Initial release
-

lib/python/Pyxsim/__init__.py

@@ -103,6 +103,7 @@ def run_on_simulator_(xe, tester=None, simthreads=[], **kwargs):
 
     do_xe_prebuild = kwargs.pop("do_xe_prebuild", False)
     capfd = kwargs.pop("capfd", None)
+    verbosity = kwargs.pop("verbosity", 0)
 
     if do_xe_prebuild:
         build_env = kwargs.pop("build_env", {})
@@ -124,6 +125,12 @@ def run_on_simulator_(xe, tester=None, simthreads=[], **kwargs):
         if not build_success:
             return False
 
+    if capfd:
+        pre_stdout, pre_stderr = capfd.readouterr()
+        with capfd.disabled():
+            sys.stdout.write(pre_stdout)
+            sys.stderr.write(pre_stderr)
+
     sim_success = run_with_pyxsim(xe, simthreads, **kwargs)
 
     if not sim_success:
@@ -133,9 +140,29 @@ def run_on_simulator_(xe, tester=None, simthreads=[], **kwargs):
         cap_output, err = capfd.readouterr()
         output = cap_output.split("\n")
         output = [x.strip() for x in output if x != ""]
+        if verbosity > 0:
+            live_output = output
+            summary_lines = []
+            if hasattr(tester, "filter_output"):
+                live_output, suppressed = tester.filter_output(output)
+                if hasattr(tester, "format_suppression_summary"):
+                    summary_lines = tester.format_suppression_summary(suppressed)
+
+            with capfd.disabled():
+                for line in live_output:
+                    sys.stdout.write(line + "\n")
+                for line in summary_lines:
+                    sys.stdout.write(line + "\n")
+                sys.stderr.write(err)
         result = tester.run(output)
         return result
 
+    if verbosity > 0 and capfd:
+        cap_output, err = capfd.readouterr()
+        with capfd.disabled():
+            sys.stdout.write(cap_output)
+            sys.stderr.write(err)
+
     return True
 
 

lib/python/Pyxsim/testers.py

@@ -3,6 +3,7 @@
 import re
 import sys
 from typing import Optional, Sequence, Union
+from colorama import Fore, Style, init
 
 
 class TestError(Exception):
@@ -50,6 +51,7 @@ def __init__(
         verbosity=0,
         suppress_multidrive_messages=True,
     ):
+        init(autoreset=False, strip=False)  # Initialize colorama, force colors even when not TTY
         self._golden = golden
         self._regexp = regexp
         self._ignore = ignore
@@ -67,6 +69,46 @@ def record_failure(self, failure_reason):
         sys.stderr.write("ERROR: %s" % failure_reason)
         self.result = False
 
+    def should_ignore_line(self, line):
+        stripped = line.strip()
+
+        if self._smm and stripped.startswith("Internal control pad and plugin driving in opposite directions"):
+            return "multidrive"
+
+        for p in self._ignore:
+            if re.match(p, stripped):
+                return "ignored"
+
+        return None
+
+    def filter_output(self, output):
+        filtered = []
+        suppressed = {}
+
+        for line in output:
+            reason = self.should_ignore_line(line)
+            if reason:
+                suppressed[reason] = suppressed.get(reason, 0) + 1
+            else:
+                filtered.append(line)
+
+        return filtered, suppressed
+
+    def format_suppression_summary(self, suppressed):
+        lines = []
+
+        if suppressed.get("multidrive"):
+            lines.append(
+                f"{Fore.CYAN}{suppressed['multidrive']} multidrive messages suppressed{Style.RESET_ALL}"
+            )
+
+        if suppressed.get("ignored"):
+            lines.append(
+                f"{Fore.CYAN}{suppressed['ignored']} ignored output lines suppressed{Style.RESET_ALL}"
+            )
+
+        return lines
+
     def run(self, output):
         golden = self._golden
         regexp = self._regexp
@@ -88,17 +130,7 @@ def run(self, output):
         num_expected = len(expected)
 
         for line in output:
-            ignore = False
-            # Check if we should suppress multidrive messages
-            if self._smm and line.strip().startswith("Internal control pad and plugin driving in opposite directions"):
-                ignore = True
-            # Check against user-provided ignore patterns
-            if not ignore:
-                for p in self._ignore:
-                    if re.match(p, line.strip()):
-                        ignore = True
-                        break
-            if ignore:
+            if self.should_ignore_line(line):
                 continue
             line_num += 1
 
@@ -108,8 +140,9 @@ def run(self, output):
                 # Golden file is shorter than output
                 expected_line = "<no line>"
 
+            if self._verbosity > 1:
+                print(f"{Fore.YELLOW}GOLDEN: {expected_line}{Style.RESET_ALL}")
             if self._verbosity > 0:
-                print(f"GOLDEN: {expected_line}")
                 print(f"OUTPUT: {line}")
 
             if line_num >= num_expected:
@@ -130,7 +163,7 @@ def run(self, output):
                     self.record_failure(
                         (
                             "Line %d of output does not match expected\n"
-                            + "  Expected: %s\n"
+                            + f"  {Fore.YELLOW}Expected: %s{Style.RESET_ALL}\n"
                             + "  Actual  : %s"
                         )
                         % (
@@ -148,14 +181,19 @@ def run(self, output):
 
                 if not match:
                     self.record_failure(
-                        ("Line %d of output not found in expected\n" + "  Actual  : %s")
+                        (
+                            "Line %d of output not found in expected\n"
+                            + f"  {Fore.YELLOW}Expected (one of matching lines){Style.RESET_ALL}\n"
+                            + "  Actual  : %s"
+                        )
                         % (line_num, line.strip())
                     )
 
         if num_expected > line_num + 1:
             self.record_failure(
-                "Length of expected output greater than output\nMissing:\n"
+                f"Length of expected output greater than output\n{Fore.RED}Missing:\n"
                 + "\n".join(expected[line_num + 1 :])  # noqa E203
+                + f"{Style.RESET_ALL}"
             )
         output = {"output": "".join(output)}
 

requirements.txt

@@ -18,6 +18,8 @@
 # pip-install this one as editable using this repository's setup.py file.  The
 # same modules should appear in the setup.py list as given below.
 
+colorama==0.4.6
+
 # Development dependencies
 #
 # Each link listed below specifies the path to a setup.py file which are

setup.py

@@ -13,4 +13,7 @@
     name="test_support",
     package_dir={"": "lib/python"},
     packages=setuptools.find_packages(),
+    install_requires=[
+        "colorama>=0.4.6",
+    ],
 )