feat(test): robust changed-file detection (CI list + self-healing git)

Author: sbryngelson

toolchain/mfc/test/coverage.py

@@ -8,6 +8,7 @@
 import gzip
 import hashlib
 import json
+import subprocess
 from pathlib import Path
 from typing import Optional, Tuple
 
@@ -115,3 +116,39 @@ def select_tests(cases, coverage_map, changed_files):
         else:  # rung 7: skip
             skipped.append(case)
     return to_run, skipped, f"selected {len(to_run)}/{len(cases)} by coverage overlap"
+
+
+def _git(args, cwd, timeout=60):
+    return subprocess.run(["git", *args], capture_output=True, text=True, cwd=cwd, timeout=timeout, check=False)
+
+
+def _merge_base(cwd, branch):
+    for ref in (branch, f"origin/{branch}"):
+        r = _git(["merge-base", ref, "HEAD"], cwd)
+        if r.returncode == 0 and r.stdout.strip():
+            return r.stdout.strip()
+    return None
+
+
+def get_changed_files(root_dir, compare_branch="master", explicit: Optional[str] = None):
+    """Set of changed repo-relative paths, or None if undeterminable (-> run all).
+
+    `explicit` is a newline-separated list from CI (paths-filter); preferred when given.
+    Otherwise use git merge-base, self-healing a shallow clone with a deepen+retry.
+    """
+    if explicit is not None:
+        return {f for f in explicit.splitlines() if f.strip()}
+    try:
+        base = _merge_base(root_dir, compare_branch)
+        if base is None:
+            _git(["fetch", "origin", f"{compare_branch}:{compare_branch}", "--depth=1"], root_dir, 120)
+            _git(["fetch", "--deepen=200"], root_dir, 120)
+            base = _merge_base(root_dir, compare_branch)
+        if base is None:
+            return None
+        diff = _git(["diff", base, "HEAD", "--name-only", "--no-color"], root_dir)
+        if diff.returncode != 0:
+            return None
+        return {f for f in diff.stdout.splitlines() if f.strip()}
+    except (subprocess.TimeoutExpired, OSError):
+        return None

toolchain/mfc/test/test_coverage_unit.py

@@ -1,7 +1,9 @@
 import tempfile
+import types as _types
 from pathlib import Path
+from unittest.mock import patch
 
-from mfc.test.coverage import is_always_run_all, load_map, param_hash, save_map, select_tests
+from mfc.test.coverage import get_changed_files, is_always_run_all, load_map, param_hash, save_map, select_tests
 
 
 def test_param_hash_is_order_independent():
@@ -134,3 +136,35 @@ def test_case_coverage_key_ignores_trace():
     a = TestCase("1D -> Foo", {"m": 100})
     b = TestCase("totally -> different -> trace", {"m": 100})
     assert a.coverage_key() == b.coverage_key()
+
+
+def test_changed_files_prefers_explicit_list():
+    files = get_changed_files("/repo", "master", explicit="src/a.fpp\nsrc/b.fpp\n")
+    assert files == {"src/a.fpp", "src/b.fpp"}
+
+
+def test_changed_files_deepens_then_recovers():
+    state = {"deepened": False}
+
+    def fake_run(cmd, **kw):
+        sub = cmd[1] if len(cmd) > 1 else ""
+        if sub == "fetch":
+            state["deepened"] = True
+            return _types.SimpleNamespace(returncode=0, stdout="", stderr="")
+        if sub == "merge-base":
+            return _types.SimpleNamespace(returncode=0 if state["deepened"] else 1, stdout="base\n", stderr="")
+        if sub == "diff":
+            return _types.SimpleNamespace(returncode=0, stdout="src/x.fpp\n", stderr="")
+        return _types.SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    with patch("subprocess.run", fake_run):
+        assert get_changed_files("/repo", "master") == {"src/x.fpp"}
+
+
+def test_changed_files_returns_none_when_unrecoverable():
+    def fake_run(cmd, **kw):
+        rc = 1 if (len(cmd) > 1 and cmd[1] == "merge-base") else 0
+        return _types.SimpleNamespace(returncode=rc, stdout="", stderr="boom")
+
+    with patch("subprocess.run", fake_run):
+        assert get_changed_files("/repo", "master") is None