fix(coverage): dedupe the per-test map before its row cap and cut whole tests

packages/cli/src/repowise/cli/commands/coverage_cmd.py

@@ -329,23 +329,25 @@ async def _do() -> bool:
                         f"repo tree[/yellow] (e.g. {escape(sample)})."
                     )
             if map_records:
-                written = await save_test_coverage(
+                saved = await save_test_coverage(
                     session,
                     repo_row.id,
                     map_records,
                     source_format=map_format or "coverage.py",
                     ingested_commit_sha=head_sha,
                 )
-                dropped = len(map_records) - written
-                extra = (
-                    f" [yellow]({dropped} dropped: duplicate or over cap)[/yellow]"
-                    if dropped
-                    else ""
-                )
+                extra = f" ({saved.merged} repeated record(s) merged)" if saved.merged else ""
                 console.print(
-                    f"[green]Built the test-to-code map: {written} test->file "
+                    f"[green]Built the test-to-code map: {saved.written} test->file "
                     f"record(s).[/green]{extra}"
                 )
+                if saved.truncated:
+                    sample = ", ".join(saved.dropped_sample)
+                    console.print(
+                        f"[yellow]The map is at its row cap: {len(saved.dropped_tests)} "
+                        f"test(s) left out whole (e.g. {escape(sample)}). Test selection "
+                        "runs every test while the map is truncated.[/yellow]"
+                    )
             if partial:
                 console.print(
                     "[yellow]Stored a fragment (--allow-partial): more than half the "

packages/core/src/repowise/core/analysis/missing_test_signal.py

@@ -63,10 +63,13 @@ class MissingTestReport:
     #: changed lines covered and a covering test file IS in the diff (or the
     #: covering rows are pathless so staleness is unknowable) - no signal.
     covered: list[str] = field(default_factory=list)
-    #: file has ZERO coverage rows - unknown, NOT untested. Never a warning.
+    #: file has ZERO coverage rows, or the map is truncated and a covering
+    #: test may be one it left out - unknown, NOT untested. Never a warning.
     no_data: list[str] = field(default_factory=list)
     #: true when the repo has no per-test map at all (nothing to say honestly).
     map_empty: bool = False
+    #: true when the map is at its row cap with whole tests left out.
+    map_truncated: bool = False
 
     def has_signal(self) -> bool:
         return bool(self.untested_changes or self.stale_test_candidates)
@@ -85,6 +88,7 @@ async def detect_missing_tests(
     whether a covering test file was itself edited (sub-signal 2).
     """
     from repowise.core.persistence.crud import get_test_coverage_summary, tests_covering
+    from repowise.core.persistence.crud.analysis.coverage_map import map_is_truncated
 
     report = MissingTestReport()
     if not changed:
@@ -95,6 +99,9 @@ async def detect_missing_tests(
         # No map ingested: we cannot honestly say anything about coverage.
         report.map_empty = True
         return report
+    # At its row cap the map lacks whole tests, so an absent covering test
+    # proves nothing: such a file is unknown, never untested or stale.
+    report.map_truncated = truncated = map_is_truncated(summary)
 
     changed_files = set(changed.keys())
 
@@ -108,6 +115,9 @@ async def detect_missing_tests(
 
         # The file IS in the map. Do any tests cover the CHANGED lines?
         hit_rows = await tests_covering(session, repository_id, source_file, lines=lines)
+        if not hit_rows and truncated:
+            report.no_data.append(source_file)
+            continue
         if not hit_rows:
             # Covered file, uncovered change: the strong signal.
             covered_here: set[int] = set()
@@ -129,6 +139,9 @@ async def detect_missing_tests(
         touched = [tf for tf in covering_test_files if tf in changed_files]
         if touched:
             report.covered.append(source_file)
+        elif covering_test_files and truncated:
+            # The edited test may be one the cut left out.
+            report.no_data.append(source_file)
         elif covering_test_files:
             # Every covering test file we can name is absent from the diff.
             report.stale_test_candidates.append(

packages/core/src/repowise/core/analysis/test_collection.py

@@ -247,7 +247,7 @@ async def collect(
     lets that narrowing read the modules a conftest takes names from.
     """
     from ..persistence.crud import get_health_metrics, get_test_coverage_summary
-    from ..persistence.crud.analysis.coverage_map import MAX_TEST_COVERAGE_ROWS
+    from ..persistence.crud.analysis.coverage_map import map_is_truncated
 
     out = empty_result(len(change.files) + len(change.deleted))
     if not out["changed_files"]:
@@ -256,7 +256,7 @@ async def collect(
 
     summary = await get_test_coverage_summary(session, repo_id)
     out["map_empty"] = summary.get("pair_count", 0) == 0
-    out["map_truncated"] = summary.get("pair_count", 0) >= MAX_TEST_COVERAGE_ROWS
+    out["map_truncated"] = map_is_truncated(summary)
     out["measured_commit"] = measured = summary.get("ingested_commit_sha")
     out["map_current"] = out["map_empty"] or measured in {change.base, change.head} - {None}
 

packages/core/src/repowise/core/analysis/test_impact.py

@@ -22,6 +22,7 @@
 from repowise.core.analysis.test_reachability import MAX_TESTS_PER_TARGET, rank_reaching
 from repowise.core.analysis.test_selection import expand_test_scopes
 from repowise.core.exclusion import is_excluded
+from repowise.core.persistence.crud.analysis.coverage_map import map_is_truncated
 from repowise.core.persistence.models import CoverageFile, Repository
 
 _BASIS_ORDER = {"measured": 0, "inferred": 1}
@@ -511,6 +512,11 @@ def assemble_test_impact(
             if aggregate_summary
             else "no_per_test_coverage_map"
         )
+    elif map_is_truncated(summary):
+        # Whole tests were cut at the row cap: a file without measured tests
+        # may have had them, so its gaps are withheld rather than reported.
+        coverage_status = "degraded"
+        coverage_reason = "per_test_coverage_map_truncated"
     elif 0 < matched_files < len(changed):
         coverage_status = "partial"
         coverage_reason = "coverage_map_matches_only_part_of_change"

packages/core/src/repowise/core/persistence/crud/analysis/coverage_map.py

@@ -14,20 +14,102 @@
 from __future__ import annotations
 
 import json
+from dataclasses import dataclass, replace
 from typing import Any
 
-from sqlalchemy import select
+from sqlalchemy import delete, insert, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ...models import TestCoverageEntry, _new_uuid
-from .._shared import _BATCH_SIZE
+from ...models import TestCoverageEntry, _new_uuid, _now_utc
 
 # Upper bound on rows persisted per repo, so a pathological O(tests x files)
-# report can't balloon the table unbounded. Excess is dropped loudly (the
-# caller compares ``written`` against the record count). Generous headroom:
-# a 2000-test suite touching 100 files each is 200k pairs.
+# report can't balloon the table unbounded. Generous headroom: a 2000-test
+# suite touching 100 files each is 200k pairs. Whole tests are kept, so a
+# truncated map stores at least this many rows (at most one test's rows more):
+# readers take ``pair_count >= MAX_TEST_COVERAGE_ROWS`` as "truncated" and run
+# everything rather than trust a map with tests missing.
 MAX_TEST_COVERAGE_ROWS = 250_000
 
+#: Rows per bulk insert statement.
+_BULK_ROWS = 5_000
+
+
+def map_is_truncated(summary: dict[str, Any]) -> bool:
+    """Whether a :func:`get_test_coverage_summary` describes a map cut at its row cap.
+
+    A cut keeps whole tests up to and past the cap (:func:`cap_test_map`), so a
+    truncated map stores at least the cap's rows. Its readers take a file's
+    missing tests as unknown then, never as untested.
+    """
+    return summary.get("pair_count", 0) >= MAX_TEST_COVERAGE_ROWS
+
+
+@dataclass(frozen=True)
+class TestMapWrite:
+    """What :func:`save_test_coverage` stored."""
+
+    # Not a pytest test class despite the ``Test`` prefix.
+    __test__ = False
+
+    written: int
+    # ``(test_id, source_file)`` pairs that repeated and were merged.
+    merged: int = 0
+    # Tests left out whole by the row cap, in the stable order they were cut.
+    dropped_tests: tuple[str, ...] = ()
+
+    @property
+    def truncated(self) -> bool:
+        return bool(self.dropped_tests)
+
+    @property
+    def dropped_sample(self) -> tuple[str, ...]:
+        """The first few dropped tests, for a report; the count stays exact."""
+        return self.dropped_tests[:5]
+
+
+def cap_test_map(
+    records: list[Any], max_rows: int = MAX_TEST_COVERAGE_ROWS
+) -> tuple[list[Any], int, tuple[str, ...]]:
+    """``(rows, merged, dropped_tests)``: *records* deduped, then capped by whole tests.
+
+    Pure. A repeated ``(test_id, file_path)`` pair - the table's unique key -
+    is merged, its covered lines unioned, before anything is counted against
+    *max_rows*. Tests are then kept in a stable order (test file, then test
+    id) while fewer than *max_rows* rows are kept; the test that reaches the
+    cap is kept whole, every later one is dropped whole. A partial test would
+    look measured and answer "these are all the tests" for its files.
+    """
+    pairs: dict[tuple[str, str], Any] = {}
+    repeated: dict[tuple[str, str], set[int]] = {}
+    for rec in records:
+        key = (rec.test_id, rec.file_path)
+        if key in pairs:
+            repeated.setdefault(key, set(pairs[key].covered_lines or ())).update(
+                rec.covered_lines or ()
+            )
+        else:
+            pairs[key] = rec
+    for key, lines in repeated.items():
+        pairs[key] = replace(pairs[key], covered_lines=sorted(lines))
+    merged = len(records) - len(pairs)
+    by_test: dict[str, list[Any]] = {}
+    for rec in pairs.values():
+        by_test.setdefault(rec.test_id, []).append(rec)
+    # A test's file sorts by its first named one, whatever order records came in.
+    test_file = {
+        t: min((r.test_file for r in recs if r.test_file), default="")
+        for t, recs in by_test.items()
+    }
+    order = sorted(by_test, key=lambda t: (test_file[t], t))
+    rows: list[Any] = []
+    dropped: list[str] = []
+    for test_id in order:
+        if len(rows) >= max_rows:
+            dropped.append(test_id)
+        else:
+            rows.extend(by_test[test_id])
+    return rows, merged, tuple(dropped)
+
 
 async def save_test_coverage(
     session: AsyncSession,
@@ -37,54 +119,37 @@ async def save_test_coverage(
     source_format: str,
     ingested_commit_sha: str | None = None,
     max_rows: int = MAX_TEST_COVERAGE_ROWS,
-) -> int:
+) -> TestMapWrite:
     """Replace the test-coverage rows for *repository_id* with *records*.
 
     *records* are resolved ``TestCoverage`` dataclasses (canonical
-    ``file_path`` / ``test_file``). Delete-then-insert, capped at *max_rows*;
-    returns the number of rows written so the caller can report truncation.
+    ``file_path`` / ``test_file``), deduped and capped by :func:`cap_test_map`.
+    One bulk delete, then batched bulk inserts.
     """
-    existing = await session.execute(
-        select(TestCoverageEntry).where(TestCoverageEntry.repository_id == repository_id)
+    rows, merged, dropped = cap_test_map(records, max_rows)
+    await session.execute(
+        delete(TestCoverageEntry).where(TestCoverageEntry.repository_id == repository_id)
     )
-    for row in existing.scalars().all():
-        await session.delete(row)
-    await session.flush()
-
-    # Deduplicate on (test_id, source_file) - the unique key - keeping the
-    # first occurrence, so a report that repeats a pair never trips the
-    # constraint mid-batch.
-    seen: set[tuple[str, str]] = set()
-    capped = records[:max_rows]
-    written = 0
-    batch: list[TestCoverageEntry] = []
-    for rec in capped:
-        key = (rec.test_id, rec.file_path)
-        if key in seen:
-            continue
-        seen.add(key)
-        batch.append(
-            TestCoverageEntry(
-                id=_new_uuid(),
-                repository_id=repository_id,
-                test_id=rec.test_id,
-                test_file=rec.test_file,
-                source_file=rec.file_path,
-                covered_lines_json=json.dumps(list(rec.covered_lines or [])),
-                source_format=source_format,
-                ingested_commit_sha=ingested_commit_sha,
-            )
+    at = _now_utc()
+    for i in range(0, len(rows), _BULK_ROWS):
+        await session.execute(
+            insert(TestCoverageEntry.__table__),
+            [
+                {
+                    "id": _new_uuid(),
+                    "repository_id": repository_id,
+                    "test_id": rec.test_id,
+                    "test_file": rec.test_file,
+                    "source_file": rec.file_path,
+                    "covered_lines_json": json.dumps(list(rec.covered_lines or [])),
+                    "source_format": source_format,
+                    "ingested_at": at,
+                    "ingested_commit_sha": ingested_commit_sha,
+                }
+                for rec in rows[i : i + _BULK_ROWS]
+            ],
         )
-        if len(batch) >= _BATCH_SIZE:
-            session.add_all(batch)
-            await session.flush()
-            written += len(batch)
-            batch = []
-    if batch:
-        session.add_all(batch)
-        await session.flush()
-        written += len(batch)
-    return written
+    return TestMapWrite(written=len(rows), merged=merged, dropped_tests=dropped)
 
 
 def _decode_lines(row: TestCoverageEntry) -> list[int]:

tests/unit/cli/test_coverage_cmd.py

@@ -8,6 +8,7 @@
 
 from repowise.cli.commands import coverage_cmd
 from repowise.cli.main import cli
+from repowise.core.persistence.crud.analysis.coverage_map import TestMapWrite
 
 
 @pytest.mark.parametrize(
@@ -371,7 +372,7 @@ def test_coverage_add_dot_coverage_builds_the_map_and_says_why_no_per_file(
 
     async def _save_map(session, repository_id, records, **kwargs):
         mapped.extend(records)
-        return len(records)
+        return TestMapWrite(written=len(records))
 
     monkeypatch.setattr("repowise.core.persistence.crud.save_test_coverage", _save_map)
     _coverage_data(tmp_path / ".coverage", "tests/test_a.py::test_one|run", tmp_path / "src" / "a.py")
@@ -449,7 +450,7 @@ def test_coverage_add_withholds_a_partial_per_test_map(
 
     async def _save_map(session, repository_id, records, **kwargs):
         mapped.extend(records)
-        return len(records)
+        return TestMapWrite(written=len(records))
 
     monkeypatch.setattr("repowise.core.persistence.crud.save_test_coverage", _save_map)
     coverage = pytest.importorskip("coverage")
@@ -479,7 +480,7 @@ def test_coverage_add_maps_a_package_per_test_lcov_like_its_aggregate(
 
     async def _save_map(session, repository_id, records, **kwargs):
         mapped.extend(records)
-        return len(records)
+        return TestMapWrite(written=len(records))
 
     monkeypatch.setattr("repowise.core.persistence.crud.save_test_coverage", _save_map)
     report = tmp_path / "packages" / "web" / "coverage" / "lcov.info"

tests/unit/persistence/test_missing_test_signal.py

@@ -138,3 +138,28 @@ async def test_uncovered_lines_are_narrowed_to_the_change(async_session):
     # this is a stale-test candidate, not an untested change.
     assert report.untested_changes == []
     assert len(report.stale_test_candidates) == 1
+
+
+async def test_a_truncated_map_reads_missing_tests_as_unknown(async_session, monkeypatch):
+    # At the row cap whole tests were left out, so neither an uncovered change
+    # nor an unedited covering test proves anything: both files are unknown.
+    from repowise.core.persistence.crud.analysis import coverage_map
+
+    repo = await _seed(
+        async_session,
+        [
+            _rec("tests/test_foo.py::test_a", "src/foo.py", [1, 2], "tests/test_foo.py"),
+            _rec("tests/test_bar.py::test_b", "src/bar.py", [5], "tests/test_bar.py"),
+        ],
+    )
+    monkeypatch.setattr(coverage_map, "MAX_TEST_COVERAGE_ROWS", 2)
+
+    report = await detect_missing_tests(
+        async_session, repo.id, {"src/foo.py": {99}, "src/bar.py": {5}}
+    )
+
+    assert report.map_truncated is True
+    assert report.untested_changes == []
+    assert report.stale_test_candidates == []
+    assert sorted(report.no_data) == ["src/bar.py", "src/foo.py"]
+    assert report.has_signal() is False