scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-08 22:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

Merge d145aa6bf8 into cdb60af024

Browse source
This commit is contained in:
devin-ai-integration[bot] 2026-08-26 10:12:48 -07:00 • committed by GitHub
commit ced6f76c5f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 4832 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.github/workflows/test-code-quality.yml vendored
View file

@ -137,6 +137,12 @@ jobs:
- name: memory_test
run: uv run --no-sync python ./tests/code_coverage_tests/memory_test.py
- name: vacuous_test_tooling
run: uv run --no-sync pytest tests/vacuous_tests/test_vacuous_tooling.py
- name: vacuous_test_ratchet
run: uv run --no-sync python ./tests/vacuous_tests/inventory.py --check
- name: documentation_test_env_keys
run: uv run --no-sync python ./tests/documentation_tests/test_env_keys.py

87
tests/vacuous_tests/README.md Normal file
View file

@ -0,0 +1,87 @@
# Vacuous test audit
A vacuous test is one that cannot fail when the code it claims to test is broken. It passes CI, adds runtime, and gives false confidence, which is worse than having no test at all
This directory holds the tooling behind two things: a CI ratchet that stops new vacuous tests from landing, and a daily automation that fixes existing ones
## Stage A: candidate inventory
`inventory.py` walks `tests/` and classifies test functions into buckets. It is static, so it only produces *candidates*, never verdicts
| bucket | meaning |
| --- | --- |
| `dead_skip` | unconditionally skipped, so it can never fail |
| `swallowed_failure` | the assertion sits in a `try` whose handler swallows the failure |
| `trivial_assert` | `assert True`, `assert <literal>`, or a value compared with itself |
| `mock_tautology` | both sides of the comparison are values the test configured on a mock |
| `no_assert` | no `assert`, `pytest.raises`/`fail`, or `assert_*` call anywhere in the body |
`no_assert` is the noisiest bucket: a test that asserts by not raising is legitimate. That is what Stage B is for
Commands:
```bash
python tests/vacuous_tests/inventory.py --report
python tests/vacuous_tests/inventory.py --check # CI ratchet
python tests/vacuous_tests/inventory.py --update-baseline # after a cleanup
python tests/vacuous_tests/inventory.py --areas # candidates per area
python tests/vacuous_tests/inventory.py --queue 15 --todays-area # the daily batch
python tests/vacuous_tests/inventory.py --queue 15 --area tests/test_litellm/proxy
```
`--todays-area` picks one area per day from the ranked list, rotating by date. That keeps every PR inside one owner's area and needs no state file, so two runs on the same day cannot disagree about where they are working
`inventory_baseline.json` names every known candidate per file and bucket, not just how many there are. `--check` fails on any candidate it does not already name, so fixing one test does not open a slot for a new vacuous one in the same file. The cost is that renaming or moving a known candidate also fails the check, which is a one-command fix: if the failure is a rename, or a deliberate assert-by-not-raising test whose docstring says so, regenerate the baseline
The scan matches pytest's own collection rules rather than a narrower guess, so `test*` names, not only `test_*`, and tests defined inside module-level `if` or `try` blocks are seen too, while nested helpers and anything under `if __name__ == "__main__"` are not, since pytest never binds those on import. One hole is left on purpose: a new vacuous test that reuses the exact name of a fixed one keys to the same baseline entry and passes, which takes deliberate effort and reads as such in review
## Stage B: does the test actually have teeth
`mutation_probe.py` decides. For one test it runs the test under coverage, subtracts the coverage floor of a no-op test in the same directory (so import-time lines are not counted), mutates only the lines the test itself executed, and re-runs the test against each mutant
```bash
python tests/vacuous_tests/mutation_probe.py "tests/x/test_y.py::test_z" --record
```
Mutants never touch the working tree: each one is written into a temp directory that symlinks the whole repo except the mutated file, and pytest runs with that directory as cwd, so a killed or crashed probe cannot leave mutated source behind
Verdicts: `vacuous` (survived every mutant), `not_vacuous` (a mutant killed it, recorded in `verified_not_vacuous.json` so it is never re-flagged), `dead` (skipped in this environment), `already_failing`, or `inconclusive`
Only a `vacuous` verdict authorizes editing a test
`not_vacuous` is a floor, not a compliment: it means the test notices when the code it covers changes behaviour or starts raising, which for an assert-by-not-raising test is all it ever claimed. Reviewing `verified_not_vacuous.json` is still worthwhile, since a cleared test can be weaker than its name suggests
## Fixing a vacuous test
Refactor first: add the assertion the test's own name and docstring imply, then re-run the probe and confirm the mutant that used to survive now dies. Delete only when the behaviour is provably covered somewhere else, and cite that test id. When neither is possible, leave it alone and record why: a human should look at it
## Not becoming flaky
`flake_gate.py` runs every touched test five times under different hash seeds, runs the owning file as a whole, and statically rejects sleeps, wall-clock reads, unseeded randomness, and unmocked network calls
```bash
python tests/vacuous_tests/flake_gate.py "tests/x/test_y.py::test_z"
```
## Guardrails
`guardrails.py` checks the daily automation's own PRs, so it fails by design on the PR that introduced this directory. It reads the committed diff (`base...HEAD`), not the working tree
It runs against the diff and rejects anything that games the metric: files outside `tests/`, edits to `conftest.py` or CI config, edits to this directory's own logic, test removals without a citation, and assertion counts dropping without tests being removed
```bash
python tests/vacuous_tests/guardrails.py --base origin/litellm_internal_staging
python tests/vacuous_tests/guardrails.py --base origin/litellm_internal_staging --allow-removals removals.json
```
Each removal needs its own entry, keyed by the removed test id:
```json
{"tests/x/test_y.py::test_z": "tests/x/test_y.py::test_z_rejects_bad_input covers this"}
```
## Daily automation
The scheduled run pulls the next batch from `--queue 15 --todays-area`, probes each candidate, fixes only the confirmed ones, clears the rest into `verified_not_vacuous.json`, runs the flake gate and the guardrails, then opens a single PR capped at 15 tests in one area. Anything it cannot fix honestly is reported rather than patched
It stops rather than lowering the bar: if fewer than three candidates survive probing it opens no PR that day, and if three or more of its own PRs are still open it skips the run entirely

190
tests/vacuous_tests/flake_gate.py Normal file
View file

