refactor(merge): polish tested_by linker per code review (#113)

- is_test_path: collapse 7 per-language conditional blocks into a data-driven _TEST_NAME_PATTERNS table; JS/TS infix stays inline - production_candidates: extract _join + module-level _add_unique to drop the nested closure and the repeated trailing-slash idiom - Drop dead _TEST_DIR_SEGMENTS constant and the local _splitext reimplementation; use os.path.splitext - link_tests: drop the impossible-malformed-tags guard, tighten the docstring, change edge description to "Path-based pairing (deterministic)", drop redundant break comment - Trim Step 5b inline block that duplicated the module-level header - Convert file-analyzer Note from blockquote to bold paragraph to match surrounding prompt style Tests: split the strip-edges test from the unrelated-edges-survive test, add empty-input and missing-filePath cases, pin sibling-before- walkup and sibling-before-mirror priority order, drop brittle report text assertion. 36 tests, all passing. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-22 10:58:03 +08:00 · 2026-05-06 22:13:37 +08:00
parent f661a03376
commit ba9eeab5bd
3 changed files with 146 additions and 115 deletions
@@ -214,7 +214,7 @@ Using the script's structural data and file categories, create edges:
 | `exports` | File exports a function or class node you created (only for exported items — use IN ADDITION to `contains`, not instead of it) | `0.8` | `forward` |
 | `depends_on` | File has runtime dependency on another project file (broader than imports -- includes dynamic requires, lazy loads) | `0.6` | `forward` |

-> **Note:** Do NOT emit `tested_by` edges. They are produced deterministically by the merge script (`merge-batch-graphs.py`) based on path conventions, in canonical `production → test` direction. Any `tested_by` edges you emit will be discarded.
+**Note:** Do NOT emit `tested_by` edges. They are produced deterministically by the merge script (`merge-batch-graphs.py`) based on path conventions, in canonical `production → test` direction. Any `tested_by` edges you emit will be discarded.

 #### Edges for non-code files:

@@ -19,6 +19,7 @@ Output:
 """

 import json
+import os
 import re
 import sys
 from collections import Counter
@@ -76,16 +77,26 @@ VALID_COMPLEXITY = {"simple", "moderate", "complex"}
 _JS_TS_EXTS: tuple[str, ...] = (".ts", ".tsx", ".js", ".jsx", ".mjs", ".cjs", ".vue")
 _JS_TS_TEST_EXTS: frozenset[str] = frozenset(_JS_TS_EXTS)

-# Test directory names — if any path segment matches one of these, the file
-# is *located in* a test area. By itself this is not enough to classify the
-# file as a test (helpers/fixtures live there too); we still require a test
-# extension on the basename.
-_TEST_DIR_SEGMENTS: frozenset[str] = frozenset({"__tests__", "tests", "test", "spec"})
-
 # Mirrored production roots — when a test sits under `tests/`, it might be
 # mirroring `src/`, `app/`, `lib/`, or the project root.
 _MIRROR_PRODUCTION_ROOTS: tuple[str, ...] = ("src", "app", "lib", "")

+# Per-extension test-name patterns: ext → (prefix_patterns, suffix_patterns).
+# A basename qualifies as a test if its stem starts with any prefix or ends
+# with any suffix listed for its extension. JS/TS family is handled separately
+# because its `.test`/`.spec` infix sits on the *stem* of a double-extension
+# basename (e.g. `foo.test.ts` has ext `.ts`, stem `foo.test`).
+_TEST_NAME_PATTERNS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = {
+    ".go": ((), ("_test",)),
+    ".py": (("test_",), ("_test",)),
+    ".java": ((), ("Test", "Tests", "IT")),
+    ".kt": ((), ("Test", "Tests")),
+    ".cs": ((), ("Test", "Tests")),
+    ".c": (("test_",), ("_test",)),
+    ".cpp": (("test_",), ("_test",)),
+    ".cc": (("test_",), ("_test",)),
+}
+

 def _num(v: Any) -> float:
    """Coerce a value to float for safe comparison (handles string weights)."""
@@ -225,14 +236,6 @@ def _basename(path: str) -> str:
    return path.rsplit("/", 1)[-1] if "/" in path else path


-def _splitext(name: str) -> tuple[str, str]:
-    """Return (stem, ext) for a basename. Single-extension only."""
-    if "." not in name:
-        return name, ""
-    stem, _, ext = name.rpartition(".")
-    return stem, "." + ext
-
-
 def is_test_path(path: str) -> bool:
    """Return True if `path` looks like a test file by basename convention.

@@ -240,45 +243,20 @@ def is_test_path(path: str) -> bool:
    do NOT carry a recognized test extension are treated as helpers/fixtures
    and classified as non-test (so `__tests__/helpers.ts` is not a test).
    """
-    name = _basename(path)
-    stem, ext = _splitext(name)
+    stem, ext = os.path.splitext(_basename(path))

-    # JS/TS family: *.test.<ext> or *.spec.<ext>
+    # JS/TS family: the test marker is an infix on the stem (foo.test.ts has
+    # stem "foo.test", ext ".ts"), not a prefix/suffix on the stem itself.
    if ext in _JS_TS_TEST_EXTS:
-        # stem may itself end with .test or .spec
-        for infix in (".test", ".spec"):
-            if stem.endswith(infix):
-                return True
+        return stem.endswith(".test") or stem.endswith(".spec")

-    # Go: *_test.go
-    if ext == ".go" and stem.endswith("_test"):
-        return True
-
-    # Python: test_*.py or *_test.py
-    if ext == ".py" and (stem.startswith("test_") or stem.endswith("_test")):
-        return True
-
-    # Java: *Test.java, *Tests.java, *IT.java
-    if ext == ".java" and (
-        stem.endswith("Test") or stem.endswith("Tests") or stem.endswith("IT")
-    ):
-        return True
-
-    # Kotlin: *Test.kt, *Tests.kt
-    if ext == ".kt" and (stem.endswith("Test") or stem.endswith("Tests")):
-        return True
-
-    # C#: *Test.cs, *Tests.cs
-    if ext == ".cs" and (stem.endswith("Test") or stem.endswith("Tests")):
-        return True
-
-    # C/C++: *_test.{c,cpp,cc} or test_*.{c,cpp,cc}
-    if ext in {".c", ".cpp", ".cc"} and (
-        stem.startswith("test_") or stem.endswith("_test")
-    ):
-        return True
-
-    return False
+    patterns = _TEST_NAME_PATTERNS.get(ext)
+    if patterns is None:
+        return False
+    prefixes, suffixes = patterns
+    return any(stem.startswith(p) for p in prefixes) or any(
+        stem.endswith(s) for s in suffixes
+    )


 def _strip_test_infix(stem: str) -> str | None:
@@ -290,14 +268,25 @@ def _strip_test_infix(stem: str) -> str | None:
    return None


+def _join(dir_path: str, name: str) -> str:
+    """Join a (possibly empty) directory path to a basename with a single
+    slash, dropping the slash entirely when there is no directory."""
+    return f"{dir_path}/{name}" if dir_path else name
+
+
+def _add_unique(out: list[str], path: str) -> None:
+    """Append `path` to `out` unless it is empty or already present."""
+    if path and path not in out:
+        out.append(path)
+
+
 def _js_ts_sibling_candidates(dir_path: str, base_stem: str) -> list[str]:
    """Build sibling candidates for a JS/TS family base stem.

    `dir_path` is the parent dir (no trailing slash, may be empty).
    `base_stem` is the stem with the test infix already stripped.
    """
-    prefix = f"{dir_path}/" if dir_path else ""
-    return [f"{prefix}{base_stem}{e}" for e in _JS_TS_EXTS]
+    return [_join(dir_path, f"{base_stem}{e}") for e in _JS_TS_EXTS]


 def production_candidates(test_path: str) -> list[str]:
@@ -308,53 +297,44 @@ def production_candidates(test_path: str) -> list[str]:
    preserving order. Caller should pick the first candidate that resolves
    to a known production node.
    """
-    name = _basename(test_path)
-    stem, ext = _splitext(name)
+    stem, ext = os.path.splitext(_basename(test_path))
    segs = _path_segments(test_path)
    dir_segs = segs[:-1]
    dir_path = "/".join(dir_segs)

    candidates: list[str] = []

-    def add(path: str) -> None:
-        if path and path not in candidates:
-            candidates.append(path)
-
    # ── JS/TS family ──────────────────────────────────────────────────
    if ext in _JS_TS_TEST_EXTS:
        base_stem = _strip_test_infix(stem)
        if base_stem is not None:
            # 1. Sibling de-infix: prefer the same extension as the test, then
            # the rest of the family.
-            sibling_dir = dir_path
-            same_ext = f"{(sibling_dir + '/') if sibling_dir else ''}{base_stem}{ext}"
-            add(same_ext)
-            for c in _js_ts_sibling_candidates(sibling_dir, base_stem):
-                add(c)
+            _add_unique(candidates, _join(dir_path, f"{base_stem}{ext}"))
+            for c in _js_ts_sibling_candidates(dir_path, base_stem):
+                _add_unique(candidates, c)

            # 2. Walk out of __tests__/ — drop the trailing __tests__ segment.
            if dir_segs and dir_segs[-1] == "__tests__":
                parent_dir = "/".join(dir_segs[:-1])
-                add(f"{(parent_dir + '/') if parent_dir else ''}{base_stem}{ext}")
+                _add_unique(candidates, _join(parent_dir, f"{base_stem}{ext}"))
                for c in _js_ts_sibling_candidates(parent_dir, base_stem):
-                    add(c)
+                    _add_unique(candidates, c)

            # 3. Mirrored tree: tests/foo/X.test.ts → src/foo/X.ts (and
            # variants for app/lib/<root>).
            if dir_segs and dir_segs[0] in ("tests", "test", "__tests__"):
-                tail = dir_segs[1:]
-                tail_path = "/".join(tail)
+                tail_path = "/".join(dir_segs[1:])
                for root in _MIRROR_PRODUCTION_ROOTS:
-                    parts = [p for p in (root, tail_path) if p]
-                    new_dir = "/".join(parts)
-                    add(f"{(new_dir + '/') if new_dir else ''}{base_stem}{ext}")
+                    new_dir = "/".join(p for p in (root, tail_path) if p)
+                    _add_unique(candidates, _join(new_dir, f"{base_stem}{ext}"))
                    for c in _js_ts_sibling_candidates(new_dir, base_stem):
-                        add(c)
+                        _add_unique(candidates, c)

    # ── Go ────────────────────────────────────────────────────────────
    elif ext == ".go" and stem.endswith("_test"):
        base_stem = stem[: -len("_test")]
-        add(f"{(dir_path + '/') if dir_path else ''}{base_stem}.go")
+        _add_unique(candidates, _join(dir_path, f"{base_stem}.go"))

    # ── Python ────────────────────────────────────────────────────────
    elif ext == ".py" and (stem.startswith("test_") or stem.endswith("_test")):
@@ -364,16 +344,14 @@ def production_candidates(test_path: str) -> list[str]:
            base_stem = stem[: -len("_test")]

        # Sibling
-        add(f"{(dir_path + '/') if dir_path else ''}{base_stem}.py")
+        _add_unique(candidates, _join(dir_path, f"{base_stem}.py"))

        # Mirrored: tests/foo/test_bar.py → src/foo/bar.py (and variants)
        if dir_segs and dir_segs[0] in ("tests", "test"):
-            tail = dir_segs[1:]
-            tail_path = "/".join(tail)
+            tail_path = "/".join(dir_segs[1:])
            for root in _MIRROR_PRODUCTION_ROOTS:
-                parts = [p for p in (root, tail_path) if p]
-                new_dir = "/".join(parts)
-                add(f"{(new_dir + '/') if new_dir else ''}{base_stem}.py")
+                new_dir = "/".join(p for p in (root, tail_path) if p)
+                _add_unique(candidates, _join(new_dir, f"{base_stem}.py"))

    # ── Java ──────────────────────────────────────────────────────────
    elif ext == ".java":
@@ -387,11 +365,10 @@ def production_candidates(test_path: str) -> list[str]:
                    and dir_segs[1] == "test"
                    and dir_segs[2] == "java"
                ):
