fix(tests): stop misreading named exceptions and coverage timeouts in the vacuous tooling

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/vacuous_tests/guardrails.py

@@ -51,7 +51,8 @@ def _test_names(source: str) -> FrozenSet[str]:
     return frozenset(
         node.name
         for node in ast.walk(tree)
-        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test_")
+        # `test*`, matching pytest's default python_functions and the inventory
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test")
     )
 
 

tests/vacuous_tests/inventory.py

@@ -39,7 +39,7 @@ import warnings
 from collections import Counter
 from dataclasses import dataclass
 from datetime import date
-from typing import Dict, Iterable, List, Optional, Sequence, Set, Tuple, Union
+from typing import Dict, FrozenSet, Iterable, List, Optional, Sequence, Set, Tuple, Union
 
 REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 TOOL_DIR = os.path.join(REPO_ROOT, "tests", "vacuous_tests")
@@ -76,6 +76,7 @@ ASSERT_CALL_PREFIXES = ("assert_", "assert", "check_", "verify_", "expect_")
 PYTEST_ASSERT_FUNCS = {"raises", "fail", "approx", "warns", "deprecated_call"}
 # Handler bodies made up only of these are swallowing the failure.
 SWALLOWING_CALLS = {"skip", "xfail", "print", "warn", "debug", "info", "warning"}
+ASSERTION_CATCHERS = frozenset({"AssertionError", "Exception", "BaseException"})
 # Attributes that only ever hold what the test itself configured or recorded on a mock.
 MOCK_CONFIG_ATTRS = ("return_value", "side_effect", "call_args", "await_args", "call_args_list")
 
@@ -276,21 +277,32 @@ def _swallowed_assertion(fn: TestFunction) -> Optional[str]:
         if not any(_is_assertive_node(sub) for stmt in node.body for sub in ast.walk(stmt)):
             continue
         for handler in node.handlers:
-            if handler.type is not None and "AssertionError" not in ast.unparse(handler.type):
-                # Only assertion-swallowing matters; `except KeyError: pass`
-                # around an assert is usually deliberate setup tolerance.
-                if not _catches_broad_exception(handler):
-                    continue
+            # `except KeyError: pass` around an assert is deliberate setup
+            # tolerance; only a handler that can eat the AssertionError counts.
+            if not _swallows_assertion_errors(handler):
+                continue
             if _handler_swallows(handler):
                 return f"assert inside try/{ast.unparse(handler.type) if handler.type else 'except'} whose handler swallows the failure (line {handler.lineno})"
     return None
 
 
-def _catches_broad_exception(handler: ast.ExceptHandler) -> bool:
+def _caught_names(handler: ast.ExceptHandler) -> FrozenSet[str]:
+    """The exception names a handler catches, unqualified and tuples flattened.
+
+    Matching the unparsed text instead reads `HTTPException` as broad.
+    """
     if handler.type is None:
-        return True
-    rendered = ast.unparse(handler.type)
-    return "Exception" in rendered or "BaseException" in rendered
+        return frozenset()
+    caught = handler.type.elts if isinstance(handler.type, ast.Tuple) else [handler.type]
+    return frozenset(
+        node.attr if isinstance(node, ast.Attribute) else node.id
+        for node in caught
+        if isinstance(node, (ast.Attribute, ast.Name))
+    )
+
+
+def _swallows_assertion_errors(handler: ast.ExceptHandler) -> bool:
+    return handler.type is None or bool(_caught_names(handler) & ASSERTION_CATCHERS)
 
 
 def _trivial_assert(fn: TestFunction, constant_names: Set[str]) -> Optional[str]:

tests/vacuous_tests/mutation_probe.py

@@ -131,7 +131,8 @@ def run_test(test_id: str, timeout: int, overlay: Optional[str] = None) -> Tuple
     return completed.returncode, (completed.stdout + completed.stderr)[-4000:]
 
 
-def _coverage_of(test_id: str, timeout: int) -> Dict[str, Set[int]]:
+def _coverage_of(test_id: str, timeout: int) -> Optional[Dict[str, Set[int]]]:
+    """Lines of litellm the test executes, or None when the run never finishes."""
     import coverage
 
     with tempfile.TemporaryDirectory() as tmp:
@@ -152,14 +153,17 @@ def _coverage_of(test_id: str, timeout: int) -> Dict[str, Set[int]]:
             "no:cacheprovider",
             f"--timeout={timeout}",
         )
-        subprocess.run(
-            command,
-            cwd=REPO_ROOT,
-            env=_pytest_env(),
-            capture_output=True,
-            text=True,
-            timeout=timeout + 120,
-        )
+        try:
+            subprocess.run(
+                command,
+                cwd=REPO_ROOT,
+                env=_pytest_env(),
+                capture_output=True,
+                text=True,
+                timeout=timeout + 120,
+            )
+        except subprocess.TimeoutExpired:
+            return None
         data = coverage.CoverageData(basename=data_file)
         data.read()
         result: Dict[str, Set[int]] = {}