@ -0,0 +1,190 @@
"""Anti-flake gate for tests the vacuous-test audit rewrote.
A test that is vacuous today must not become flaky tomorrow, so every test the
daily run touches has to clear this before its PR opens: repeated runs under
different hash seeds, a full run of the owning file, and a static scan for the
usual sources of nondeterminism.
Usage:
python tests/vacuous_tests/flake_gate.py tests/x/test_y.py::test_z [more ...]
"""
from __future__ import annotations
import argparse
import ast
import os
import subprocess
import sys
from dataclasses import dataclass
from typing import Dict, FrozenSet, List, Optional, Sequence, Tuple, Union
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# (dotted call or attribute, why it is a flake risk)
FLAKY_CALLS: Tuple[Tuple[str, str], ...] = (
("time.sleep", "wall-clock sleep: slow and racy under load"),
("asyncio.sleep", "wall-clock sleep: racy under load, prefer awaiting the real signal"),
("datetime.now", "current time in a test makes it depend on when it runs"),
("datetime.utcnow", "current time in a test makes it depend on when it runs"),
("time.time", "current time in a test makes it depend on when it runs"),
("random.random", "unseeded randomness"),
("random.choice", "unseeded randomness"),
("uuid.uuid4", "unseeded randomness in an assertion is unpredictable"),
("requests.get", "real network call"),
("requests.post", "real network call"),
("httpx.get", "real network call"),
("httpx.post", "real network call"),
("litellm.completion", "hits a live provider unless mocked or replayed"),
("litellm.acompletion", "hits a live provider unless mocked or replayed"),
)
MOCK_MARKERS = ("mock", "patch", "respx", "vcr", "cassette", "monkeypatch", "AsyncMock", "MagicMock")
# Safe when the transport is patched for the call, unlike time and randomness,
# which stay nondeterministic however the test is written.
NETWORK_CALLS = frozenset(
{
"requests.get",
"requests.post",
"httpx.get",
"httpx.post",
"litellm.completion",
"litellm.acompletion",
}
)
@dataclass(frozen=True)
class Finding:
test_id: str
problem: str
def _env(seed: str) -> Dict[str, str]:
return {
**os.environ,
"LITELLM_LOCAL_MODEL_COST_MAP": "True",
"PYTHONHASHSEED": seed,
"PYTHONDONTWRITEBYTECODE": "1",
}
def _run(target: str, seed: str, timeout: int) -> Tuple[int, str]:
command: Sequence[str] = (
sys.executable,
"-m",
"pytest",
target,
"-q",
"--no-header",
"-p",
"no:cacheprovider",
f"--timeout={timeout}",
)
completed = subprocess.run(command, cwd=REPO_ROOT, env=_env(seed), capture_output=True, text=True)
return completed.returncode, (completed.stdout + completed.stderr)[-2000:]
TestFunction = Union[ast.FunctionDef, ast.AsyncFunctionDef]
def _find_test(path: str, name: str) -> Optional[TestFunction]:
with open(os.path.join(REPO_ROOT, path), "r", encoding="utf-8") as handle:
tree = ast.parse(handle.read())
wanted = name.split("::")[-1]
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == wanted:
return node
return None
def _dotted(node: ast.expr) -> str:
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
prefix = _dotted(node.value)
return f"{prefix}.{node.attr}" if prefix else node.attr
return ""
def _mocked(marker_source: str) -> bool:
return any(marker in marker_source for marker in MOCK_MARKERS)
def mocked_lines(node: TestFunction) -> FrozenSet[int]:
"""Lines where a patch is in force, either from a decorator or an enclosing with.
A mock anywhere in the test body is not enough: a test can patch one client
and still call a live provider two lines later.
"""
if any(_mocked(ast.unparse(decorator)) for decorator in node.decorator_list):
return frozenset(range(node.lineno, (node.end_lineno or node.lineno) + 1))
return frozenset(
line
for statement in ast.walk(node)
if isinstance(statement, (ast.With, ast.AsyncWith))
and any(_mocked(ast.unparse(item.context_expr)) for item in statement.items)
for line in range(statement.lineno, (statement.end_lineno or statement.lineno) + 1)
)
def static_findings(test_id: str) -> List[Finding]:
path, _, name = test_id.partition("::")
node = _find_test(path, name)
if node is None:
return [Finding(test_id, "test not found in file")]
patched = mocked_lines(node)
reasons: Dict[str, str] = dict(FLAKY_CALLS)
findings: List[Finding] = []
for call in ast.walk(node):
if not isinstance(call, ast.Call):
continue
dotted = _dotted(call.func)
matched = next((known for known in reasons if dotted == known or dotted.endswith(f".{known}")), None)
if matched is None or (matched in NETWORK_CALLS and call.lineno in patched):
continue
findings.append(Finding(test_id, f"uses `{matched}`: {reasons[matched]}"))
return findings
def dynamic_findings(test_id: str, repeat: int, timeout: int) -> List[Finding]:
findings: List[Finding] = []
for index in range(repeat):
seed = str(index * 7919 + 1)
code, output = _run(test_id, seed, timeout)
if code != 0:
findings.append(
Finding(test_id, f"failed on repeat {index + 1}/{repeat} with PYTHONHASHSEED={seed}:\n{output[-600:]}")
)
break
path = test_id.partition("::")[0]
code, output = _run(path, "0", timeout)
if code != 0:
findings.append(Finding(test_id, f"owning file {path} does not pass as a whole:\n{output[-600:]}"))
return findings
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("test_ids", nargs="+")
parser.add_argument("--repeat", type=int, default=5)
parser.add_argument("--timeout", type=int, default=180)
parser.add_argument("--static-only", action="store_true")
args = parser.parse_args()
findings: List[Finding] = []
for test_id in args.test_ids:
findings.extend(static_findings(test_id))
if not args.static_only:
findings.extend(dynamic_findings(test_id, args.repeat, args.timeout))
if findings:
print("flake gate FAILED")
for finding in findings:
print(f" - {finding.test_id}: {finding.problem}")
return 1
print(f"flake gate OK for {len(args.test_ids)} test(s)")
return 0
if __name__ == "__main__":
sys.exit(main())

137
tests/vacuous_tests/guardrails.py Normal file
View file

@ -0,0 +1,137 @@
"""Guardrails on what the vacuous-test audit is allowed to change.
The failure mode of an automated "make the vacuous-test count go down" loop is
gaming the metric: deleting tests, weakening assertions, or editing production
code until the suite agrees. This runs against a diff and rejects it unless the
change is confined to test bodies.
Rules:
1. Only files under tests/ may change, and never conftest.py, CI config, or
the audit tooling itself.
2. The number of test functions may not drop, unless --allow-removals is
passed with a citations file naming the test that now covers each removal.
3. Assertions may not be net removed from a test file.
Usage:
python tests/vacuous_tests/guardrails.py --base origin/litellm_internal_staging
"""
from __future__ import annotations
import argparse
import ast
import json
import os
import subprocess
import sys
from typing import Dict, FrozenSet, List, Optional
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
FORBIDDEN_NAMES = ("conftest.py", "pytest.ini", "pyproject.toml", "ruff.toml", "Makefile")
FORBIDDEN_PREFIXES = (".github/",)
# The audit's own logic is off limits; its JSON ledgers are exactly what the
# daily run has to update.
TOOLING_DIR = "tests/vacuous_tests/"
def _git(*args: str) -> str:
return subprocess.run(("git", *args), cwd=REPO_ROOT, capture_output=True, text=True, check=True).stdout
def changed_files(base: str) -> List[str]:
output = _git("diff", "--name-only", f"{base}...HEAD")
return [line for line in output.splitlines() if line.strip()]
def _test_names(source: str) -> FrozenSet[str]:
try:
tree = ast.parse(source)
except SyntaxError:
return frozenset()
return frozenset(
node.name
for node in ast.walk(tree)
# `test*`, matching pytest's default python_functions and the inventory
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test")
)
def _assert_count(source: str) -> int:
try:
tree = ast.parse(source)
except SyntaxError:
return 0
return sum(1 for node in ast.walk(tree) if isinstance(node, ast.Assert))
def _blob(ref: str, path: str) -> str:
try:
return _git("show", f"{ref}:{path}")
except subprocess.CalledProcessError:
return ""
def check(base: str, allow_removals: Optional[str]) -> List[str]:
violations: List[str] = []
files = changed_files(base)
if not files:
return ["diff is empty"]
for path in files:
if not path.startswith("tests/"):
violations.append(f"{path}: outside tests/; this automation may only change tests")
if os.path.basename(path) in FORBIDDEN_NAMES or path.startswith(FORBIDDEN_PREFIXES):
violations.append(f"{path}: shared test or CI configuration is off limits")
if path.startswith(TOOLING_DIR) and path.endswith(".py"):
violations.append(f"{path}: the audit may not edit its own detection logic")
citations: Dict[str, str] = {}
if allow_removals:
with open(allow_removals, "r", encoding="utf-8") as handle:
citations = json.load(handle)
for path in files:
if not path.endswith(".py"):
continue
before_source = _blob(base, path)
after_source = _blob("HEAD", path)
before_names = _test_names(before_source)
after_names = _test_names(after_source)
violations.extend(uncited_removals(path, before_names - after_names, citations))
before_asserts = _assert_count(before_source)
after_asserts = _assert_count(after_source)
if after_asserts < before_asserts and len(after_names) >= len(before_names):
violations.append(
f"{path}: assertion count dropped from {before_asserts} to {after_asserts} "
"without removing tests, which looks like weakening"
)
return violations
def uncited_removals(path: str, removed: FrozenSet[str], citations: Dict[str, str]) -> List[str]:
"""Every removed test needs its own citation naming the test that now covers it."""
return [
f'{path}::{name} was removed without a citation; add "{path}::{name}": "<test id that now covers this>"'
for name in sorted(removed)
if not citations.get(f"{path}::{name}", "").strip()
]
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", required=True, help="base ref to diff against")
parser.add_argument("--allow-removals", metavar="PATH", help="JSON map of removed test id -> covering test id")
args = parser.parse_args()
violations = check(args.base, args.allow_removals)
if violations:
print("vacuous-test guardrails FAILED")
for violation in violations:
print(f" - {violation}")
return 1
print("vacuous-test guardrails OK")
return 0
if __name__ == "__main__":
sys.exit(main())

650
tests/vacuous_tests/inventory.py Normal file
View file

