Handle xemu crashing during pgraph test suite

Dockerfile

@@ -22,20 +22,11 @@ RUN apk add --upgrade --no-cache curl libcurl git
 
 WORKDIR /work
 
-RUN mkdir -p /data/TestNXDKPgraphTests
+RUN mkdir -p /data/TestNxdkPgraphTests
 RUN curl \
     -L https://github.com/abaire/nxdk_pgraph_tests/releases/download/v2026-01-08_17-34-01-989184157/nxdk_pgraph_tests_xiso.iso \
-    --output clean_nxdk_pgraph_tests_xiso.iso
-
-RUN cp /usr/src/nxdk/tools/extract-xiso/build/extract-xiso /bin \
-    && extract-xiso -x clean_nxdk_pgraph_tests_xiso.iso
-COPY test-pgraph/config.json clean_nxdk_pgraph_tests_xiso/nxdk_pgraph_tests_config.json
-RUN extract-xiso -c clean_nxdk_pgraph_tests_xiso nxdk_pgraph_tests_xiso.iso \
-    && mv nxdk_pgraph_tests_xiso.iso /data/TestNXDKPgraphTests/ \
-    ;
-
-RUN git clone --depth 1 https://github.com/abaire/nxdk_pgraph_tests_golden_results.git /data/TestNXDKPgraphTests/nxdk_pgraph_tests_golden_results
-
+    --output /data/TestNxdkPgraphTests/nxdk_pgraph_tests_xiso.iso
+RUN git clone --depth 1 https://github.com/abaire/nxdk_pgraph_tests_golden_results.git /data/TestNxdkPgraphTests/nxdk_pgraph_tests_golden_results
 
 FROM ubuntu:25.10 AS ubuntu-base
 RUN set -xe; \
@@ -95,7 +86,7 @@ RUN apt-get -qy install \
 # Combine test data
 FROM scratch AS data
 COPY --from=test-xbe-data /data /data
-COPY --from=pgraph-data /data/TestNXDKPgraphTests /data/TestNXDKPgraphTests
+COPY --from=pgraph-data /data/TestNxdkPgraphTests /data/TestNxdkPgraphTests
 
 #
 # Build final test container

xemutest/test_nxdk_pgraph_tests.py

@@ -1,25 +1,95 @@
 """Test harness for nxdk_pgraph_tests."""
 
-import logging
+import json
 import shutil
+import re
+import logging
+from dataclasses import dataclass, field
+from typing import NamedTuple
 from pathlib import Path
 
 import test_base
+from pyfatx import Fatx
 
 log = logging.getLogger(__file__)
 
 TIMEOUT_SECONDS = 10 * 60
+STARTING_RE = re.compile(r"^Starting (?P<suite>.*?)::(?P<test>.*)")
+COMPLETED_RE = re.compile(r"Completed '(?P<test>.*?)' in (?P<duration>.*)")
 
 
-class TestNXDKPgraphTests(test_base.TestBase):
-    """Runs the nxdk_pgraph_tests suite and validates output."""
+class PgraphTestId(NamedTuple):
+    suite: str
+    name: str
+
+
+@dataclass
+class PgraphTestSuiteAnalysis:
+    tests_completed: list[PgraphTestId] = field(default_factory=list)
+    tests_failed: list[PgraphTestId] = field(default_factory=list)
+
+
+class NxdkPgraphTestExecutor(test_base.TestBase):
+    """Runs the nxdk_pgraph_tests suite."""
 
     def __init__(
         self,
         test_env: test_base.TestEnvironment,
-        results_path: str | Path,
-        test_data_path: str | Path,
+        results_path: Path,
+        test_data_path: Path,
+        config,
     ) -> None:
+        test_data_path = Path(test_data_path)
+        iso_path = test_data_path / "nxdk_pgraph_tests_xiso.iso"
+        if not iso_path.is_file():
+            msg = f"{iso_path} was not installed with the package. You need to build or download it."
+            raise FileNotFoundError(msg)
+        self.config = config
+
+        super().__init__(
+            test_env, "nxdk_pgraph_tests", results_path, iso_path, TIMEOUT_SECONDS
+        )
+
+    def setup_hdd_files(self, fs: test_base.Fatx):
+        super().setup_hdd_files(fs)  # Releases fs
+
+        log.info("Writing config: %r", self.config)
+        fs_e = Fatx(str(self.hdd_path), drive="e")
+        fs_e.mkdir("/nxdk_pgraph_tests")
+        fs_e.write(
+            "/nxdk_pgraph_tests/nxdk_pgraph_tests_config.json",
+            json.dumps(self.config, indent=2).encode("utf-8"),
+        )
+        del fs_e
+
+    def analyze_results(self):
+        """Check xemu exit status."""
+        if self.xemu_exit_status is None:
+            log.warning("xemu exited due to timeout, results are likely partial")
+        elif self.xemu_exit_status:
+            log.warning(
+                "xemu terminated due to error (%d), results may be partial due to a crash",
+                self.xemu_exit_status,
+            )
+
+
+class TestNxdkPgraphTests(test_base.TestBase):
+    """Exhaustively runs the nxdk_pgraph_tests suite and validates output."""
+
+    test_env: test_base.TestEnvironment
+    results_path: Path
+    test_data_path: Path
+
+    def __init__(
+        self,
+        test_env: test_base.TestEnvironment,
+        results_path: Path,
+        test_data_path: Path,
+    ) -> None:
+        self.test_env = test_env
+        self.results_path = results_path
+        self.test_data_path = test_data_path
+
         test_data_path = Path(test_data_path)
         iso_path = test_data_path / "nxdk_pgraph_tests_xiso.iso"
         if not iso_path.is_file():
