feat(ci-artifacts,merge-fix): cluster failed CI jobs and search refs through scripts

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

tests/test_ci_tools.py

@@ -0,0 +1,96 @@
+"""Exercise ci_failures clustering and find_fix's ref search."""
+
+from __future__ import annotations
+
+import importlib.util
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+CI_FAILURES = REPO_ROOT / "skills" / "ci-artifacts" / "scripts" / "ci_failures.py"
+FIND_FIX = REPO_ROOT / "skills" / "merge-fix" / "scripts" / "find_fix.py"
+
+
+def _load(path: Path):
+    spec = importlib.util.spec_from_file_location(path.stem, path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class TestCiFailures(unittest.TestCase):
+    def setUp(self):
+        self.ci = _load(CI_FAILURES)
+
+    def test_jobs_differing_only_in_identifiers_share_a_cluster(self):
+        one = "2026-10-30T10:00:00.1Z setup ok\n##[error]Timeout on host 10.0.0.12 after 300s\n"
+        two = "\x1b[31m2026-10-30T11:00:00.1Z ##[error]Timeout on host 10.0.0.47 after 299s\x1b[0m\n"
+        other = "PLAY RECAP failed=0\nfatal: unable to reach registry\n"
+        groups = self.ci.cluster([("a", one), ("b", two), ("c", other)])
+        self.assertEqual(list(groups.values()), [["a", "b"], ["c"]])
+
+    def test_command_echo_and_retries_are_skipped_and_task_named(self):
+        text = (
+            '##[group]Run make deploy || echo "failed"\n'
+            "make deploy\n"
+            "##[endgroup]\n"
+            "FAILED - RETRYING: wait (3 retries left)\n"
+            "TASK [web-app-y : Probe] ****\n"
+            "Traceback (most recent call last): recovered on retry\n"
+            "TASK [web-app-x : Wait for health] ****\n"
+            'fatal: [localhost]: FAILED! => {"msg": "timeout 120"}\n'
+            "##[error]Process completed with exit code 2.\n"
+        )
+        self.assertEqual(
+            self.ci.signature(text),
+            'TASK [web-app-x : Wait for health] **** fatal: [localhost]: FAILED! => {"msg": "timeout <n>"}',
+        )
+
+    def test_generic_exit_code_line_is_not_the_signature(self):
+        text = "Error: disk full\n##[error]Process completed with exit code 1.\n"
+        self.assertEqual(self.ci.signature(text), "Error: disk full")
+
+
+@unittest.skipUnless(shutil.which("git"), "git not installed")
+class TestFindFix(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.repo = Path(self.tmp.name)
+        self.env = {
+            "PATH": "/usr/bin:/bin",
+            "HOME": str(self.repo),
+            "GIT_AUTHOR_NAME": "t",
+            "GIT_AUTHOR_EMAIL": "t@t",
+            "GIT_COMMITTER_NAME": "t",
+            "GIT_COMMITTER_EMAIL": "t@t",
+        }
+        self._git("init", "-q", "-b", "main")
+        self._git("commit", "-q", "--allow-empty", "-m", "fix: probe in head")
+        self._git("switch", "-q", "-c", "side")
+        self._git("commit", "-q", "--allow-empty", "-m", "fix: probe on side")
+        self._git("switch", "-q", "main")
+
+    def _git(self, *args: str) -> None:
+        subprocess.run(["git", *args], check=True, capture_output=True, env=self.env, cwd=self.repo)
+
+    def test_only_commits_outside_head_are_candidates(self):
+        result = subprocess.run(
+            [sys.executable, str(FIND_FIX), "--grep", "probe"],
+            capture_output=True,
+            text=True,
+            check=True,
+            env=self.env,
+            cwd=self.repo,
+        )
+        self.assertIn("hits=2 in_head=1 candidates=1", result.stdout)
+        self.assertIn("fix: probe on side", result.stdout)
+        self.assertNotIn("probe in head", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()