@ -0,0 +1,650 @@
"""Stage A of the vacuous-test audit: a static inventory of *candidate* vacuous tests.
A vacuous test is one that cannot fail when the code it claims to test is
broken. That property is not statically decidable, so this script only
produces candidates; `mutation_probe.py` (Stage B) is what actually decides,
by mutating the lines a single test covers and checking whether the test
notices.
Two jobs:
1. Ratchet (CI). `--check` compares the candidates found now against the ones
named in `inventory_baseline.json` and fails on any candidate the baseline
does not already name, so a new vacuous test cannot land by taking the slot
of one that was fixed. Regenerate with `--update-baseline` after a cleanup
or a rename.
2. Queue (automation). `--queue N` prints the next N candidates for the daily
run, skipping anything Stage B has already cleared in
`verified_not_vacuous.json`. `--todays-area` keeps a run inside one area,
rotating by date so each PR stays reviewable by one owner and no state file
is needed.
Usage:
python tests/vacuous_tests/inventory.py --report
python tests/vacuous_tests/inventory.py --check
python tests/vacuous_tests/inventory.py --update-baseline
python tests/vacuous_tests/inventory.py --areas
python tests/vacuous_tests/inventory.py --queue 15 --todays-area
python tests/vacuous_tests/inventory.py --queue 15 --area tests/litellm_utils_tests
"""
from __future__ import annotations
import argparse
import ast
import json
import os
import sys
import warnings
from collections import Counter
from dataclasses import dataclass
from datetime import date
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")
BASELINE_PATH = os.path.join(TOOL_DIR, "inventory_baseline.json")
CLEARED_PATH = os.path.join(TOOL_DIR, "verified_not_vacuous.json")
TESTS_ROOT = os.path.join(REPO_ROOT, "tests")
# Buckets, most to least specific. A test is reported under exactly one bucket
# (the first that matches) so counts stay stable when a test has several
# problems.
BUCKETS = (
"dead_skip",
"swallowed_failure",
"trivial_assert",
"mock_tautology",
"no_assert",
)
# Directories whose tests are out of scope: they talk to live providers, take
# minutes, or measure performance rather than behaviour, so "does a mutant kill
# it" is either unaffordable or meaningless.
EXCLUDED_DIRS = (
"tests/load_tests",
"tests/benchmarks",
"tests/e2e",
"tests/multi_instance_e2e_tests",
"tests/old_proxy_tests",
"tests/pass_through_tests",
"tests/vacuous_tests",
)
MOCK_FACTORIES = {"Mock", "MagicMock", "AsyncMock", "NonCallableMock", "patch"}
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")
@dataclass(frozen=True)
class Candidate:
path: str # repo-relative
lineno: int
name: str # dotted: Class.test_method or test_func
bucket: str
evidence: str
@property
def test_id(self) -> str:
return f"{self.path}::{'::'.join(self.name.split('.'))}"
def to_json(self) -> Dict[str, object]:
return {
"test_id": self.test_id,
"path": self.path,
"lineno": self.lineno,
"name": self.name,
"bucket": self.bucket,
"evidence": self.evidence,
}
TestFunction = Union[ast.FunctionDef, ast.AsyncFunctionDef]
def _decorator_names(node: TestFunction) -> List[str]:
names = []
for dec in node.decorator_list:
target = dec.func if isinstance(dec, ast.Call) else dec
names.append(ast.unparse(target))
return names
def _decorators(node: TestFunction) -> List[ast.expr]:
return list(node.decorator_list)
def _call_name(node: ast.AST) -> Optional[str]:
if not isinstance(node, ast.Call):
return None
func = node.func
if isinstance(func, ast.Name):
return func.id
if isinstance(func, ast.Attribute):
return func.attr
return None
def _mock_expression(value: ast.expr, mock_names: Set[str]) -> bool:
"""True for a mock factory call or an attribute chain rooted at a known mock.
Deliberately narrow: an expression that merely *passes* a mock into real code
(`result = handler(mock_client)`) is not mock-derived, because production code
still decides the outcome.
"""
node = value.value if isinstance(value, ast.Await) else value
if isinstance(node, ast.Call):
return _call_name(node) in MOCK_FACTORIES
if isinstance(node, ast.Name):
return node.id in mock_names
while isinstance(node, ast.Attribute):
node = node.value
return isinstance(node, ast.Name) and node.id in mock_names
def _mock_bound_names(fn: TestFunction) -> Set[str]:
"""Names in `fn` bound to a Mock/patch result, transitively."""
mock_names: Set[str] = set()
def rhs_is_mock(value: ast.expr) -> bool:
return _mock_expression(value, mock_names)
# Two passes so `b = a.child` picks up a mock bound later in source order.
for _ in range(2):
for node in ast.walk(fn):
targets: List[ast.expr] = []
value: Optional[ast.expr] = None
if isinstance(node, ast.Assign):
targets, value = list(node.targets), node.value
elif isinstance(node, ast.AnnAssign) and node.value is not None:
targets, value = [node.target], node.value
elif isinstance(node, (ast.With, ast.AsyncWith)):
for item in node.items:
if item.optional_vars is not None and rhs_is_mock(item.context_expr):
targets.append(item.optional_vars)
value = item.context_expr
if value is None or not rhs_is_mock(value):
continue
for target in targets:
for sub in ast.walk(target):
if isinstance(sub, ast.Name):
mock_names.add(sub.id)
# Fixture/decorator-injected mocks: `@patch(...)` passes the mock in as an
# argument, conventionally named mock_*.
for arg in fn.args.args:
if arg.arg.startswith("mock") or arg.arg.endswith("_mock"):
mock_names.add(arg.arg)
return mock_names
def _local_names(node: ast.AST) -> Set[str]:
return {sub.id for sub in ast.walk(node) if isinstance(sub, ast.Name)}
def _is_assertive_node(node: ast.AST) -> bool:
if isinstance(node, ast.Assert):
return True
name = _call_name(node)
if name is None:
return False
if name in PYTEST_ASSERT_FUNCS:
return True
if name.startswith(ASSERT_CALL_PREFIXES):
return True
return False
def _has_assertion(fn: TestFunction) -> bool:
return any(_is_assertive_node(node) for node in ast.walk(fn))
def _is_truthy_constant(node: ast.expr) -> bool:
if isinstance(node, ast.Constant):
return bool(node.value)
if isinstance(node, (ast.List, ast.Tuple, ast.Set)):
return bool(node.elts)
if isinstance(node, ast.Dict):
return bool(node.keys)
return False
def _constant_locals(fn: TestFunction) -> Set[str]:
"""Names assigned a truthy literal exactly once and never rebound."""
assigned: Dict[str, int] = {}
literal: Set[str] = set()
for node in ast.walk(fn):
if not isinstance(node, ast.Assign) or len(node.targets) != 1:
continue
target = node.targets[0]
if not isinstance(target, ast.Name):
continue
assigned[target.id] = assigned.get(target.id, 0) + 1
if _is_truthy_constant(node.value):
literal.add(target.id)
return {name for name in literal if assigned.get(name) == 1}
def _unconditional_skip(fn: TestFunction, decorators: Iterable[ast.expr]) -> Optional[str]:
for dec in decorators:
target = dec.func if isinstance(dec, ast.Call) else dec
rendered = ast.unparse(target)
if rendered.endswith("mark.skip"):
return f"@{rendered}"
if rendered.endswith("mark.skipif") and isinstance(dec, ast.Call) and dec.args:
if _is_truthy_constant(dec.args[0]):
return f"@{rendered}(<always true>)"
body = [s for s in fn.body if not _is_docstring(s)]
if body and isinstance(body[0], ast.Expr):
name = _call_name(body[0].value)
if name in {"skip", "xfail"} and isinstance(body[0].value, ast.Call):
func = body[0].value.func
if isinstance(func, ast.Attribute) and ast.unparse(func).startswith("pytest."):
return f"unconditional pytest.{name}() as first statement"
return None
def _is_docstring(stmt: ast.stmt) -> bool:
return isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Constant) and isinstance(stmt.value.value, str)
def _handler_swallows(handler: ast.ExceptHandler) -> bool:
"""True when the handler cannot surface the failure it caught."""
body = [s for s in handler.body if not _is_docstring(s)]
if not body:
return True
for stmt in body:
if isinstance(stmt, (ast.Pass, ast.Continue, ast.Break)):
continue
if isinstance(stmt, ast.Return) and stmt.value is None:
continue
if isinstance(stmt, ast.Expr):
name = _call_name(stmt.value)
if name in SWALLOWING_CALLS:
continue
return False
return True
def _swallowed_assertion(fn: TestFunction) -> Optional[str]:
for node in ast.walk(fn):
if not isinstance(node, ast.Try):
continue
if not any(_is_assertive_node(sub) for stmt in node.body for sub in ast.walk(stmt)):
continue
for handler in node.handlers:
# `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 _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 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]:
for node in ast.walk(fn):
if not isinstance(node, ast.Assert):
continue
test = node.test
if _is_truthy_constant(test):
return f"`assert {ast.unparse(test)}` (line {node.lineno})"
if isinstance(test, ast.Name) and test.id in constant_names:
return f"`assert {test.id}` where {test.id} is a literal (line {node.lineno})"
if isinstance(test, ast.Compare) and len(test.comparators) == 1:
if ast.unparse(test.left) == ast.unparse(test.comparators[0]):
return f"`assert {ast.unparse(test)}` compares a value with itself (line {node.lineno})"
return None
def _mock_tautology(fn: TestFunction, mock_names: Set[str]) -> Optional[str]:
if not mock_names:
return None
for node in ast.walk(fn):
if not isinstance(node, ast.Assert):
continue
test = node.test
if not isinstance(test, ast.Compare) or len(test.comparators) != 1:
continue
sides = (test.left, test.comparators[0])
if not all(_mock_expression(side, mock_names) for side in sides):
continue
if not all(any(attr in ast.unparse(side) for attr in MOCK_CONFIG_ATTRS) for side in sides):
continue
if _local_names(test) - mock_names:
continue
return f"`assert {ast.unparse(test)}` compares two mock-derived values (line {node.lineno})"
return None
def _module_level_skip(tree: ast.Module) -> Optional[str]:
for node in tree.body:
targets = node.targets if isinstance(node, ast.Assign) else []
if not any(isinstance(t, ast.Name) and t.id == "pytestmark" for t in targets):
continue
rendered = ast.unparse(node.value)
if "mark.skip" in rendered and "skipif" not in rendered:
return "module-level pytestmark skip"
return None
def _is_main_guard(node: ast.If) -> bool:
test = node.test
return (
isinstance(test, ast.Compare)
and isinstance(test.left, ast.Name)
and test.left.id == "__name__"
and len(test.comparators) == 1
and isinstance(test.comparators[0], ast.Constant)
and test.comparators[0].value == "__main__"
)
def _nested_scopes(child: ast.stmt) -> Tuple[Sequence[ast.stmt], ...]:
if isinstance(child, ast.If):
# Nothing under `if __name__ == "__main__"` runs on import, so pytest never sees it
return (child.orelse,) if _is_main_guard(child) else (child.body, child.orelse)
if isinstance(child, ast.Try):
return (child.body, child.orelse, child.finalbody, *(handler.body for handler in child.handlers))
if isinstance(child, (ast.With, ast.AsyncWith)):
return (child.body,)
if isinstance(child, (ast.For, ast.AsyncFor, ast.While)):
return (child.body, child.orelse)
return ()
def _scope_statements(body: Sequence[ast.stmt]) -> List[ast.stmt]:
"""Every statement pytest sees at this scope.
A `def test_x` guarded by `if` or `try` still binds on the module or class,
so pytest collects it; only another function's body is a different scope.
"""
nested = [stmt for child in body for group in _nested_scopes(child) for stmt in group]
return list(body) + (_scope_statements(nested) if nested else [])
def classify_file(path: str, source: str) -> List[Candidate]:
rel = os.path.relpath(path, REPO_ROOT).replace(os.sep, "/")
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore", SyntaxWarning)
tree = ast.parse(source)
except SyntaxError:
return []
module_skip = _module_level_skip(tree)
candidates: List[Candidate] = []
def visit(node: Union[ast.Module, ast.ClassDef], prefix: str) -> None:
for child in _scope_statements(node.body):
if isinstance(child, ast.ClassDef):
if child.name.startswith("Test"):
visit(child, f"{prefix}{child.name}.")
continue
if not isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
# pytest's default python_functions is `test*`, so `testFoo` counts too
if not child.name.startswith("test"):
continue
found = classify_test(child, module_skip)
if found is not None:
bucket, evidence = found
candidates.append(
Candidate(
path=rel,
lineno=child.lineno,
name=f"{prefix}{child.name}",
bucket=bucket,
evidence=evidence,
)
)
visit(tree, "")
return candidates
def classify_test(fn: TestFunction, module_skip: Optional[str]) -> Optional[Tuple[str, str]]:
decorators = _decorators(fn)
skip = module_skip or _unconditional_skip(fn, decorators)
if skip:
return "dead_skip", skip
swallowed = _swallowed_assertion(fn)
if swallowed:
return "swallowed_failure", swallowed
trivial = _trivial_assert(fn, _constant_locals(fn))
if trivial:
return "trivial_assert", trivial
mock_names = _mock_bound_names(fn)
tautology = _mock_tautology(fn, mock_names)
if tautology:
return "mock_tautology", tautology
if not _has_assertion(fn):
return "no_assert", "no assert, pytest.raises/fail, or assert_* call in the test body"
return None
def iter_test_files(root: str = TESTS_ROOT) -> Iterable[str]:
for dirpath, dirnames, filenames in os.walk(root):
rel_dir = os.path.relpath(dirpath, REPO_ROOT).replace(os.sep, "/")
if any(rel_dir == ex or rel_dir.startswith(ex + "/") for ex in EXCLUDED_DIRS):
dirnames[:] = []
continue
dirnames[:] = sorted(d for d in dirnames if d not in {"__pycache__", ".pytest_cache"})
for filename in sorted(filenames):
if filename.endswith(".py") and (filename.startswith("test_") or filename.endswith("_test.py")):
yield os.path.join(dirpath, filename)
def collect(root: str = TESTS_ROOT) -> List[Candidate]:
candidates: List[Candidate] = []
for path in iter_test_files(root):
with open(path, "r", encoding="utf-8", errors="ignore") as handle:
candidates.extend(classify_file(path, handle.read()))
return sorted(candidates, key=lambda c: (c.path, c.lineno))
def to_identities(candidates: Iterable[Candidate]) -> Dict[str, Dict[str, List[str]]]:
"""Which tests are candidates, per file and bucket.
The baseline records names, not counts, so that a fixed test being replaced
by a newly vacuous one in the same file cannot ride through on an unchanged
count.
"""
grouped: Dict[str, Dict[str, List[str]]] = {}
for candidate in candidates:
grouped.setdefault(candidate.path, {}).setdefault(candidate.bucket, []).append(candidate.name)
return {
path: {bucket: sorted(names) for bucket, names in sorted(buckets.items())}
for path, buckets in sorted(grouped.items())
}
def load_json(path: str, default: object) -> object:
if not os.path.exists(path):
return default
with open(path, "r", encoding="utf-8") as handle:
return json.load(handle)
def cleared_ids() -> Set[str]:
data = load_json(CLEARED_PATH, {})
if isinstance(data, dict):
return set(data.get("test_ids", []))
return set()
def write_baseline(identities: Dict[str, Dict[str, List[str]]]) -> None:
totals: Dict[str, int] = {}
for buckets in identities.values():
for bucket, names in buckets.items():
totals[bucket] = totals.get(bucket, 0) + len(names)
payload = {
"_comment": (
"Ratchet baseline for tests/vacuous_tests/inventory.py. It names every known "
"candidate per file and bucket, and any candidate missing from it fails the "
"check. Regenerate with --update-baseline after a cleanup or a rename."
),
"totals": dict(sorted(totals.items())),
"files": identities,
}
with open(BASELINE_PATH, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2, sort_keys=False)
handle.write("\n")
def regressions(
identities: Dict[str, Dict[str, List[str]]],
baseline_files: Dict[str, Dict[str, List[str]]],
) -> List[str]:
return sorted(
f"{path}::{name} is a new {bucket} candidate"
for path, buckets in identities.items()
for bucket, names in buckets.items()
for name in names
if name not in baseline_files.get(path, {}).get(bucket, [])
)
def check_against_baseline(identities: Dict[str, Dict[str, List[str]]]) -> int:
baseline = load_json(BASELINE_PATH, None)
if baseline is None:
print(
f"ERROR: no baseline at {os.path.relpath(BASELINE_PATH, REPO_ROOT)}; run with --update-baseline",
file=sys.stderr,
)
return 1
base_files: Dict[str, Dict[str, List[str]]] = baseline["files"]
failures = regressions(identities, base_files)
if failures:
print("Vacuous-test ratchet failed. New candidate vacuous tests:\n", file=sys.stderr)
for line in failures:
print(f" - {line}", file=sys.stderr)
print(
"\nEach bucket is explained in tests/vacuous_tests/README.md. Make the new "
"test assert something a mutant can break. A renamed or moved candidate "
"lands here too; if this is a rename, or a deliberate assert-by-not-raising "
"test with a docstring saying so, regenerate the baseline with:\n"
" python tests/vacuous_tests/inventory.py --update-baseline",
file=sys.stderr,
)
return 1
improvements = sum(
1
for path, buckets in base_files.items()
for bucket, names in buckets.items()
for name in names
if name not in identities.get(path, {}).get(bucket, [])
)
print(f"Vacuous-test ratchet OK ({improvements} candidate(s) below baseline).")
return 0
def print_report(candidates: List[Candidate]) -> None:
totals: Dict[str, int] = {bucket: 0 for bucket in BUCKETS}
for candidate in candidates:
totals[candidate.bucket] += 1
print(f"candidate vacuous tests: {len(candidates)} across {len({c.path for c in candidates})} files")
for bucket in BUCKETS:
print(f" {bucket:<20} {totals[bucket]}")
def area_of(path: str) -> str:
parts = path.split("/")
return "/".join(parts[:3]) if len(parts) > 3 else os.path.dirname(path)
def areas(candidates: Sequence[Candidate]) -> Tuple[Tuple[str, int], ...]:
cleared = cleared_ids()
open_candidates = tuple(c for c in candidates if c.test_id not in cleared)
return tuple(
sorted(
Counter(area_of(c.path) for c in open_candidates).items(),
key=lambda item: (-item[1], item[0]),
)
)
def rotated_area(candidates: Sequence[Candidate], day: date) -> Optional[str]:
"""Pick one area per day without storing state, so reviewers get one area per PR."""
ranked = areas(candidates)
return ranked[day.toordinal() % len(ranked)][0] if ranked else None
def print_queue(candidates: List[Candidate], limit: int, area: Optional[str]) -> None:
cleared = cleared_ids()
queue = [c for c in candidates if c.test_id not in cleared and (area is None or c.path.startswith(area))]
# Most-specific buckets first: they are the highest-confidence candidates,
# so the daily run spends its mutation budget where it pays off.
order = {bucket: index for index, bucket in enumerate(BUCKETS)}
queue.sort(key=lambda c: (order[c.bucket], c.path, c.lineno))
print(json.dumps([c.to_json() for c in queue[:limit]], indent=2))
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--check", action="store_true", help="fail if counts grew over baseline")
parser.add_argument("--update-baseline", action="store_true")
parser.add_argument("--report", action="store_true", help="print per-bucket totals")
parser.add_argument("--json", metavar="PATH", help="write the full candidate list")
parser.add_argument("--queue", type=int, metavar="N", help="print the next N candidates")
parser.add_argument("--area", help="restrict --queue to a path prefix")
parser.add_argument("--areas", action="store_true", help="print candidate counts per area")
parser.add_argument(
"--todays-area",
action="store_true",
help="print the area this day's run should take, rotating by date",
)
parser.add_argument("--root", default=TESTS_ROOT, help="tests root to scan")
args = parser.parse_args()
candidates = collect(args.root)
identities = to_identities(candidates)
if args.json:
with open(args.json, "w", encoding="utf-8") as handle:
json.dump([c.to_json() for c in candidates], handle, indent=2)
handle.write("\n")
if args.update_baseline:
write_baseline(identities)
print(f"wrote {os.path.relpath(BASELINE_PATH, REPO_ROOT)}")
today = rotated_area(candidates, date.today())
if args.areas:
for area, count in areas(candidates):
print(f" {area:<50} {count}")
if args.todays_area and not args.queue:
print(today or "")
if args.queue:
print_queue(candidates, args.queue, args.area or (today if args.todays_area else None))
if args.report or not (
args.check or args.update_baseline or args.queue or args.json or args.areas or args.todays_area
):
print_report(candidates)
if args.check:
return check_against_baseline(identities)
return 0
if __name__ == "__main__":
sys.exit(main())