@@ -37,15 +107,115 @@ class TestNXDKPgraphTests(test_base.TestBase):
             test_env, "nxdk_pgraph_tests", results_path, iso_path, TIMEOUT_SECONDS
         )
 
+    def run(self):
+        num_iterations = 0
+        tests_completed = []
+        tests_failed = []
+        tests_ran = []
+        should_run = True
+
+        while should_run:
+            results_path = self.results_path / f"iteration_{num_iterations}"
+
+            executor = NxdkPgraphTestExecutor(
+                self.test_env,
+                results_path,
+                self.test_data_path,
+                config=self._build_pgraph_test_config(tests_to_skip=tests_ran),
+            )
+            executor.run()
+
+            progress_analysis = self._analyze_pgraph_progress_log(
+                results_path / "pgraph_progress_log.txt"
+            )
+            tests_completed.extend(progress_analysis.tests_completed)
+            tests_failed.extend(progress_analysis.tests_failed)
+            tests_ran.extend(progress_analysis.tests_completed)
+            tests_ran.extend(progress_analysis.tests_failed)
+
+            num_iterations += 1
+            should_run = bool(
+                progress_analysis.tests_failed or progress_analysis.tests_completed
+            )
+
+        for test in tests_failed:
+            log.error("%s::%s failed", test.suite, test.name)
+
+        self.analyze_results()
+
+    @staticmethod
+    def _build_pgraph_test_config(
+        tests_to_skip: list[PgraphTestId] | None = None,
+    ) -> dict:
+        config = {
+            "settings": {
+                "enable_progress_log": True,
+                "disable_autorun": False,
+                "enable_autorun_immediately": True,
+                "enable_shutdown_on_completion": True,
+                "enable_pgraph_region_diff": False,
+                "skip_tests_by_default": False,
+                "delay_milliseconds_between_tests": 0,
+                "network": {
+                    "enable": False,
+                    "config_automatic": False,
+                    "config_dhcp": False,
+                    "static_ip": "",
+                    "static_netmask": "",
+                    "static_gateway": "",
+                    "static_dns_1": "",
+                    "static_dns_2": "",
+                    "ftp": {
+                        "ftp_ip": "",
+                        "ftp_port": 0,
+                        "ftp_user": "",
+                        "ftp_password": "",
+                        "ftp_timeout_milliseconds": 0,
+                    },
+                },
+                "output_directory_path": "c:/nxdk_pgraph_tests",
+            },
+            "test_suites": {},
+        }
+
+        if tests_to_skip:
+            for test in tests_to_skip:
+                if test.suite not in config["test_suites"]:
+                    config["test_suites"][test.suite] = {}
+                if test.name not in config["test_suites"][test.suite]:
+                    config["test_suites"][test.suite][test.name] = {}
+                config["test_suites"][test.suite][test.name]["skipped"] = True
+
+        return config
+
+    @staticmethod
+    def _analyze_pgraph_progress_log(path: Path) -> PgraphTestSuiteAnalysis:
+        """Analyze the nxdk_pgraph_tests progress log to determine which tests ran."""
+        analysis = PgraphTestSuiteAnalysis()
+        with open(path) as file:
+            test_started: PgraphTestId | None = None
+            for line in file.readlines():
+                line = line.strip()
+                if starting_matches := STARTING_RE.match(line):
+                    assert test_started is None, "Unmatched starting/completed sequence"
+                    suite, test = starting_matches.group("suite", "test")
+                    test_started = PgraphTestId(suite, test)
+                elif completed_matches := COMPLETED_RE.match(line):
+                    test = completed_matches.group("test")
+                    assert (
+                        test_started is not None and test_started.name == test
+                    ), "Unmatched starting/completed sequence"
+                    analysis.tests_completed.append(test_started)
+                    test_started = None
+                else:
+                    log.warning("Unexpected log entry: %s", line)
+        if test_started:
+            log.warning("Test %r was not completed! Assumed crashed.", test_started)
+            analysis.tests_failed.append(test_started)
+        return analysis
+
     def analyze_results(self):
         """Processes the generated image files, diffing against the golden result set."""
-        if self.xemu_exit_status is None:
-            log.warning("xemu exited due to timeout, results are likely partial")
-        elif self.xemu_exit_status:
-            log.warning(
-                "xemu terminated due to error (%d), results may be partial due to a crash",
-                self.xemu_exit_status,
-            )
 
         if not self.test_env.perceptualdiff_enabled:
             log.warning("Missing perceptual diff, skipping result analysis")
@@ -87,8 +257,12 @@ class TestNXDKPgraphTests(test_base.TestBase):
             if not file.endswith(".png"):
                 continue
 
+            path_relative_to_iteration = Path(*root_relative_to_out_path.parts[1:])
+            expected_path = (
+                self.golden_results_path / path_relative_to_iteration / file
+            ).resolve()
+
             relative_file_path = root_relative_to_out_path / file
-            expected_path = (self.golden_results_path / relative_file_path).resolve()
             actual_path = (self.results_out_path / relative_file_path).resolve()
             diff_path = diff_results_dir / relative_file_path
             diff_path.parent.mkdir(parents=True, exist_ok=True)