-                    new_segs = ["src", "main", "java"] + list(dir_segs[3:])
-                    new_dir = "/".join(new_segs)
-                    add(f"{new_dir}/{base_stem}.java")
+                    new_dir = "/".join(["src", "main", "java"] + list(dir_segs[3:]))
+                    _add_unique(candidates, f"{new_dir}/{base_stem}.java")
                # Sibling fallback
-                add(f"{(dir_path + '/') if dir_path else ''}{base_stem}.java")
+                _add_unique(candidates, _join(dir_path, f"{base_stem}.java"))
                break

    # ── Kotlin ────────────────────────────────────────────────────────
@@ -405,10 +382,9 @@ def production_candidates(test_path: str) -> list[str]:
                    and dir_segs[1] == "test"
                    and dir_segs[2] == "kotlin"
                ):
-                    new_segs = ["src", "main", "kotlin"] + list(dir_segs[3:])
-                    new_dir = "/".join(new_segs)
-                    add(f"{new_dir}/{base_stem}.kt")
-                add(f"{(dir_path + '/') if dir_path else ''}{base_stem}.kt")
+                    new_dir = "/".join(["src", "main", "kotlin"] + list(dir_segs[3:]))
+                    _add_unique(candidates, f"{new_dir}/{base_stem}.kt")
+                _add_unique(candidates, _join(dir_path, f"{base_stem}.kt"))
                break

    # ── C# ────────────────────────────────────────────────────────────