2681
tests/vacuous_tests/inventory_baseline.json Normal file
View file

File diff suppressed because it is too large Load diff

586
tests/vacuous_tests/mutation_probe.py Normal file
View file

@ -0,0 +1,586 @@
"""Stage B of the vacuous-test audit: decide whether one test is actually vacuous.
Runs a single test under coverage to find the production lines it executes,
mutates only those lines, and re-runs the same test against each mutant. A test
that survives every mutant cannot fail when the code it covers is broken, which
is the working definition of vacuous. A single kill proves the test has teeth,
and is recorded in verified_not_vacuous.json so the daily run stops re-flagging
it.
Unlike a whole-folder mutmut run (see .github/workflows/mutation-test.yml) this
is scoped to one test and a handful of mutants, so it finishes in minutes.
Usage:
python tests/vacuous_tests/mutation_probe.py \
"tests/test_litellm/foo/test_bar.py::test_baz" --json report.json
"""
from __future__ import annotations
import argparse
import ast
import json
import os
import subprocess
import sys
import tempfile
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Callable, Dict, Iterable, Iterator, List, Optional, Sequence, Set, Tuple
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")
CLEARED_PATH = os.path.join(TOOL_DIR, "verified_not_vacuous.json")
# Files that are configuration or generated data rather than behaviour: mutating
# them says nothing about whether a test has teeth.
SKIPPED_SOURCES = ("litellm/types/", "litellm/proxy/_types.py", "litellm/litellm_core_utils/model_param_helper.py")
# Below this, "coverage" of a file is incidental (a lazy import, a decorator) rather
# than the test exercising it.
MIN_LINES_PER_FILE = 3
COMPARE_SWAPS: Dict[type, type] = {
ast.Eq: ast.NotEq,
ast.NotEq: ast.Eq,
ast.Lt: ast.GtE,
ast.LtE: ast.Gt,
ast.Gt: ast.LtE,
ast.GtE: ast.Lt,
ast.Is: ast.IsNot,
ast.IsNot: ast.Is,
ast.In: ast.NotIn,
ast.NotIn: ast.In,
}
BINOP_SWAPS: Dict[type, type] = {
ast.Add: ast.Sub,
ast.Sub: ast.Add,
ast.Mult: ast.FloorDiv,
ast.Div: ast.Mult,
}
@dataclass(frozen=True)
class Mutant:
path: str
lineno: int
description: str
source: str
# A test that asserts by not raising can only be killed by a mutant that
# raises, so these go first in the budget.
swallow: bool = False
@dataclass
class MutantResult:
path: str
lineno: int
description: str
outcome: str # killed | survived | timeout | broken
@dataclass
class ProbeReport:
test_id: str
verdict: str
detail: str
covered_files: Dict[str, int] = field(default_factory=dict)
mutants: List[MutantResult] = field(default_factory=list)
def to_json(self) -> Dict[str, object]:
return {
"test_id": self.test_id,
"verdict": self.verdict,
"detail": self.detail,
"covered_files": self.covered_files,
"mutants": [vars(m) for m in self.mutants],
}
def _pytest_env() -> Dict[str, str]:
return {
**os.environ,
"LITELLM_LOCAL_MODEL_COST_MAP": "True",
"PYTHONDONTWRITEBYTECODE": "1",
}
def run_test(test_id: str, timeout: int, overlay: Optional[str] = None) -> Tuple[int, str]:
command: Sequence[str] = (
sys.executable,
"-m",
"pytest",
test_id,
"-q",
"--no-header",
"-p",
"no:cacheprovider",
f"--timeout={timeout}",
)
try:
completed = subprocess.run(
command,
cwd=overlay or REPO_ROOT,
env=_pytest_env(),
capture_output=True,
text=True,
timeout=timeout + 60,
)
except subprocess.TimeoutExpired:
return 124, "pytest wall-clock timeout"
return completed.returncode, (completed.stdout + completed.stderr)[-4000:]
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:
data_file = os.path.join(tmp, ".coverage")
command: Sequence[str] = (
sys.executable,
"-m",
"coverage",
"run",
f"--data-file={data_file}",
"--source=litellm",
"-m",
"pytest",
test_id,
"-q",
"--no-header",
"-p",
"no:cacheprovider",
f"--timeout={timeout}",
)
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]] = {}
for measured in data.measured_files():
rel = os.path.relpath(measured, REPO_ROOT).replace(os.sep, "/")
if not rel.startswith("litellm/") or rel.startswith(SKIPPED_SOURCES):
continue
lines = data.lines(measured) or []
if lines:
result[rel] = set(lines)
return result
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()))
if own:
result[path] = own
return result
@contextmanager
def _noop_test(directory: str) -> Iterator[str]:
handle = tempfile.NamedTemporaryFile(
mode="w", suffix=".py", prefix="test_vacuous_probe_floor_", dir=directory, delete=False
)
try:
handle.write("def test_noop() -> None:\n assert True\n")
handle.close()
rel = os.path.relpath(handle.name, REPO_ROOT).replace(os.sep, "/")
yield f"{rel}::test_noop"
finally:
os.unlink(handle.name)
NodeKey = Tuple[str, int, int]
def _position(node: ast.AST) -> Tuple[int, int]:
if isinstance(node, (ast.expr, ast.stmt, ast.ExceptHandler)):
return (node.lineno, node.col_offset)
return (-1, -1)
def _key(node: ast.AST) -> NodeKey:
lineno, col_offset = _position(node)
return (type(node).__name__, lineno, col_offset)
Mutation = Callable[[ast.AST], ast.AST]
def _mutant_source(original: str, key: NodeKey, mutate: Mutation) -> str:
class Transformer(ast.NodeTransformer):
def visit(self, node: ast.AST) -> ast.AST:
if _key(node) == key:
return mutate(node)
return super().visit(node)
mutated = Transformer().visit(ast.parse(original))
ast.fix_missing_locations(mutated)
return ast.unparse(mutated)
def function_body_lines(tree: ast.Module) -> Set[int]:
"""Line numbers inside a function body.
Module and class level lines run at import, so a mutant there is killed (or
not) by whether the module still imports, which says nothing about the test.
"""
lines: Set[int] = set()
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for statement in node.body:
end = statement.end_lineno or statement.lineno
lines.update(range(statement.lineno, end + 1))
return lines
def behavioural_lines(path: str, own_lines: Iterable[int]) -> List[int]:
with open(os.path.join(REPO_ROOT, path), "r", encoding="utf-8") as handle:
tree = ast.parse(handle.read())
inside = function_body_lines(tree)
return sorted(set(own_lines) & inside)
def imported_modules(test_path: str) -> Set[str]:
with open(os.path.join(REPO_ROOT, test_path), "r", encoding="utf-8") as handle:
tree = ast.parse(handle.read())
modules: Set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
modules.update(alias.name for alias in node.names)
elif isinstance(node, ast.ImportFrom) and node.module:
modules.add(node.module)
modules.update(f"{node.module}.{alias.name}" for alias in node.names)
return {module for module in modules if module.startswith("litellm")}
def _module_name(path: str) -> str:
trimmed = path[: -len(".py")] if path.endswith(".py") else path
return trimmed.replace("/", ".").removesuffix(".__init__")
def _is_under_test(path: str, imports: Set[str]) -> bool:
"""True when the test file imports this module, directly or as a parent package."""
module = _module_name(path)
return any(imported == module or imported.startswith(module + ".") for imported in imports)
def generate_mutants(path: str, lines: Iterable[int]) -> List[Mutant]:
absolute = os.path.join(REPO_ROOT, path)
with open(absolute, "r", encoding="utf-8") as handle:
original = handle.read()
tree = ast.parse(original)
covered = set(lines)
mutants: List[Mutant] = []
def emit(node: ast.AST, description: str, mutate: Mutation, swallow: bool = False) -> None:
lineno, _ = _position(node)
if lineno not in covered:
return
try:
source = _mutant_source(original, _key(node), mutate)
except Exception:
return
mutants.append(
Mutant(path=path, lineno=lineno, description=description, source=source, swallow=swallow)
)
for node in ast.walk(tree):
if isinstance(node, ast.ExceptHandler) and not _bare_reraise(node):
emit(
node,
f"stop swallowing `{_snippet(node.type) if node.type else 'except'}` and re-raise",
lambda n: ast.ExceptHandler(type=n.type, name=n.name, body=[ast.Raise()]),
swallow=True,
)
continue
if isinstance(node, ast.Compare) and len(node.ops) == 1 and type(node.ops[0]) in COMPARE_SWAPS:
swap = COMPARE_SWAPS[type(node.ops[0])]
emit(
node,
f"flip `{type(node.ops[0]).__name__}` to `{swap.__name__}` in `{_snippet(node)}`",
lambda n, swap=swap: ast.Compare(left=n.left, ops=[swap()], comparators=n.comparators),
)
elif isinstance(node, ast.BoolOp):
swap = ast.Or if isinstance(node.op, ast.And) else ast.And
emit(
node,
f"swap `{'and' if isinstance(node.op, ast.And) else 'or'}` in `{_snippet(node)}`",
lambda n, swap=swap: ast.BoolOp(op=swap(), values=n.values),
)
elif isinstance(node, ast.BinOp) and type(node.op) in BINOP_SWAPS:
swap = BINOP_SWAPS[type(node.op)]
emit(
node,
f"swap `{type(node.op).__name__}` for `{swap.__name__}` in `{_snippet(node)}`",
lambda n, swap=swap: ast.BinOp(left=n.left, op=swap(), right=n.right),
)
elif isinstance(node, ast.Constant) and isinstance(node.value, bool):
emit(
node,
f"replace `{node.value}` with `{not node.value}`",
lambda n: ast.Constant(value=not n.value),
)
elif isinstance(node, ast.Constant) and isinstance(node.value, int):
emit(
node,
f"replace `{node.value}` with `{node.value + 1}`",
lambda n: ast.Constant(value=n.value + 1),
)
elif isinstance(node, ast.Constant) and isinstance(node.value, str) and node.value:
emit(
node,
f"replace string `{node.value[:30]}` with a different value",
lambda n: ast.Constant(value=f"litellm_mutant_{n.value}"),
)
elif isinstance(node, ast.Return) and node.value is not None and not _is_none(node.value):
emit(
node,
f"replace `return {_snippet(node.value)}` with `return None`",
lambda n: ast.Return(value=ast.Constant(value=None)),
)
return mutants
def _is_none(node: ast.expr) -> bool:
return isinstance(node, ast.Constant) and node.value is None
def _bare_reraise(handler: ast.ExceptHandler) -> bool:
only = handler.body[0] if len(handler.body) == 1 else None
return isinstance(only, ast.Raise) and only.exc is None
def _snippet(node: ast.AST, limit: int = 60) -> str:
try:
rendered = ast.unparse(node)
except Exception:
return "<expr>"
return rendered if len(rendered) <= limit else rendered[: limit - 3] + "..."
def select_mutants(mutants: Sequence[Mutant], limit: int) -> List[Mutant]:
"""Round-robin over distinct lines so the budget spreads across the code path.
Swallowed-exception mutants come first: a test whose only claim is that the
call does not raise cannot be killed by anything else, and getting that
wrong would call a real regression test vacuous.
"""
swallows = [mutant for mutant in mutants if mutant.swallow][:limit]
if swallows:
return swallows + select_mutants([m for m in mutants if not m.swallow], limit - len(swallows))
by_line: Dict[Tuple[str, int], List[Mutant]] = {}
for mutant in mutants:
by_line.setdefault((mutant.path, mutant.lineno), []).append(mutant)
ordered: List[Mutant] = []
depth = 0
while len(ordered) < limit:
added = False
for key in sorted(by_line):
bucket = by_line[key]
if depth < len(bucket):
ordered.append(bucket[depth])
added = True
if len(ordered) == limit:
break
if not added:
break
depth += 1
return ordered
def interleave(groups: Sequence[Sequence[Mutant]], limit: int) -> List[Mutant]:
"""Take from each file in turn, so the module under test always gets mutants.
Files are already ranked with the module under test first, and a busy shared
module like a cache can otherwise eat the whole budget.
"""
return [
group[index]
for index in range(max((len(group) for group in groups), default=0))
for group in groups
if index < len(group)
][:limit]
def probe(test_id: str, max_mutants: int, max_files: int, timeout: int) -> ProbeReport:
code, output = run_test(test_id, timeout)
if code == 5:
return ProbeReport(test_id, "inconclusive", "test not collected (renamed or parametrized away)")
if code != 0:
return ProbeReport(test_id, "already_failing", f"baseline run failed:\n{output[-800:]}")
if " skipped" in output and " passed" not in output:
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())
if len(lines) >= MIN_LINES_PER_FILE
}
if not behavioural:
return ProbeReport(
test_id,
"vacuous",
"test executes no litellm function body of its own; it only imports modules",
)
imports = imported_modules(test_id.partition("::")[0])
ranked = sorted(
behavioural.items(),
key=lambda item: (not _is_under_test(item[0], imports), -len(item[1]), item[0]),
)[:max_files]
if not any(_is_under_test(path, imports) for path, _ in ranked):
return ProbeReport(
test_id,
"inconclusive",
"the test runs no function body of the modules it imports, only shared "
f"infrastructure ({', '.join(path for path, _ in ranked)}); needs a human",
covered_files={path: len(lines) for path, lines in ranked},
)
report = ProbeReport(
test_id,
"inconclusive",
"",
covered_files={path: len(lines) for path, lines in ranked},
)
candidates = interleave(
[select_mutants(generate_mutants(path, lines), max_mutants) for path, lines in ranked],
max_mutants,
)
if not candidates:
report.detail = "no mutable statements on the covered lines"
return report
kills = 0
for mutant in candidates:
outcome = _run_mutant(mutant, test_id, timeout)
report.mutants.append(MutantResult(mutant.path, mutant.lineno, mutant.description, outcome))
if outcome in {"killed", "timeout"}:
kills += 1
break
if kills:
report.verdict = "not_vacuous"
report.detail = f"killed by mutant: {report.mutants[-1].description} ({report.mutants[-1].path}:{report.mutants[-1].lineno})"
return report
tested = [m for m in report.mutants if m.outcome != "broken"]
if len(tested) < 3:
report.detail = f"only {len(tested)} usable mutant(s); not enough signal"
return report
report.verdict = "vacuous"
report.detail = f"survived all {len(tested)} mutants on the code it covers"
return report
def _mirror(source: str, destination: str, remaining: Sequence[str]) -> None:
os.makedirs(destination, exist_ok=True)
head = remaining[0]
for entry in os.scandir(source):
if entry.name != head:
os.symlink(entry.path, os.path.join(destination, entry.name))
if len(remaining) > 1:
_mirror(os.path.join(source, head), os.path.join(destination, head), remaining[1:])
@contextmanager
def mutant_overlay(mutant: Mutant) -> Iterator[str]:
"""A working directory where only the mutated file differs from the real tree.
Everything except the mutated file's own directory chain is symlinked, so the
real source is never written to: a crashed or killed probe cannot leave a
mutant behind in the repo. Running pytest with this as cwd puts it at the
front of sys.path, so the mutated module wins over the installed one.
"""
with tempfile.TemporaryDirectory(prefix="vacuous_mutant_") as overlay:
parts = mutant.path.split("/")
_mirror(REPO_ROOT, overlay, parts)
with open(os.path.join(overlay, mutant.path), "w", encoding="utf-8") as handle:
handle.write(mutant.source)
yield overlay
def _run_mutant(mutant: Mutant, test_id: str, timeout: int) -> str:
with mutant_overlay(mutant) as overlay:
code, output = run_test(test_id, timeout, overlay=overlay)
if code == 124:
return "timeout"
if code == 0:
return "survived"
if "ImportError" in output or "SyntaxError" in output or code == 5:
return "broken"
return "killed"
def record_cleared(test_id: str) -> None:
payload = {"test_ids": []}
if os.path.exists(CLEARED_PATH):
with open(CLEARED_PATH, "r", encoding="utf-8") as handle:
payload = json.load(handle)
ids = sorted(set(payload.get("test_ids", [])) | {test_id})
payload["test_ids"] = ids
payload.setdefault(
"_comment",
"Tests a mutation probe proved have teeth. inventory.py --queue skips these.",
)
with open(CLEARED_PATH, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2)
handle.write("\n")
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("test_id", help="pytest node id, e.g. tests/x/test_y.py::test_z")
parser.add_argument("--max-mutants", type=int, default=12)
parser.add_argument("--max-files", type=int, default=2)
parser.add_argument("--timeout", type=int, default=180, help="per-test-run timeout in seconds")
parser.add_argument("--json", metavar="PATH", help="write the report as JSON")
parser.add_argument("--record", action="store_true", help="record a not_vacuous verdict")
args = parser.parse_args()
report = probe(args.test_id, args.max_mutants, args.max_files, args.timeout)
if args.json:
with open(args.json, "w", encoding="utf-8") as handle:
json.dump(report.to_json(), handle, indent=2)
handle.write("\n")
print(f"{report.verdict}: {report.test_id}")
print(f" {report.detail}")
for mutant in report.mutants:
print(f" [{mutant.outcome}] {mutant.path}:{mutant.lineno} {mutant.description}")
if args.record and report.verdict == "not_vacuous":
record_cleared(report.test_id)
print(f" recorded in {os.path.relpath(CLEARED_PATH, REPO_ROOT)}")
return 0
if __name__ == "__main__":
sys.exit(main())

