merge_coverage: canonicalize file paths so aliases collapse

src/slipcover/slipcover.py

@@ -256,8 +256,29 @@ def add_summaries(cov: dict) -> None:
     cov['summary'] = g_summary
 
 
+def _canonical_path(p: str) -> str:
+    """Resolve a path key to a canonical form for cross-step equivalence.
+
+    The same physical file can be recorded under different path spellings
+    across steps of a workload — most commonly a cwd-relative form when one
+    step runs a script directly and an absolute editable-install form when
+    another step imports it as a package. Without canonicalization the merge
+    treats them as separate files and the headline percentage is halved.
+    """
+    try:
+        return str(Path(p).resolve())
+    except OSError:
+        return p
+
+
 def merge_coverage(a: dict, b: dict) -> dict:
-    """Merges coverage result 'b' into 'a'."""
+    """Merges coverage result 'b' into 'a'.
+
+    File entries are grouped by canonical path so that aliases of the same
+    physical file (relative vs absolute, symlinked, etc.) collapse into a
+    single merged entry. The shortest original spelling is used as the
+    display key for each group.
+    """
 
     if a.get('meta', {}).get('software', None) != 'slipcover':
         raise SlipcoverError('Cannot merge coverage: only SlipCover format supported.')
@@ -273,28 +294,53 @@ def merge_coverage(a: dict, b: dict) -> dict:
     a_files = a['files']
     b_files = b['files']
 
-    def both(f, field):
-        return (a_files[f][field] if f in a_files else []) + b_files[f][field]
+    # Group aliases by canonical path.
+    groups: dict = defaultdict(lambda: {'a': [], 'b': []})
+    for k in a_files:
+        groups[_canonical_path(k)]['a'].append(k)
+    for k in b_files:
+        groups[_canonical_path(k)]['b'].append(k)
+
+    new_files: dict = {}
+    for aliases in groups.values():
+        executed_lines: set = set()
+        missing_lines: set = set()
+        executed_branches: set = set()
+        missing_branches: set = set()
+
+        for k in aliases['a']:
+            entry = a_files[k]
+            executed_lines.update(entry.get('executed_lines', ()))
+            missing_lines.update(entry.get('missing_lines', ()))
+            if branch_coverage:
+                executed_branches.update(tuple(br) for br in entry.get('executed_branches', ()))
+                missing_branches.update(tuple(br) for br in entry.get('missing_branches', ()))
+        for k in aliases['b']:
+            entry = b_files[k]
+            executed_lines.update(entry.get('executed_lines', ()))
+            missing_lines.update(entry.get('missing_lines', ()))
+            if branch_coverage:
+                executed_branches.update(tuple(br) for br in entry.get('executed_branches', ()))
+                missing_branches.update(tuple(br) for br in entry.get('missing_branches', ()))
 
-    for f in b_files:
-        executed_lines = set(both(f, 'executed_lines'))
-        missing_lines = set(both(f, 'missing_lines'))
         missing_lines -= executed_lines
-        update = {
+        missing_branches -= executed_branches
+
+        # Prefer the shortest original spelling as the display key (typically
+        # the cwd-relative form when both relative and absolute are present).
+        display = min(aliases['a'] + aliases['b'], key=lambda s: (len(s), s))
+
+        update: dict = {
             'executed_lines': sorted(executed_lines),
-            'missing_lines': sorted(missing_lines)
+            'missing_lines': sorted(missing_lines),
         }
-
         if branch_coverage:
-            executed_branches = set(tuple(br) for br in both(f, 'executed_branches'))
-            missing_branches = set(tuple(br) for br in both(f, 'missing_branches'))
-            missing_branches -= executed_branches
-            update.update({
-                'executed_branches': sorted(list(br) for br in executed_branches),
-                'missing_branches': sorted(list(br) for br in missing_branches)
-            })
-
-        a_files[f] = update
+            update['executed_branches'] = sorted(list(br) for br in executed_branches)
+            update['missing_branches'] = sorted(list(br) for br in missing_branches)
+        new_files[display] = update
+
+    a_files.clear()
+    a_files.update(new_files)
 
     add_summaries(a)
     return a

tests/test_coverage.py

@@ -880,6 +880,70 @@ def test_merge_coverage(tmp_path, monkeypatch, do_branch):
     check_summaries(a)
 
 
+@pytest.mark.parametrize("do_branch", [True, False])
+def test_merge_coverage_canonicalizes_paths(tmp_path, monkeypatch, do_branch):
+    """When the same physical file is recorded under two different path
+    spellings (e.g. relative + absolute, or via a symlink), merge_coverage
+    should canonicalize the paths and collapse the entries into one.
+
+    This happens in practice when one workload step records files via a
+    cwd-relative path (e.g. running `python -m pkg pkg/`) while another step
+    imports the same files as a package, causing Python's import machinery
+    to expose them via their editable-install absolute path.
+    """
+    monkeypatch.chdir(tmp_path)
+
+    (tmp_path / "t.py").write_text("""\
+import sys
+if len(sys.argv) < 2:   # 2
+    print("A branch")
+else:
+    print("B branch")   # 5
+""")
+
+    subprocess.run([sys.executable, '-m', 'slipcover'] +
+                   (['--branch'] if do_branch else []) +
+                   ['--json', '--out', tmp_path / "a.json", "t.py"], check=True)
+    subprocess.run([sys.executable, '-m', 'slipcover'] +
+                   (['--branch'] if do_branch else []) +
+                   ['--json', '--out', tmp_path / "b.json", "t.py", "X"], check=True)
+
+    with (tmp_path / "a.json").open() as f:
+        a = json.load(f)
+    with (tmp_path / "b.json").open() as f:
+        b = json.load(f)
+
+    # Re-key b's entry to the absolute path that the same file resolves to.
+    abs_path = str((tmp_path / "t.py").resolve())
+    assert abs_path != "t.py"
+    b['files'][abs_path] = b['files'].pop('t.py')
+
+    assert 't.py' in a['files']
+    assert abs_path in b['files']
+
+    sc.merge_coverage(a, b)
+
+    # Aliases collapse into a single entry.
+    t_keys = [k for k in a['files'] if k.endswith('t.py')]
+    assert len(t_keys) == 1, f"expected one entry for t.py, got: {t_keys}"
+
+    # The shorter (relative) display form wins when both forms exist.
+    assert t_keys[0] == 't.py'
+
+    # Coverage from both runs is unioned.
+    # Run a (no extra arg) executes lines 1,2,3; run b (with arg) executes 1,2,5.
+    assert {1, 2, 3, 5} <= set(a['files']['t.py']['executed_lines'])
+    assert [] == a['files']['t.py']['missing_lines']
+
+    if do_branch:
+        # Both branches of the `if` at line 2 should be covered.
+        assert [2, 3] in a['files']['t.py']['executed_branches']
+        assert [2, 5] in a['files']['t.py']['executed_branches']
+        assert [] == a['files']['t.py']['missing_branches']
+
+    check_summaries(a)
+
+
 @pytest.fixture
 def cov_merge_fixture(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)