uutils · kevinburke · Jun 15, 2026
@@ -127,6 +127,31 @@ def analyze_test_results(json_data):
     }
 
 
+def load_or_aggregate_results(json_files):
+    if not json_files:
+        raise ValueError("no JSON input files provided")
+
+    if len(json_files) == 1:
+        try:
+            with open(json_files[0], "r") as file:
+                return json.load(file)
+        except FileNotFoundError:
+            print(f"Error: File '{json_files[0]}' not found.", file=sys.stderr)
+            sys.exit(1)
+        except json.JSONDecodeError:
+            print(
+                f"Error: '{json_files[0]}' is not a valid JSON file.", file=sys.stderr
+            )
+            sys.exit(1)
+
+    return aggregate_results(json_files)
+
+
+def write_results(output_file, json_data):
+    with open(output_file, "w") as f:
+        json.dump(json_data, f, indent=2)
+
+
 def main():
     """
     Main function to process JSON files and export variables.
@@ -147,30 +172,18 @@ def main():
         output_file = json_files[0][3:]
         json_files = json_files[1:]
 
-    # Process the files
-    if len(json_files) == 1:
-        # Single file analysis
-        try:
-            with open(json_files[0], "r") as file:
-                json_data = json.load(file)
-            results = analyze_test_results(json_data)
-        except FileNotFoundError:
-            print(f"Error: File '{json_files[0]}' not found.", file=sys.stderr)
-            sys.exit(1)
-        except json.JSONDecodeError:
-            print(
-                f"Error: '{json_files[0]}' is not a valid JSON file.", file=sys.stderr
-            )
-            sys.exit(1)
-    else:
-        # Multiple files - aggregate them
-        json_data = aggregate_results(json_files)
-        results = analyze_test_results(json_data)
-
-        # Save aggregated data if output file is specified
-        if output_file:
-            with open(output_file, "w") as f:
-                json.dump(json_data, f, indent=2)
+    try:
+        json_data = load_or_aggregate_results(json_files)
+    except ValueError as e:
+        print(f"Error: {e}.", file=sys.stderr)
+        sys.exit(1)
+
+    results = analyze_test_results(json_data)
+
+    # Save aggregated data if output file is specified. For a single input, this
+    # writes the normalized input so downstream workflow steps always have it.
+    if output_file:
+        write_results(output_file, json_data)
 
     # Print export statements for shell evaluation
     print(f"export TOTAL={results['TOTAL']}")

@@ -7,56 +7,115 @@
 import sys
 from pathlib import Path
 
-out = {}
-
-if len(sys.argv) != 2:
-    print("Usage: python gnu-json-result.py <gnu_test_directory>")
-    sys.exit(1)
-
-test_dir = Path(sys.argv[1])
-if not test_dir.is_dir():
-    print(f"Directory {test_dir} does not exist.")
-    sys.exit(1)
-
-# Test all the logs from the test execution
-for filepath in test_dir.glob("**/*.log"):
-    path = Path(filepath)
-    if path.name == "testsuite.log":
-        # Handle Autotest testsuite.log
+STATUS_MAP = {"ok": "PASS", "FAILED": "FAIL", "skipped": "SKIP"}
+
+# GNU tar's Autotest log emits successful/skipped tests as:
+#   4. interspersed options (options02.at:26): ok     (0m0.000s 0m0.004s)
+# and failed tests in the detailed section as:
+#   1. version.at:19: 1. tar version (version.at:19): FAILED (version.at:21)
+AUTOTEST_RESULT_RE = re.compile(
+    r"^\s*(\d+)\.\s+"
+    r"(?:(?:\S+\.at:\d+):\s+)?"
+    r"(?:\d+\.\s+)?"
+    r"(.+?)\s+\(\S+\.at:\d+\):\s+"
+    r"(ok|FAILED|skipped)\b"
+)
+
+# Older Autotest output uses a colon after the test number.
+AUTOTEST_COLON_RESULT_RE = re.compile(
+    r"^\s*(\d+):\s+(.*?)\s+(ok|FAILED|skipped)(?:\s+\(.*\))?$"
+)
+
+AUTOMAKE_RESULT_RE = re.compile(
+    r"(PASS|FAIL|SKIP|ERROR) [^ ]+ \(exit status: \d+\)$"
+)
+
+
+def parse_autotest_line(line):
+    match = AUTOTEST_RESULT_RE.match(line)
+    if not match:
+        match = AUTOTEST_COLON_RESULT_RE.match(line)
+    if not match:
+        return None
+
+    num, name, status = match.groups()
+    return f"test {num}", name.strip(), STATUS_MAP.get(status, status)
+
+
+def parse_autotest_log(path):
+    results = {}
+    with open(path, "r", errors="ignore") as f:
+        for line in f:
+            parsed = parse_autotest_line(line)
+            if parsed:
+                test_group, name, status = parsed
+                results[test_group] = {name: status}
+    return results
+
+
+def parse_automake_log(path):
+    with open(path, "r", errors="ignore") as f:
+        # Only read the end of the file where the result is usually located.
+        f.seek(0, 2)
+        size = f.tell()
+        f.seek(max(0, size - 1000), 0)
+        content = f.read()
+        result = AUTOMAKE_RESULT_RE.search(content)
+        if result:
+            return result.group(1)
+    return None
+
+
+def extract_results(test_dir):
+    out = {}
+
+    # Test all the logs from the test execution.
+    for filepath in test_dir.glob("**/*.log"):
+        path = Path(filepath)
+        if path == test_dir / "testsuite.log":
+            try:
+                out.update(parse_autotest_log(path))
+            except Exception as e:
+                print(f"Error processing testsuite.log {path}: {e}", file=sys.stderr)
+            continue
+
         try:
-            with open(path, "r", errors="ignore") as f:
-                for line in f:
-                    # Look for lines like: "  1: basic functionality                ok"
-                    # or " 10: ...                                     FAILED (basic.at:123)"
-                    match = re.match(r"^\s*(\d+):\s+(.*?)\s+(ok|FAILED|skipped)(?:\s+\(.*\))?$", line)
-                    if match:
-                        num, name, status = match.groups()
-                        # Map Autotest status to Automake-style status
-                        status_map = {"ok": "PASS", "FAILED": "FAIL", "skipped": "SKIP"}
-                        out[f"test {num}"] = {name.strip(): status_map.get(status, status)}
+            result = parse_automake_log(path)
         except Exception as e:
-            print(f"Error processing testsuite.log {path}: {e}", file=sys.stderr)
-        continue
-
-    # Handle individual Automake-style .log files
-    current = out
-    for key in path.parent.relative_to(test_dir).parts:
-        if key not in current:
-            current[key] = {}
-        current = current[key]
-    try:
-        with open(path, "r", errors="ignore") as f:
-            # Only read the end of the file where the result is usually located
-            f.seek(0, 2)
-            size = f.tell()
-            f.seek(max(0, size - 1000), 0)
-            content = f.read()
-            result = re.search(
-                r"(PASS|FAIL|SKIP|ERROR) [^ ]+ \(exit status: \d+\)$", content
-            )
-            if result:
-                current[path.name] = result.group(1)
-    except Exception as e:
-        print(f"Error processing file {path}: {e}", file=sys.stderr)
-
-print(json.dumps(out, indent=2, sort_keys=True))
+            print(f"Error processing file {path}: {e}", file=sys.stderr)
+            continue
+        if not result:
+            continue
+
+        # Handle individual Automake-style .log files.
+        current = out
+        for key in path.parent.relative_to(test_dir).parts:
+            if key not in current:
+                current[key] = {}
+            current = current[key]
+        current[path.name] = result
+
+    return out
+
+
+def main():
+    if len(sys.argv) != 2:
+        print("Usage: python gnu-json-result.py <gnu_test_directory>")
+        return 1
+
+    test_dir = Path(sys.argv[1])
+    if not test_dir.is_dir():
+        print(f"Directory {test_dir} does not exist.", file=sys.stderr)
+        return 1
+
+    out = extract_results(test_dir)
+    if not out:
+        print(f"No GNU test results found in {test_dir}", file=sys.stderr)
+        return 1
+
+    print(json.dumps(out, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,39 @@
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+
+SCRIPT_PATH = Path(__file__).with_name("analyze-gnu-results.py")
+SPEC = importlib.util.spec_from_file_location("analyze_gnu_results", SCRIPT_PATH)
+analyze_gnu_results = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(analyze_gnu_results)
+
+
+class AnalyzeGnuResultsTests(unittest.TestCase):
+    def test_single_input_can_be_written_as_aggregated_output(self):
+        data = {
+            "test 1": {"tar version": "FAIL"},
+            "test 4": {"interspersed options": "PASS"},
+            "test 11": {"--pax-option compatibility": "SKIP"},
+        }
+
+        with tempfile.TemporaryDirectory() as temp:
+            temp_dir = Path(temp)
+            input_file = temp_dir / "gnu-full-result.json"
+            output_file = temp_dir / "aggregated-result.json"
+            input_file.write_text(json.dumps(data))
+
+            results = analyze_gnu_results.load_or_aggregate_results([input_file])
+            analyze_gnu_results.write_results(output_file, results)
+
+            self.assertEqual(json.loads(output_file.read_text()), data)
+
+    def test_load_or_aggregate_requires_input_files(self):
+        with self.assertRaisesRegex(ValueError, "no JSON input files provided"):
+            analyze_gnu_results.load_or_aggregate_results([])
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,61 @@
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+
+
+SCRIPT_PATH = Path(__file__).with_name("gnu-json-result.py")
+SPEC = importlib.util.spec_from_file_location("gnu_json_result", SCRIPT_PATH)
+gnu_json_result = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(gnu_json_result)
+
+
+class GnuJsonResultTests(unittest.TestCase):
+    def test_parse_gnu_tar_autotest_results(self):
+        with tempfile.TemporaryDirectory() as temp:
+            test_dir = Path(temp)
+            (test_dir / "testsuite.log").write_text(
+                "\n".join(
+                    [
+                        "4. interspersed options (options02.at:26): ok     (0m0.000s 0m0.004s)",
+                        "11. --pax-option compatibility (opcomp06.at:21): skipped (opcomp06.at:24)",
+                        "1. version.at:19: 1. tar version (version.at:19): FAILED (version.at:21)",
+                    ]
+                )
+            )
+
+            self.assertEqual(
+                gnu_json_result.extract_results(test_dir),
+                {
+                    "test 1": {"tar version": "FAIL"},
+                    "test 4": {"interspersed options": "PASS"},
+                    "test 11": {"--pax-option compatibility": "SKIP"},
+                },
+            )
+
+    def test_parse_legacy_autotest_result(self):
+        self.assertEqual(
+            gnu_json_result.parse_autotest_line(
+                "  1: basic functionality                ok"
+            ),
+            ("test 1", "basic functionality", "PASS"),
+        )
+
+    def test_ignores_nested_testsuite_log_without_result(self):
+        with tempfile.TemporaryDirectory() as temp:
+            test_dir = Path(temp)
+            (test_dir / "testsuite.log").write_text(
+                "4. interspersed options (options02.at:26): ok     (0m0.000s 0m0.004s)"
+            )
+            nested = test_dir / "testsuite.dir" / "004"
+            nested.mkdir(parents=True)
+            (nested / "testsuite.log").write_text("test detail without trailer\n")
+
+            self.assertEqual(
+                gnu_json_result.extract_results(test_dir),
+                {"test 4": {"interspersed options": "PASS"}},
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()