491
tests/vacuous_tests/test_vacuous_tooling.py Normal file
View file

@ -0,0 +1,491 @@
"""Tests for the vacuous-test audit tooling.
Both directions are pinned: the classifier flags what it claims to flag, and it
leaves healthy tests alone. A false positive here sends the daily automation to
rewrite a working test
"""
from __future__ import annotations
import ast
import os
import sys
import textwrap
from datetime import date
from typing import List, Optional
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import flake_gate
import guardrails
import inventory
import mutation_probe
def findings_for(source: str, monkeypatch, tmp_path) -> List[str]:
path = tmp_path / "test_target.py"
path.write_text(textwrap.dedent(source))
monkeypatch.setattr(flake_gate, "REPO_ROOT", str(tmp_path))
return [finding.problem for finding in flake_gate.static_findings("test_target.py::test_thing")]
def bucket_of(source: str, name: str = "test_thing") -> Optional[str]:
candidates = inventory.classify_file("/repo/tests/test_sample.py", textwrap.dedent(source))
for candidate in candidates:
if candidate.name.endswith(name):
return candidate.bucket
return None
def test_flags_trivial_assert() -> None:
assert bucket_of("def test_thing():\n do_work()\n assert True\n") == "trivial_assert"
def test_flags_self_comparison() -> None:
source = """
def test_thing():
value = compute()
assert value == value
"""
assert bucket_of(source) == "trivial_assert"
def test_flags_missing_assertion() -> None:
assert bucket_of("def test_thing():\n result = compute()\n print(result)\n") == "no_assert"
def test_flags_swallowed_assertion() -> None:
source = """
def test_thing():
try:
assert compute() == 3
except Exception:
pass
"""
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")
def test_thing():
assert compute() == 3
"""
assert bucket_of(source) == "dead_skip"
def test_flags_mock_only_comparison() -> None:
source = """
def test_thing():
client = MagicMock()
other = MagicMock()
assert client.send.return_value == other.send.return_value
"""
assert bucket_of(source) == "mock_tautology"
def test_ignores_real_assertion() -> None:
source = """
def test_thing():
assert compute() == 3
"""
assert bucket_of(source) is None
def test_ignores_pytest_raises_only_test() -> None:
source = """
def test_thing():
with pytest.raises(ValueError):
compute()
"""
assert bucket_of(source) is None
def test_ignores_assertion_in_shared_helper() -> None:
source = """
def test_thing():
assert_response_matches(compute(), expected)
"""
assert bucket_of(source) is None
def test_ignores_mock_passed_into_real_code() -> None:
source = """
def test_thing():
client = MagicMock()
result = handler(client)
assert result == client.send.return_value
"""
assert bucket_of(source) is None
def test_ignores_conditional_skip() -> None:
source = """
@pytest.mark.skipif(sys.platform == "win32", reason="posix only")
def test_thing():
assert compute() == 3
"""
assert bucket_of(source) is None
def test_classifies_methods_of_test_classes() -> None:
source = """
class TestThings:
def test_thing(self):
compute()
"""
candidates = inventory.classify_file("/repo/tests/test_sample.py", textwrap.dedent(source))
assert [c.name for c in candidates] == ["TestThings.test_thing"]
def test_classifies_camel_case_test_names_pytest_still_collects() -> None:
source = """
def testNoAssertCamelCase():
compute()
"""
assert bucket_of(source, "testNoAssertCamelCase") == "no_assert"
def test_classifies_tests_defined_inside_module_level_blocks() -> None:
source = """
if sys.version_info >= (3, 9):
def test_guarded():
compute()
try:
def test_in_try():
compute()
except ImportError:
pass
"""
candidates = inventory.classify_file("/repo/tests/test_sample.py", textwrap.dedent(source))
assert sorted(c.name for c in candidates) == ["test_guarded", "test_in_try"]
def test_ignores_tests_under_a_main_guard() -> None:
source = """
if __name__ == "__main__":
def test_manual_script():
compute()
"""
assert inventory.classify_file("/repo/tests/test_sample.py", textwrap.dedent(source)) == []
def test_ignores_nested_helpers_pytest_does_not_collect() -> None:
source = """
def test_outer():
def test_inner():
compute()
assert compute() == 3
"""
assert inventory.classify_file("/repo/tests/test_sample.py", textwrap.dedent(source)) == []
def test_ratchet_rejects_new_candidates() -> None:
failures = inventory.regressions(
{"tests/test_sample.py": {"no_assert": ["test_known", "test_new"]}},
{"tests/test_sample.py": {"no_assert": ["test_known"]}},
)
assert failures == ["tests/test_sample.py::test_new is a new no_assert candidate"]
def test_ratchet_rejects_a_replacement_that_keeps_the_count_unchanged() -> None:
failures = inventory.regressions(
{"tests/test_sample.py": {"no_assert": ["test_new"]}},
{"tests/test_sample.py": {"no_assert": ["test_fixed"]}},
)
assert failures == ["tests/test_sample.py::test_new is a new no_assert candidate"]
def test_ratchet_allows_fewer_candidates_and_untouched_files() -> None:
baseline = {
"tests/test_sample.py": {"no_assert": ["test_a", "test_b"]},
"tests/test_other.py": {"dead_skip": ["test_c"]},
}
assert inventory.regressions({"tests/test_sample.py": {"no_assert": ["test_a"]}}, baseline) == []
def test_ratchet_rejects_candidates_in_a_new_file() -> None:
failures = inventory.regressions({"tests/test_new.py": {"trivial_assert": ["test_thing"]}}, {})
assert failures == ["tests/test_new.py::test_thing is a new trivial_assert candidate"]
def test_identities_group_candidate_names_per_file_and_bucket() -> None:
source = """
def test_no_assert():
compute()
def test_trivial():
assert True
"""
path = os.path.join(inventory.REPO_ROOT, "tests", "test_sample.py")
candidates = inventory.classify_file(path, textwrap.dedent(source))
assert inventory.to_identities(candidates) == {
"tests/test_sample.py": {"no_assert": ["test_no_assert"], "trivial_assert": ["test_trivial"]}
}
def _mutant_descriptions(source: str, lines: List[int], tmp_path: str) -> List[str]:
path = os.path.join(tmp_path, "module.py")
with open(path, "w", encoding="utf-8") as handle:
handle.write(textwrap.dedent(source))
relative = os.path.relpath(path, mutation_probe.REPO_ROOT)
return [mutant.description for mutant in mutation_probe.generate_mutants(relative, lines)]
def test_mutates_only_covered_lines(tmp_path) -> None:
source = """
def covered(value):
return value > 3
def uncovered(value):
return value < 9
"""
descriptions = _mutant_descriptions(source, [3], str(tmp_path))
assert any("value > 3" in description for description in descriptions)
assert not any("value < 9" in description for description in descriptions)
def test_mutant_source_is_valid_python(tmp_path) -> None:
source = """
def covered(value):
if value == 3 and value is not None:
return "three"
return None
"""
path = os.path.join(str(tmp_path), "module.py")
with open(path, "w", encoding="utf-8") as handle:
handle.write(textwrap.dedent(source))
relative = os.path.relpath(path, mutation_probe.REPO_ROOT)
mutants = mutation_probe.generate_mutants(relative, [3, 4, 5])
assert mutants
for mutant in mutants:
ast.parse(mutant.source)
assert any("flip `Eq`" in mutant.description for mutant in mutants)
def test_import_time_lines_are_not_behavioural(tmp_path) -> None:
source = """
DEFAULT = True
class Config:
enabled = False
def check(self):
return self.enabled is True
"""
path = os.path.join(str(tmp_path), "module.py")
with open(path, "w", encoding="utf-8") as handle:
handle.write(textwrap.dedent(source))
relative = os.path.relpath(path, mutation_probe.REPO_ROOT)
assert mutation_probe.behavioural_lines(relative, range(1, 9)) == [8]
def test_overlay_isolates_the_mutant_from_the_real_tree(tmp_path, monkeypatch) -> None:
root = tmp_path / "root"
(root / "pkg" / "sub").mkdir(parents=True)
(root / "tests").mkdir()
target = root / "pkg" / "sub" / "mod.py"
target.write_text("X = 1\n")
(root / "pkg" / "other.py").write_text("Y = 1\n")
monkeypatch.setattr(mutation_probe, "REPO_ROOT", str(root))
mutant = mutation_probe.Mutant(path="pkg/sub/mod.py", lineno=1, description="d", source="X = 2\n")
with mutation_probe.mutant_overlay(mutant) as overlay:
assert open(os.path.join(overlay, "pkg", "sub", "mod.py"), encoding="utf-8").read() == "X = 2\n"
assert not os.path.islink(os.path.join(overlay, "pkg", "sub", "mod.py"))
assert os.path.islink(os.path.join(overlay, "tests"))
assert os.path.islink(os.path.join(overlay, "pkg", "other.py"))
assert target.read_text() == "X = 1\n"
assert not os.path.exists(overlay)
assert target.read_text() == "X = 1\n"
def test_module_under_test_is_recognised_from_imports() -> None:
imports = {"litellm.llms.bedrock.base_aws_llm", "litellm.llms.bedrock.base_aws_llm.BaseAWSLLM"}
assert mutation_probe._is_under_test("litellm/llms/bedrock/base_aws_llm.py", imports)
assert not mutation_probe._is_under_test("litellm/caching/dual_cache.py", imports)
def test_swallowed_exception_becomes_a_mutant_that_a_no_assert_test_can_notice(tmp_path, monkeypatch) -> None:
root = tmp_path / "root"
(root / "litellm").mkdir(parents=True)
target = root / "litellm" / "hooks.py"
target.write_text(
textwrap.dedent(
"""
def record(value):
try:
return int(value)
except ValueError:
return 0
"""
).lstrip()
)
monkeypatch.setattr(mutation_probe, "REPO_ROOT", str(root))
mutants = mutation_probe.generate_mutants("litellm/hooks.py", range(1, 6))
swallows = [m for m in mutants if m.swallow]
assert len(swallows) == 1
assert "stop swallowing" in swallows[0].description
assert "raise" in swallows[0].source
assert "return 0" not in swallows[0].source
# A test that only claims "this does not raise" dies to that mutant and to
# nothing else, so it has to be tried first.
assert mutation_probe.select_mutants(mutants, 2)[0].swallow
def test_already_reraising_handlers_produce_no_mutant(tmp_path, monkeypatch) -> None:
root = tmp_path / "root"
(root / "litellm").mkdir(parents=True)
(root / "litellm" / "hooks.py").write_text(
textwrap.dedent(
"""
def record(value):
try:
return int(value)
except ValueError:
raise
"""
).lstrip()
)
monkeypatch.setattr(mutation_probe, "REPO_ROOT", str(root))
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 = [
inventory.Candidate(path=path, lineno=index, name=f"test_{index}", bucket="no_assert", evidence="e")
for index, path in enumerate(
["tests/a/one.py"] * 3 + ["tests/b/two.py"] * 2 + ["tests/c/three.py"],
)
]
assert inventory.areas(candidates) == (("tests/a", 3), ("tests/b", 2), ("tests/c", 1))
picks = [inventory.rotated_area(candidates, date(2026, 8, day)) for day in (15, 16, 17, 18)]
assert len(set(picks[:3])) == 3
assert picks[3] == picks[0]
assert inventory.rotated_area(candidates, date(2026, 8, 15)) == picks[0]
assert inventory.rotated_area([], date(2026, 8, 15)) is None
def test_budget_reaches_the_module_under_test() -> None:
def mutant(path: str, lineno: int) -> mutation_probe.Mutant:
return mutation_probe.Mutant(path=path, lineno=lineno, description=f"{path}:{lineno}", source="")
under_test = [mutant("litellm/llms/bedrock/base_aws_llm.py", line) for line in (10, 11)]
shared = [mutant("litellm/caching/dual_cache.py", line) for line in range(100, 120)]
chosen = mutation_probe.interleave([under_test, shared], 8)
assert [m.path for m in chosen].count("litellm/llms/bedrock/base_aws_llm.py") == 2
assert len(chosen) == 8
def test_flake_gate_flags_sleep_and_wall_clock(monkeypatch, tmp_path) -> None:
source = """
def test_thing():
time.sleep(0.5)
started = datetime.datetime.now()
assert started
"""
problems = findings_for(source, monkeypatch, tmp_path)
assert any("time.sleep" in problem for problem in problems)
assert any("datetime.now" in problem for problem in problems)
def test_flake_gate_flags_live_call_outside_the_patch(monkeypatch, tmp_path) -> None:
source = """
def test_thing():
with patch("litellm.main.completion") as mocked:
litellm.completion(model="gpt-5", messages=[])
litellm.acompletion(model="gpt-5", messages=[])
"""
problems = findings_for(source, monkeypatch, tmp_path)
assert problems == ["uses `litellm.acompletion`: hits a live provider unless mocked or replayed"]
def test_flake_gate_accepts_patched_network_call(monkeypatch, tmp_path) -> None:
source = """
@patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post")
def test_thing(mocked_post):
response = litellm.completion(model="gpt-5", messages=[])
assert response.choices
"""
assert findings_for(source, monkeypatch, tmp_path) == []
def test_flake_gate_does_not_excuse_sleep_in_a_patched_test(monkeypatch, tmp_path) -> None:
source = """
@patch("litellm.llms.custom_httpx.http_handler.HTTPHandler.post")
def test_thing(mocked_post):
time.sleep(1)
assert mocked_post.called
"""
assert findings_for(source, monkeypatch, tmp_path) == [
"uses `time.sleep`: wall-clock sleep: slow and racy under load"
]
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"}
problems = guardrails.uncited_removals("tests/test_sample.py", removed, citations)
assert len(problems) == 1
assert "test_two was removed without a citation" in problems[0]
def test_blank_citation_does_not_count() -> None:
problems = guardrails.uncited_removals(
"tests/test_sample.py", frozenset({"test_one"}), {"tests/test_sample.py::test_one": " "}
)
assert len(problems) == 1

4
tests/vacuous_tests/verified_not_vacuous.json Normal file
View file

@ -0,0 +1,4 @@
{
"_comment": "Tests a mutation probe proved have teeth. inventory.py --queue skips these.",
"test_ids": []
}