@@ -416,7 +392,7 @@ def production_candidates(test_path: str) -> list[str]:
        for suffix in ("Tests", "Test"):
            if stem.endswith(suffix):
                base_stem = stem[: -len(suffix)]
-                add(f"{(dir_path + '/') if dir_path else ''}{base_stem}.cs")
+                _add_unique(candidates, _join(dir_path, f"{base_stem}.cs"))
                break

    # ── C/C++ ─────────────────────────────────────────────────────────
@@ -428,7 +404,7 @@ def production_candidates(test_path: str) -> list[str]:
        else:
            base_stem = None
        if base_stem is not None:
-            add(f"{(dir_path + '/') if dir_path else ''}{base_stem}{ext}")
+            _add_unique(candidates, _join(dir_path, f"{base_stem}{ext}"))

    return candidates

@@ -451,8 +427,8 @@ def link_tests(
    """Strip LLM-emitted `tested_by` edges, then link production files to
    their tests deterministically.

-    Mutates `nodes_by_id` (adds "tested" tag) and `edges` (in-place strip
-    + append).
+    Mutates node values via `nodes_by_id` (adds `tested` tag) and `edges`
+    (drops LLM `tested_by`, appends deterministic ones).

    Returns (added, dropped, tagged):
      added:   number of deterministic tested_by edges appended
@@ -500,18 +476,14 @@ def link_tests(
                "type": "tested_by",
                "direction": "forward",
                "weight": 0.5,
-                "description": "Linked by path convention",
+                "description": "Path-based pairing (deterministic)",
            })
            added += 1
            tags = prod_node.setdefault("tags", [])
-            if not isinstance(tags, list):
-                # Defensive: replace malformed tags with a fresh list.
-                tags = []
-                prod_node["tags"] = tags
            if "tested" not in tags:
                tags.append("tested")
                tagged += 1
-            break  # one production match per test file
+            break

    return added, dropped, tagged

@@ -600,10 +572,7 @@ def merge_and_normalize(batches: list[dict[str, Any]]) -> tuple[dict[str, Any],
        nodes_by_id[nid] = node

    # ── Step 5b: Deterministic tested_by linker ──────────────────────
-    # Strip every LLM-emitted tested_by edge (direction unreliable across
-    # batches) and replace with canonical `production → test` edges derived
-    # from path conventions. Production files that gain a paired test get
-    # the "tested" tag.
+    # See module-level "Deterministic tested_by linker" section above.
    tested_by_added, tested_by_dropped, tested_by_tagged = link_tests(
        nodes_by_id, all_edges
    )
@@ -199,6 +199,26 @@ class ProductionCandidatesTests(unittest.TestCase):
        cands = mbg.production_candidates("src/test/kotlin/com/foo/BarTest.kt")
        self.assertIn("src/main/kotlin/com/foo/Bar.kt", cands)

+    def test_priority_underscore_tests_sibling_before_walkup(self) -> None:
+        # When a test sits in `src/__tests__/`, the sibling-de-infix path
+        # (same directory) ranks before the walk-out path (parent directory).
+        # This is load-bearing: if a project happens to have both
+        # `src/__tests__/X.ts` and `src/X.ts`, we should pair with the
+        # nearer one.
+        cands = mbg.production_candidates("src/__tests__/X.test.ts")
+        self.assertEqual(cands[0], "src/__tests__/X.ts")
+        self.assertIn("src/X.ts", cands)
+        self.assertLess(cands.index("src/__tests__/X.ts"), cands.index("src/X.ts"))
+
+    def test_priority_mirrored_tree_sibling_before_mirror(self) -> None:
+        # `tests/foo/X.test.ts` sibling path is `tests/foo/X.ts`, which must
+        # rank above the mirrored `src/foo/X.ts` variant. Same rationale:
+        # closer pairing wins.
+        cands = mbg.production_candidates("tests/foo/X.test.ts")
+        self.assertEqual(cands[0], "tests/foo/X.ts")
+        self.assertIn("src/foo/X.ts", cands)
+        self.assertLess(cands.index("tests/foo/X.ts"), cands.index("src/foo/X.ts"))
+

 # ── link_tests (end-to-end) ───────────────────────────────────────────────

@@ -255,14 +275,6 @@ class LinkTestsTests(unittest.TestCase):
                "weight": 0.5,
                "description": "from LLM",
            },
-            # Unrelated edge — should survive untouched
-            {
-                "source": "file:src/foo.ts",
-                "target": "file:src/foo.test.ts",
-                "type": "imports",
-                "direction": "forward",
-                "weight": 0.7,
-            },
        ]

        added, dropped, tagged = mbg.link_tests(nodes_by_id, edges)
@@ -276,9 +288,37 @@ class LinkTestsTests(unittest.TestCase):
        self.assertEqual(tested_by_edges[0]["source"], "file:src/foo.ts")
        self.assertEqual(tested_by_edges[0]["target"], "file:src/foo.test.ts")

-        # Imports edge survives
+    def test_unrelated_edges_survive_strip(self) -> None:
+        nodes_by_id = {
+            "file:src/foo.ts": _file_node("src/foo.ts"),
+            "file:src/foo.test.ts": _file_node("src/foo.test.ts"),
+        }
+        edges: list[dict[str, Any]] = [
+            # LLM tested_by edge that gets stripped
+            {
+                "source": "file:src/foo.test.ts",
+                "target": "file:src/foo.ts",
+                "type": "tested_by",
+                "direction": "forward",
+                "weight": 0.5,
+            },
+            # Unrelated edge — should survive untouched
+            {
+                "source": "file:src/foo.ts",
+                "target": "file:src/foo.test.ts",
+                "type": "imports",
+                "direction": "forward",
+                "weight": 0.7,
+            },
+        ]
+
+        mbg.link_tests(nodes_by_id, edges)
+
        import_edges = [e for e in edges if e["type"] == "imports"]
        self.assertEqual(len(import_edges), 1)
+        self.assertEqual(import_edges[0]["source"], "file:src/foo.ts")
+        self.assertEqual(import_edges[0]["target"], "file:src/foo.test.ts")
+        self.assertEqual(import_edges[0]["weight"], 0.7)

    def test_direction_always_forward_production_to_test(self) -> None:
        nodes_by_id = {
@@ -377,6 +417,32 @@ class LinkTestsTests(unittest.TestCase):
        self.assertEqual(tags.count("tested"), 1)
        self.assertIn("core", tags)

+    def test_empty_input(self) -> None:
+        edges: list[dict[str, Any]] = []
+        added, dropped, tagged = mbg.link_tests({}, edges)
+        self.assertEqual((added, dropped, tagged), (0, 0, 0))
+        self.assertEqual(edges, [])
+
+    def test_node_without_filepath_falls_back_to_id(self) -> None:
+        # A file node with only `id` (no `filePath`) should still pair via
+        # the path embedded in the ID.
+        prod = {"id": "file:src/foo.ts", "type": "file", "name": "foo.ts", "tags": []}
+        test = {
+            "id": "file:src/foo.test.ts",
+            "type": "file",
+            "name": "foo.test.ts",
+            "tags": [],
+        }
+        nodes_by_id = {prod["id"]: prod, test["id"]: test}
+        edges: list[dict[str, Any]] = []
+
+        added, dropped, tagged = mbg.link_tests(nodes_by_id, edges)
+
+        self.assertEqual((added, dropped, tagged), (1, 0, 1))
+        self.assertEqual(edges[0]["source"], "file:src/foo.ts")
+        self.assertEqual(edges[0]["target"], "file:src/foo.test.ts")
+        self.assertIn("tested", prod["tags"])
+

 # ── merge_and_normalize integration ───────────────────────────────────────

@@ -417,7 +483,7 @@ class MergeIntegrationTests(unittest.TestCase):
            ],
        }

-        assembled, report = mbg.merge_and_normalize([batch])
+        assembled, _report = mbg.merge_and_normalize([batch])

        # Output should have exactly one tested_by edge with canonical direction
        tested_by_edges = [e for e in assembled["edges"] if e["type"] == "tested_by"]
@@ -429,10 +495,6 @@ class MergeIntegrationTests(unittest.TestCase):
        prod_node = next(n for n in assembled["nodes"] if n["id"] == "file:src/foo.ts")
        self.assertIn("tested", prod_node["tags"])

-        # Report mentions the linker
-        report_text = "\n".join(report)
-        self.assertIn("tested_by", report_text.lower())
-

 if __name__ == "__main__":
    unittest.main()