@@ -173,18 +177,23 @@ def _coverage_of(test_id: str, timeout: int) -> Dict[str, Set[int]]:
         return result
 
 
-def covered_lines(test_id: str, timeout: int) -> Dict[str, List[int]]:
+def covered_lines(test_id: str, timeout: int) -> Optional[Dict[str, List[int]]]:
     """Lines the test itself exercises, with import-time coverage subtracted.
 
     Collecting any test in a directory imports litellm and that directory's
     conftest, which lights up thousands of module-level lines. Those lines are
     covered no matter what the test does, so mutating them measures the import,
     not the test. A no-op test in the same directory gives the floor to subtract.
+
+    None when a coverage run does not finish, which is not the same as a test
+    that covers nothing.
     """
     test_path = test_id.split("::", 1)[0]
     with _noop_test(os.path.dirname(os.path.join(REPO_ROOT, test_path))) as noop_id:
         floor = _coverage_of(noop_id, timeout)
     actual = _coverage_of(test_id, timeout)
+    if floor is None or actual is None:
+        return None
     result: Dict[str, List[int]] = {}
     for path, lines in actual.items():
         own = sorted(lines - floor.get(path, set()))
@@ -431,6 +440,8 @@ def probe(test_id: str, max_mutants: int, max_files: int, timeout: int) -> Probe
         return ProbeReport(test_id, "dead", "test is skipped in this environment, so it can never fail")
 
     coverage_map = covered_lines(test_id, timeout)
+    if coverage_map is None:
+        return ProbeReport(test_id, "inconclusive", "the coverage run did not finish, so nothing was mutated")
     behavioural = {
         path: lines
         for path, lines in ((path, behavioural_lines(path, own)) for path, own in coverage_map.items())

tests/vacuous_tests/test_vacuous_tooling.py

@@ -65,6 +65,39 @@ def test_flags_swallowed_assertion() -> None:
     assert bucket_of(source) == "swallowed_failure"
 
 
+def test_does_not_treat_a_named_exception_as_broad() -> None:
+    source = """
+    def test_thing():
+        try:
+            assert compute() == 3
+        except HTTPException:
+            pass
+    """
+    assert bucket_of(source) is None
+
+
+def test_flags_a_handler_that_catches_assertion_error_by_name() -> None:
+    source = """
+    def test_thing():
+        try:
+            assert compute() == 3
+        except AssertionError:
+            pass
+    """
+    assert bucket_of(source) == "swallowed_failure"
+
+
+def test_flags_a_swallowing_handler_that_catches_a_tuple_including_exception() -> None:
+    source = """
+    def test_thing():
+        try:
+            assert compute() == 3
+        except (KeyError, builtins.Exception):
+            pass
+    """
+    assert bucket_of(source) == "swallowed_failure"
+
+
 def test_flags_unconditional_skip() -> None:
     source = """
     @pytest.mark.skip(reason="broken")
@@ -358,6 +391,14 @@ def test_already_reraising_handlers_produce_no_mutant(tmp_path, monkeypatch) ->
     assert not [m for m in mutation_probe.generate_mutants("litellm/hooks.py", range(1, 6)) if m.swallow]
 
 
+def test_a_coverage_run_that_never_finishes_is_not_read_as_covering_nothing(tmp_path, monkeypatch) -> None:
+    (tmp_path / "tests").mkdir()
+    monkeypatch.setattr(mutation_probe, "REPO_ROOT", str(tmp_path))
+    monkeypatch.setattr(mutation_probe, "_coverage_of", lambda test_id, timeout: None)
+
+    assert mutation_probe.covered_lines("tests/test_sample.py::test_thing", 10) is None
+
+
 def test_area_rotation_moves_on_each_day_and_is_stable_within_one(monkeypatch) -> None:
     monkeypatch.setattr(inventory, "cleared_ids", lambda: frozenset())
     candidates = [
@@ -430,6 +471,11 @@ def test_flake_gate_does_not_excuse_sleep_in_a_patched_test(monkeypatch, tmp_pat
     ]
 
 
+def test_guardrails_count_every_name_pytest_collects() -> None:
+    source = "def testCamelCase():\n    pass\n\n\ndef test_snake():\n    pass\n\n\ndef helper():\n    pass\n"
+    assert guardrails._test_names(source) == frozenset({"testCamelCase", "test_snake"})
+
+
 def test_removals_need_a_citation_each() -> None:
     removed = frozenset({"test_one", "test_two"})
     citations = {"tests/test_sample.py::test_one": "tests/test_sample.py::test_covers_one"}