scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

feat(ci): ratchet the test suite's zero-assert, mock-echo and global-state debt (#37588)

Browse source 
* feat(ci): ratchet the test suite's zero-assert, mock-echo and global-state debt

The suite's dominant failure mode is tests that cannot fail for the reason anyone
would want them to. The testing-strategy audit measured five shapes of it, and
nothing mechanical stops any of them from reproducing, so they keep reproducing.

`scripts/check_test_quality.py` is an AST checker for those five, emitting the
same `path:line: CODE message` contract as `scripts/check_type_discipline.py`:

  TQ001  a collectible test with no assertion of any kind
  TQ002  mock-echo, where every assertion only inspects the mock that was patched
  TQ003  sys.path.insert inside the test tree
  TQ004  raw `os.environ[...] =`, which leaks into whatever runs next
  TQ005  `litellm.<attr> =`, the process-wide leak the 491-line conftest undoes

`scripts/test_quality_gate.py` caps each rule against test-quality-budget.json,
seeded at exactly today's count, and fails only when a rule is both over its
limit and higher than the base being merged into, so a change is blamed for what
it adds and never for drift already in the base. `--update` lowers a limit by
what a branch cleared, so the ceilings only ever fall. It runs in the existing
required lint job, which means it enforces without a ruleset change.

TQ001 follows assertions into helpers defined in the same module, transitively.
Without that it flagged 111 tests in tests/e2e, the harness this program holds up
as the reference, because that suite factors its assertions into shared helpers
(`assert_auth_denied(result, ...)`). Following them leaves 25, all of which reach
their assertions across a module boundary; those are grandfathered and documented
rather than papered over.

The seeded counts land within about 10% of the audit's independent numbers for
every rule measured on the same subtree, which is the cross-check that the
definitions here match the ones the audit pinned.

* fix(ci): resolve test helpers per scope, not by bare name

The helper walk keyed every function in a module by its bare name, so two
same-named helpers in different classes collided and the last one parsed won.
A test calling `self._check()` could be cleared by a `_check` belonging to a
different class, or flagged because of one.

Resolution is now scoped: a bare name looks up the module-level functions, and
`self.<name>` looks up the enclosing class's own methods and no other class's.
Recursion is tracked by function identity rather than by name, so the cycle
guard cannot be confused by the same collision.

This surfaced one real zero-assert test that a same-named helper elsewhere had
been clearing, so TQ001 seeds at 750 rather than 749.

The test module has to register itself in sys.modules before exec_module:
`@dataclass(slots=True)` rebuilds its class through `sys.modules[__module__]`,
and Scope fails to construct without it. Recorded at the call site, since it
reads like avoidable global mutation otherwise.

* fix: register test-quality-budget.json with the ratchet alarm

The repo keeps one census over its budget files: every *-budget.json on disk
must appear in DEFAULT_BUDGETS, or its ceilings can be raised with no signal.
tests/test_litellm/test_budget_ratchet_check.py asserts that set equality and
caught the new budget on the way in.

Registering it also turns the alarm on for TQ001-TQ005, so a later PR cannot
quietly raise a test-quality ceiling. The file already uses the {limit: N}
schema the ratchet reads, so no other change was needed.
This commit is contained in:
yuneng-jiang 2026-08-20 10:08:49 -07:00 • committed by GitHub
parent 7de8f458e0
commit ffab5a39d0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 1250 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.github/workflows/test-linting.yml vendored
View file

@ -132,6 +132,11 @@ jobs:
run: |
uv run --no-sync python scripts/type_discipline_gate.py --base "$GATE_BASE_SHA"
- name: Check test-quality budget (zero-assert / mock-echo tests, sys.path.insert, raw env writes, litellm global mutation, delta vs base)
if: steps.changes.outputs.decision != 'skip'
run: |
uv run --no-sync python scripts/test_quality_gate.py --base "$GATE_BASE_SHA"
- name: Print OpenAI version
if: steps.changes.outputs.decision != 'skip'
run: |

18
Makefile
View file

@ -7,6 +7,7 @@
info lint lint-inner lint-dev lint-checks format \
lint-basedpyright lint-e2e-basedpyright lint-basedpyright-budget-update lint-type-discipline lint-type-discipline-budget-update \
lint-ruff-budget lint-ruff-budget-update lint-budget-update lint-gate \
lint-test-quality lint-test-quality-budget-update \
install-dev install-proxy-dev install-test-deps install-hooks \
install-helm-unittest check-circular-imports check-import-safety check check-inner pre-commit \
lint-install lint-fetch-base bootstrap
@ -35,7 +36,8 @@ help:
@echo " make lint-ruff-budget - Gate the codebase total of each strict ruff rule against its limit"
@echo " make lint-gate - Strict ruff gate in CI-parity mode (fetches staging, simulates the merge)"
@echo " make lint-ruff-budget-update - Ratchet ruff-strict-budget.json limits down by what this branch fixed"
@echo " make lint-budget-update - Ratchet all budgets down (ruff + type-discipline + basedpyright)"
@echo " make lint-test-quality - Gate the test suite against test-quality-budget.json"
@echo " make lint-budget-update - Ratchet all budgets down (ruff + type-discipline + test quality + basedpyright)"
@echo " make check-circular-imports - Check for circular imports"
@echo " make check-import-safety - Check import safety"
@echo " make test - Run all tests"
@ -200,6 +202,11 @@ lint-e2e-basedpyright: $(LINT_E2E_DEP_INSTALL)
lint-type-discipline: $(LINT_DEP_INSTALL) $(LINT_DEP_BASE)
$(UV_RUN) python scripts/type_discipline_gate.py --base origin/litellm_internal_staging
# Test-quality budget (zero-assert / mock-echo tests, sys.path.insert, raw env writes,
# litellm module-global mutation), counted across tests/ the same delta-vs-base way.
lint-test-quality: $(LINT_DEP_INSTALL) $(LINT_DEP_BASE)
$(UV_RUN) python scripts/test_quality_gate.py --base origin/litellm_internal_staging
# --update lowers each limit by what this branch fixed since its branch point, so
# it needs the base ref fetched to resolve the merge-base.
lint-basedpyright-budget-update: install-dev lint-fetch-base
@ -221,8 +228,11 @@ lint-ruff-budget-update: install-dev lint-fetch-base
lint-type-discipline-budget-update: install-dev lint-fetch-base
$(UV_RUN) python scripts/type_discipline_gate.py --update
# Ratchet all budgets in one shot (ruff strict + type-discipline + basedpyright)
lint-budget-update: lint-ruff-budget-update lint-type-discipline-budget-update lint-basedpyright-budget-update
lint-test-quality-budget-update: install-dev lint-fetch-base
$(UV_RUN) python scripts/test_quality_gate.py --update
# Ratchet all budgets in one shot (ruff strict + type-discipline + test quality + basedpyright)
lint-budget-update: lint-ruff-budget-update lint-type-discipline-budget-update lint-test-quality-budget-update lint-basedpyright-budget-update
check-circular-imports: $(LINT_DEP_INSTALL)
cd litellm && $(UV_RUN) python ../tests/documentation_tests/test_circular_imports.py && cd ..
@ -244,7 +254,7 @@ lint:
lint-inner: lint-install lint-fetch-base
$(MAKE) -j $(LINT_JOBS) $(LINT_OUTPUT_SYNC) LINT_DEP_INSTALL= LINT_E2E_DEP_INSTALL= LINT_DEP_BASE= lint-checks
lint-checks: lint-format-check-changed lint-ruff lint-gate lint-type-discipline lint-basedpyright lint-e2e-basedpyright check-circular-imports check-import-safety
lint-checks: lint-format-check-changed lint-ruff lint-gate lint-type-discipline lint-test-quality lint-basedpyright lint-e2e-basedpyright check-circular-imports check-import-safety
# Faster linting for local development (only checks changed code)
lint-dev: lint-format-changed check-circular-imports check-import-safety

1
scripts/budget_ratchet_check.py
View file

@ -44,6 +44,7 @@ DEFAULT_BUDGETS: tuple[str, ...] = (
"ruff-strict-budget.json",
"type-discipline-budget.json",
"basedpyright-code-budget.json",
"test-quality-budget.json",
)
GRADUATION_CONFIGS = MappingProxyType({"ruff-strict-budget.json": "ruff.toml"})

492
scripts/check_test_quality.py Normal file
View file

@ -0,0 +1,492 @@
#!/usr/bin/env python3
"""Test-quality checker: the test-suite smells no linter enforces.
Sibling of scripts/check_type_discipline.py, same output contract
(``path:line: CODE message``) and same stdlib-only constraint, aimed at the test
tree instead of the package. Each rule is a shape the testing-strategy audit
measured and named; scripts/test_quality_gate.py caps the codebase total of each
one against test-quality-budget.json so the counts can only ratchet down.
Rules
-----
TQ001 A collectible test function whose body contains no assertion of any kind:
no `assert` statement, no `pytest.raises`/`warns`/`deprecated_call`/`fail`,
and no `assert*` method call (mock's `assert_called_once`, unittest's
`assertEqual`, `numpy.testing.assert_allclose`). Such a test passes as long
as the code under it does not raise, so it pins nothing and cannot fail for
the reason anyone would want it to. Assert the observable output instead.
The whole function subtree counts, nested helper definitions included, so a
test that asserts inside a locally-defined async helper passes.
TQ002 Mock-echo: a test that patches something and whose every assertion only
inspects the mock that replaced it (`assert_called_once_with`, `.called`,
`.call_args`, `.call_count`, `.mock_calls`). The test restates the
implementation back at itself: it verifies that the code called what the
code calls, so it survives any refactor that keeps the call and breaks the
behavior. Assert what the caller observes -- the returned value, the
rebuilt response, the raised exception -- and fake at the HTTP boundary
(respx / MockTransport) rather than patching litellm internals.
A test with no assertions at all is TQ001, never TQ002.
TQ003 `sys.path.insert(...)` inside the test tree. pytest's rootdir handling and
the installed package already make `litellm` importable, so these are
no-ops carried by copy-paste; the ones that are not no-ops make the test's
imports depend on the working directory it happens to run from.
TQ004 Raw `os.environ[...] = ...` assignment. The write outlives the test and
leaks into whatever runs next in the same process, which is how a suite
acquires an ordering dependency. Use `monkeypatch.setenv`, which is undone
at teardown.
TQ005 `litellm.<attr> = ...` module-global mutation. The SDK's module globals are
process-wide, so this is the same leak as TQ004 one level up, and it is
what the 491-line save/restore conftest exists to paper over. Inject the
dependency or use a fixture that restores it.
Every rule is suppressible with `# test-quality-ok: <reason>` on the reported
line, following the repo's `*-ok: <reason>` convention. A suppression without a
reason does not suppress.
What counts as an assertion
---------------------------
An `assert` statement; `pytest.raises` / `warns` / `deprecated_call` / `fail`,
qualified or bare (`skip` and `xfail` are deliberately excluded, since they abort
the test rather than pin a behaviour); and any callable whose name starts with
`assert`, qualified (`m.assert_called_once`, `self.assertEqual`,
`np.testing.assert_allclose`) or bare (`assert_auth_denied(...)`, the shape the
e2e harness uses). A test also counts as asserting when it reaches an assertion
through a function defined in the same module, followed transitively, because
extracting the assertions into a shared helper is good factoring rather than a
test that pins nothing. A helper imported from another module is not followed, so
a test whose only assertions live across a module boundary still reports TQ001
and needs a suppression.
What counts as mock inspection (TQ002)
--------------------------------------
An `assert_`-prefixed call, which is mock's own family, or a reference to
`called` / `call_args` / `call_args_list` / `call_count` / `mock_calls` and their
await-counterparts. unittest's `assertEqual` has no underscore after "assert" and
so is never mistaken for one. A patch is installed by any call or decorator whose
name is `patch` or `patch.object` / `patch.dict` / `patch.multiple`, which covers
`unittest.mock` however it was imported as well as pytest-mock's `mocker.patch`.
Scope
-----
Only files under the test roots passed on the command line are examined, and
TQ001/TQ002 only look at functions pytest would collect: a `test_`-prefixed
function at module level, or a `test_`-prefixed method of a `Test`-prefixed
class that defines no `__init__`.
Usage
-----
python check_test_quality.py tests/
Exit code 1 if any violation is found. Stdlib only.
"""
from __future__ import annotations
import ast
import io
import re
import sys
import tokenize
from collections.abc import Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Final, NamedTuple
TEST_FUNCTION_PREFIX: Final = "test_"
TEST_CLASS_PREFIX: Final = "Test"
MIN_REASON_LEN: Final = 3
SUPPRESSION_TOKEN: Final = "test-quality-ok"
SUPPRESSION_RE: Final = re.compile(r"#\s*test-quality-ok(?::\s*(?P<reason>.*))?")
PYTEST_ASSERTION_HELPERS: Final = frozenset(("raises", "warns", "deprecated_call", "fail"))
MOCK_INSPECTION_ATTRIBUTES: Final = frozenset((
"called", "call_args", "call_args_list", "call_count", "mock_calls",
"await_args", "await_args_list", "await_count", "awaited",
))
MOCK_ASSERTION_PREFIX: Final = "assert_"
PATCH_MEMBERS: Final = frozenset(("object", "dict", "multiple"))
FunctionNode = ast.FunctionDef | ast.AsyncFunctionDef
class Violation(NamedTuple):
path: Path
line: int
code: str
message: str
def render(self) -> str:
return f"{self.path}:{self.line}: {self.code} {self.message}"
def _dotted_name(node: ast.expr) -> str:
"""`a.b.c` for an attribute chain rooted in a plain name, else ""."""
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
root: Final = _dotted_name(node.value)
return f"{root}.{node.attr}" if root else ""
return ""
def suppressed_lines(source: str) -> frozenset[int]:
"""Lines carrying `# test-quality-ok: <reason>` with a reason of usable length."""
try:
tokens: Final = tuple(tokenize.generate_tokens(io.StringIO(source).readline))
except (tokenize.TokenError, IndentationError, SyntaxError):
return frozenset()
return frozenset(
token.start[0]
for token in tokens
if token.type == tokenize.COMMENT
and (match := SUPPRESSION_RE.search(token.string)) is not None
and len((match.group("reason") or "").strip()) >= MIN_REASON_LEN
)
def _is_collectible_class(node: ast.ClassDef) -> bool:
"""pytest collects `Test`-prefixed classes that define no constructor."""
if not node.name.startswith(TEST_CLASS_PREFIX):
return False
return not any(
isinstance(child, ast.FunctionDef) and child.name == "__init__"
for child in node.body
)
def iter_test_functions(tree: ast.Module) -> Iterator[FunctionNode]:
"""Every function pytest would collect from this module, in source order."""
for node in tree.body:
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
if node.name.startswith(TEST_FUNCTION_PREFIX):
yield node
elif isinstance(node, ast.ClassDef) and _is_collectible_class(node):
yield from (
child
for child in node.body
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef))
and child.name.startswith(TEST_FUNCTION_PREFIX)
)
def _is_pytest_assertion_call(call: ast.Call) -> bool:
func: Final = call.func
if isinstance(func, ast.Attribute):
return func.attr in PYTEST_ASSERTION_HELPERS
if isinstance(func, ast.Name):
return func.id in PYTEST_ASSERTION_HELPERS
return False
def _is_assertion_helper_call(call: ast.Call) -> bool:
"""Any `assert*` callable: `x.assertEqual(...)`, `m.assert_called_once()`,
`np.testing.assert_allclose(...)`, and the bare shared helpers the e2e harness
uses (`assert_auth_denied(result, ...)`)."""
func: Final = call.func
if isinstance(func, ast.Attribute):
return func.attr.startswith("assert")
return isinstance(func, ast.Name) and func.id.startswith("assert")
def iter_assertions(function: FunctionNode) -> Iterator[ast.stmt | ast.Call]:
"""Every node in the function that pins a behaviour, nested definitions included."""
for node in ast.walk(function):
if isinstance(node, ast.Assert):
yield node
elif isinstance(node, ast.Call) and (
_is_pytest_assertion_call(node) or _is_assertion_helper_call(node)
):
yield node
class CallTarget(NamedTuple):
"""A call that might resolve to a function defined in this module: either a bare
name, looked up among the module-level functions, or a `self.` attribute, looked
up among the enclosing class's own methods."""
through_self: bool
name: str
@dataclass(frozen=True, slots=True)
class Scope:
"""What one function can reach by name. Keeping methods per-class is what stops
two same-named helpers in different classes from resolving to each other."""
module_level: Mapping[str, FunctionNode]
methods: Mapping[str, FunctionNode]
def resolve(self, target: CallTarget) -> FunctionNode | None:
source: Final = self.methods if target.through_self else self.module_level
return source.get(target.name)
def _call_target(func: ast.expr) -> CallTarget | None:
if isinstance(func, ast.Name):
return CallTarget(through_self=False, name=func.id)
if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name) and func.value.id == "self":
return CallTarget(through_self=True, name=func.attr)
return None
def _call_targets(function: FunctionNode) -> frozenset[CallTarget]:
return frozenset(
target
for node in ast.walk(function)
if isinstance(node, ast.Call)
for target in (_call_target(node.func),)
if target is not None
)
def _functions_in(body: Iterable[ast.stmt]) -> Mapping[str, FunctionNode]:
return MappingProxyType({
node.name: node
for node in body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
})
def build_scopes(tree: ast.Module) -> Mapping[FunctionNode, Scope]:
"""Every function in the module paired with what it can reach by name. A
module-level function sees only module-level functions; a method also sees its
own class's methods, and no other class's."""
module_level: Final = _functions_in(tree.body)
module_scope: Final = Scope(module_level=module_level, methods=MappingProxyType({}))
class_scopes: Final = tuple(
(node, Scope(module_level=module_level, methods=_functions_in(node.body)))
for node in tree.body
if isinstance(node, ast.ClassDef)
)
return MappingProxyType({
**{function: module_scope for function in module_level.values()},
**{
function: scope
for node, scope in class_scopes
for function in scope.methods.values()
},
})
def _reaches_assertion(
function: FunctionNode,
scopes: Mapping[FunctionNode, Scope],
seen: frozenset[FunctionNode],
) -> bool:
if function in seen:
return False
if any(iter_assertions(function)):
return True
scope: Final = scopes.get(function)
if scope is None:
return False
return any(
_reaches_assertion(callee, scopes, seen | frozenset((function,)))
for target in _call_targets(function)
for callee in (scope.resolve(target),)
if callee is not None
)
def asserts_through_helpers(
function: FunctionNode, scopes: Mapping[FunctionNode, Scope]
) -> bool:
"""Whether the test reaches an assertion through a function defined in this
module, followed transitively. Extracting the assertions into a shared helper is
good factoring rather than a test that pins nothing, so following one is what
keeps TQ001 honest."""
scope: Final = scopes.get(function)
if scope is None:
return False
return any(
_reaches_assertion(callee, scopes, frozenset((function,)))
for target in _call_targets(function)
for callee in (scope.resolve(target),)
if callee is not None
)
def _is_patch_installer(dotted: str) -> bool:
"""`patch`, `mock.patch`, `mocker.patch`, `patch.object`, `mock.patch.dict`, ..."""
parts: Final = dotted.split(".")
if parts[-1] == "patch":
return True
return len(parts) >= 2 and parts[-2] == "patch" and parts[-1] in PATCH_MEMBERS
def _installs_patch(function: FunctionNode) -> bool:
decorators: Final = tuple(
_dotted_name(d.func) if isinstance(d, ast.Call) else _dotted_name(d)
for d in function.decorator_list
)
if any(name and _is_patch_installer(name) for name in decorators):
return True
return any(
_is_patch_installer(_dotted_name(node.func))
for node in ast.walk(function)
if isinstance(node, ast.Call) and _dotted_name(node.func)
)
def _only_inspects_a_mock(node: ast.stmt | ast.Call) -> bool:
"""True when this assertion reads a mock's call record and nothing else."""
if isinstance(node, ast.Call):
func = node.func
return isinstance(func, ast.Attribute) and func.attr.startswith(MOCK_ASSERTION_PREFIX)
return any(
isinstance(child, ast.Attribute)
and (
child.attr in MOCK_INSPECTION_ATTRIBUTES
or child.attr.startswith(MOCK_ASSERTION_PREFIX)
)
for child in ast.walk(node)
)
def iter_assertion_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
scopes: Final = build_scopes(tree)
for function in iter_test_functions(tree):
assertions: Final = tuple(iter_assertions(function))
if not assertions and asserts_through_helpers(function, scopes):
continue
if not assertions:
yield Violation(
path,
function.lineno,
"TQ001",
f"test `{function.name}` asserts nothing, so it can only fail by raising; "
f"assert the observable output (suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
)
elif _installs_patch(function) and all(map(_only_inspects_a_mock, assertions)):
yield Violation(
path,
function.lineno,
"TQ002",
f"test `{function.name}` patches something and only asserts that the mock was "
f"called, which restates the implementation; assert what the caller observes "
f"(suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
)
def iter_sys_path_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
for node in ast.walk(tree):
if isinstance(node, ast.Call) and _dotted_name(node.func) == "sys.path.insert":
yield Violation(
path,
node.lineno,
"TQ003",
"sys.path.insert in a test; pytest's rootdir and the installed package already "
f"make litellm importable (suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
)
def _environ_subscript_targets(target: ast.expr) -> Iterator[ast.Subscript]:
if isinstance(target, ast.Tuple):
for element in target.elts:
yield from _environ_subscript_targets(element)
return
if isinstance(target, ast.Subscript) and _dotted_name(target.value) in ("os.environ", "environ"):
yield target
def iter_environ_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
for node in ast.walk(tree):
targets: Final = (
node.targets if isinstance(node, ast.Assign)
else (node.target,) if isinstance(node, (ast.AugAssign, ast.AnnAssign))
else ()
)
for target in targets:
for subscript in _environ_subscript_targets(target):
yield Violation(
path,
subscript.lineno,
"TQ004",
"raw os.environ write leaks into every test that runs after this one; "
f"use monkeypatch.setenv (suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
)
def _litellm_attribute_targets(target: ast.expr) -> Iterator[ast.Attribute]:
if isinstance(target, ast.Tuple):
for element in target.elts:
yield from _litellm_attribute_targets(element)
return
if isinstance(target, ast.Attribute) and _dotted_name(target.value) == "litellm":
yield target
def iter_global_mutation_violations(path: Path, tree: ast.Module) -> Iterator[Violation]:
for node in ast.walk(tree):
targets: Final = (
node.targets if isinstance(node, ast.Assign)
else (node.target,) if isinstance(node, (ast.AugAssign, ast.AnnAssign))
else ()
)
for target in targets:
for attribute in _litellm_attribute_targets(target):
yield Violation(
path,
attribute.lineno,
"TQ005",
f"litellm.{attribute.attr} is a process-wide global; writing it here is what the "
"save/restore conftest exists to undo, so inject the dependency or use a fixture "
f"(suppress: `# {SUPPRESSION_TOKEN}: <reason>`)",
)
def check_file(path: Path) -> tuple[Violation, ...]:
try:
source: Final = path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError) as exc:
return (Violation(path, 0, "TQ000", f"unreadable: {exc}"),)
try:
tree: Final = ast.parse(source, filename=str(path))
except SyntaxError as exc:
return (Violation(path, exc.lineno or 0, "TQ000", f"syntax error: {exc.msg}"),)
skip: Final = suppressed_lines(source)
return tuple(
violation
for violation in (
*iter_assertion_violations(path, tree),
*iter_sys_path_violations(path, tree),
*iter_environ_violations(path, tree),
*iter_global_mutation_violations(path, tree),
)
if violation.line not in skip
)
def collect_paths(raw: Iterable[str]) -> Iterator[Path]:
for item in raw:
candidate: Final = Path(item)
if candidate.is_dir():
yield from sorted(candidate.rglob("*.py"))
elif candidate.suffix == ".py":
yield candidate
def main(argv: Sequence[str]) -> int:
paths: Final = tuple(a for a in argv if not a.startswith("-"))
if not paths:
print("usage: check_test_quality.py <files-or-dirs>...", file=sys.stderr)
return 2
violations: Final = sorted(v for path in collect_paths(paths) for v in check_file(path))
for violation in violations:
print(violation.render())
if violations:
print(f"\n{len(violations)} violation(s).", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))

289
scripts/test_quality_gate.py Normal file
View file

@ -0,0 +1,289 @@
#!/usr/bin/env python3
"""Total-count gate for the TQ* rules in scripts/check_test_quality.py.
Sibling of scripts/type_discipline_gate.py, pointed at the test tree instead of
the package. Each rule listed in test-quality-budget.json has a hard ``limit``.
The gate counts each rule across the whole `tests` tree and fails when a rule is
both over its limit and higher than the base it merges into, so a change is
blamed for the violations it adds, never for drift that already exists in the
base.
Every rule is seeded at exactly its count on the day the gate landed, so the
suite's existing debt is grandfathered and any net-new violation trips the gate
immediately. ``--update`` ratchets a limit down by the violations this branch
fixed relative to its branch point (the merge-base), so the ceilings only ever
fall. A rule absent from the budget at the merge-base was seeded on this branch;
``--update`` leaves its limit untouched, because the base tree predates the rule
and its whole grandfathered count would otherwise be misread as "fixed".
The deliberate difference from its sibling: this gate has no headroom anywhere.
Type discipline seeded LIT010/LIT011 at 1.5x to leave room for an in-flight
sweep; a test-quality violation has no such transition to absorb, so the line is
today's count and the only legal direction is down.
"""
from __future__ import annotations
import argparse
import json
import re
import shutil
import subprocess
import sys
import tempfile
from collections import Counter
from collections.abc import Mapping, Sequence
from pathlib import Path
from types import MappingProxyType
from typing import Final, NamedTuple
REPO_ROOT: Final = Path(__file__).resolve().parent.parent
CHECKER: Final = REPO_ROOT / "scripts" / "check_test_quality.py"
BUDGET_PATH: Final = REPO_ROOT / "test-quality-budget.json"
TARGET: Final = "tests"
DEFAULT_BASE: Final = "origin/litellm_internal_staging"
_HUNK: Final = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@", re.MULTILINE)
_FILE_HEADER: Final = re.compile(r"^\+\+\+ b/(.+)$", re.MULTILINE)
_LINE: Final = re.compile(r"^(?P<file>.+?):(?P<line>\d+): (?P<code>TQ\d+) ")
class Violation(NamedTuple):
file: str
line: int
code: str
class Breach(NamedTuple):
rule: str
total: int
cap: int
added: int
def _run(cmd: Sequence[str], cwd: Path = REPO_ROOT) -> str:
proc: Final = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
if proc.returncode not in (0, 1):
sys.stderr.write(proc.stderr)
raise SystemExit(f"{cmd[0]} exited {proc.returncode}")
return proc.stdout
def resolve_base_point(base_ref: str, cwd: Path = REPO_ROOT) -> str:
"""The snapshot commit base counts are measured at: merge-base(base_ref, HEAD),
made aware of an in-progress merge. Mid-merge, HEAD is still the pre-merge tip,
so its merge-base is the old branch point and every violation the base gained
since then would be blamed on this change."""
head_point: Final = _run(["git", "merge-base", base_ref, "HEAD"], cwd=cwd).strip()
if not head_point:
return base_ref
merge_head: Final = _run(["git", "rev-parse", "--verify", "--quiet", "MERGE_HEAD"], cwd=cwd).strip()
if not merge_head:
return head_point
merge_point: Final = _run(["git", "merge-base", base_ref, merge_head], cwd=cwd).strip()
if not merge_point:
return head_point
older: Final = _run(["git", "merge-base", head_point, merge_point], cwd=cwd).strip()
return merge_point if older == head_point else head_point
def _check(root: Path, checker: Path) -> tuple[Violation, ...]:
# macOS tempfile dirs (/var/...) resolve to /private/var/..., so relative_to needs both sides resolved.
resolved: Final = root.resolve()
out: Final = _run([sys.executable, str(checker), str(resolved / TARGET)], cwd=resolved)
return tuple(
Violation(
(resolved / match.group("file")).resolve().relative_to(resolved).as_posix(),
int(match.group("line")),
match.group("code"),
)
for line in out.splitlines()
if (match := _LINE.match(line)) is not None
)
def head_violations() -> tuple[Violation, ...]:
return _check(REPO_ROOT, CHECKER)
def count_by_rule(violations: Sequence[Violation]) -> Mapping[str, int]:
return MappingProxyType(dict(Counter(v.code for v in violations)))
def base_counts(ref: str) -> Mapping[str, int]:
"""Rule counts at `ref`, measured with the *current* rule logic rather than
whatever the checker looked like at that commit."""
parent: Final = Path(tempfile.mkdtemp(prefix="tq_base_"))
worktree: Final = parent / "wt"
try:
_run(["git", "worktree", "add", "--detach", str(worktree), ref])
(worktree / "scripts").mkdir(parents=True, exist_ok=True)
checker: Final = worktree / "scripts" / "check_test_quality.py"
shutil.copy(CHECKER, checker)
return count_by_rule(_check(worktree, checker))
finally:
# Teardown must never raise, or it masks the real error when the body failed.
subprocess.run(
["git", "worktree", "remove", "--force", str(worktree)],
cwd=REPO_ROOT, capture_output=True, text=True,
)
shutil.rmtree(parent, ignore_errors=True)
def over_ceiling(head: Mapping[str, int], budget: Mapping[str, Mapping[str, int]]) -> frozenset[str]:
"""Rules whose head count already exceeds their limit. When none are, the base
comparison cannot change the verdict and the base worktree scan is skipped."""
return frozenset(
rule for rule, spec in budget.items() if head.get(rule, 0) > spec["limit"]
)
def evaluate(
head: Mapping[str, int],
base: Mapping[str, int],
budget: Mapping[str, Mapping[str, int]],
) -> tuple[Breach, ...]:
return tuple(sorted(
Breach(rule, head.get(rule, 0), spec["limit"], head.get(rule, 0) - base.get(rule, 0))
for rule, spec in budget.items()
if head.get(rule, 0) > spec["limit"] and head.get(rule, 0) > base.get(rule, 0)
))
def _hunk_lines(body: str) -> frozenset[int]:
return frozenset(
line
for match in _HUNK.finditer(body)
for start in (int(match.group(1)),)
for line in range(start, start + (int(match.group(2)) if match.group(2) is not None else 1))
)
def parse_changed_lines(diff_text: str) -> Mapping[str, frozenset[int]]:
"""Each file in the diff mapped to the line numbers it adds. Splitting on the
`+++ b/` headers keeps this a pure expression: `split` hands back
[preamble, path, body, path, body, ...], so each file's hunks are already
grouped with it."""
parts: Final = _FILE_HEADER.split(diff_text)
return MappingProxyType({
path: _hunk_lines(body)
for path, body in zip(parts[1::2], parts[2::2])
})
def introduced(
violations: Sequence[Violation], changed: Mapping[str, frozenset[int]]
) -> tuple[Violation, ...]:
return tuple(v for v in violations if v.line in changed.get(v.file, frozenset()))
def cmd_check(base: str) -> None:
budget: Final = json.loads(BUDGET_PATH.read_text())
head: Final = head_violations()
head_counts: Final = count_by_rule(head)
if not over_ceiling(head_counts, budget):
print(f"OK: every TQ rule is within its test-suite ceiling (base {base})")
return
base_point: Final = resolve_base_point(base)
breaches: Final = evaluate(head_counts, base_counts(base_point), budget)
if not breaches:
print(f"OK: every TQ rule is within its test-suite ceiling (base {base})")
return
new: Final = introduced(
head,
parse_changed_lines(
_run(["git", "diff", base_point, "--unified=0", "--no-color", "--", TARGET])
),
)
print(f"FAIL: TQ-rule totals exceed their limit (base {base}):")
for breach in breaches:
print(
f" {breach.rule}: total {breach.total} over limit {breach.cap} "
f"(this change added {breach.added})"
)
for violation in sorted(v for v in new if v.code == breach.rule):
print(f" {violation.file}:{violation.line}")
print(
"Fix the new violations, or give each one a reason "
"(`# test-quality-ok: <reason>`), or remove an equal number elsewhere; "
"the ceiling is the limit in test-quality-budget.json. "
"Run `python scripts/check_test_quality.py tests/` to see every finding."
)
raise SystemExit(1)
def ratcheted_budget(
budget: Mapping[str, Mapping[str, int]],
current: Mapping[str, int],
base: Mapping[str, int],
seeded: frozenset[str] = frozenset(),
) -> Mapping[str, Mapping[str, int]]:
"""Each rule's limit lowered by the violations `current` fixed vs `base`. The drop
is clamped to what was actually cleared, so a limit only ever falls. Rules in
`seeded` were introduced on this branch and pass through untouched."""
return MappingProxyType({
rule: {
"limit": spec["limit"] if rule in seeded
else max(0, spec["limit"] - max(0, base.get(rule, 0) - current.get(rule, 0)))
}
for rule, spec in sorted(budget.items())
})
def _base_budget_rules(base_point: str) -> frozenset[str]:
proc: Final = subprocess.run(
["git", "show", f"{base_point}:{BUDGET_PATH.name}"],
cwd=REPO_ROOT, capture_output=True, text=True,
)
if proc.returncode != 0:
return frozenset()
return frozenset(json.loads(proc.stdout))
def cmd_update(base_ref: str = DEFAULT_BASE) -> None:
"""Ratchet each rule's limit down by the violations this branch fixed."""
budget: Final = json.loads(BUDGET_PATH.read_text())
base_point: Final = resolve_base_point(base_ref)
seeded: Final = frozenset(budget) - _base_budget_rules(base_point)
updated: Final = ratcheted_budget(
budget, count_by_rule(head_violations()), base_counts(base_point), seeded
)
BUDGET_PATH.write_text(json.dumps(dict(updated), indent=2, sort_keys=True) + "\n")
cleared: Final = sum(budget[rule]["limit"] - updated[rule]["limit"] for rule in updated)
print(f"Ratcheted TQ-rule limits down by {cleared} violations this branch fixed")
if seeded:
print(
"Left untouched (seeded on this branch, absent from the base budget): "
+ ", ".join(sorted(seeded))
)
def cmd_seed() -> None:
"""Write the budget from the working tree's current counts. Used once, to land
the gate; afterwards `--update` is the only thing that may move a limit."""
counts: Final = count_by_rule(head_violations())
BUDGET_PATH.write_text(
json.dumps({rule: {"limit": counts[rule]} for rule in sorted(counts)}, indent=2) + "\n"
)
print(f"Seeded {BUDGET_PATH.name} at " + ", ".join(f"{r}={counts[r]}" for r in sorted(counts)))
def main() -> None:
parser: Final = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", default=DEFAULT_BASE)
parser.add_argument("--update", action="store_true")
parser.add_argument("--seed", action="store_true")
args: Final = parser.parse_args()
from gate_slot_lock import held_slot
with held_slot():
if args.seed:
cmd_seed()
elif args.update:
cmd_update(args.base)
else:
cmd_check(args.base)
if __name__ == "__main__":
main()

17
test-quality-budget.json Normal file
View file

@ -0,0 +1,17 @@
{
"TQ001": {
"limit": 750
},
"TQ002": {
"limit": 742
},
"TQ003": {
"limit": 1078
},
"TQ004": {
"limit": 770
},
"TQ005": {
"limit": 2835
}
}

307
tests/test_litellm/test_check_test_quality.py Normal file
View file

@ -0,0 +1,307 @@
"""Tests for scripts/check_test_quality.py.
Every rule is exercised on a snippet that violates it and on one that does not, so
dropping a rule, widening it, or inverting the suppression check makes a test fail.
The helper-resolution cases are the regression for the false positives the rule
produced against tests/e2e, where the assertions live in a shared helper rather than
in the test body.
"""
import importlib.util
import sys
from pathlib import Path
_REPO_ROOT = Path(__file__).resolve().parents[2]
_MODULE_PATH = _REPO_ROOT / "scripts" / "check_test_quality.py"
_spec = importlib.util.spec_from_file_location("check_test_quality", _MODULE_PATH)
checker = importlib.util.module_from_spec(_spec)
# @dataclass(slots=True) rebuilds its class through sys.modules[__module__], so the
# module has to be registered before exec_module runs or Scope fails to construct.
sys.modules[_spec.name] = checker
_spec.loader.exec_module(checker)
def _codes(tmp_path, source):
snippet = tmp_path / "test_snippet.py"
snippet.write_text(source, encoding="utf-8")
return [v.code for v in checker.check_file(snippet)]
def test_zero_assert_test_is_flagged(tmp_path):
assert _codes(tmp_path, "def test_nothing():\n compute()\n") == ["TQ001"]
def test_plain_assert_statement_clears_the_rule(tmp_path):
assert _codes(tmp_path, "def test_value():\n assert compute() == 3\n") == []
def test_pytest_raises_counts_as_an_assertion(tmp_path):
source = "import pytest\n\n\ndef test_raises():\n with pytest.raises(ValueError):\n compute()\n"
assert _codes(tmp_path, source) == []
def test_unittest_style_assertion_counts(tmp_path):
source = "class TestThing:\n def test_equal(self):\n self.assertEqual(compute(), 3)\n"
assert _codes(tmp_path, source) == []
def test_bare_assert_helper_call_counts(tmp_path):
source = "def test_denied():\n assert_auth_denied(call(), 'missing header')\n"
assert _codes(tmp_path, source) == []
def test_assertion_inside_a_module_local_helper_clears_the_rule(tmp_path):
source = (
"def _drive_and_check(client):\n"
" assert client.status == 429\n"
"\n"
"\n"
"def test_budget_blocks(client):\n"
" _drive_and_check(client)\n"
)
assert _codes(tmp_path, source) == []
def test_helper_chain_is_followed_transitively(tmp_path):
source = (
"def _inner(x):\n"
" assert x == 1\n"
"\n"
"\n"
"def _outer(x):\n"
" _inner(x)\n"
"\n"
"\n"
"def test_chain():\n"
" _outer(1)\n"
)
assert _codes(tmp_path, source) == []
def test_a_same_named_helper_in_another_class_does_not_clear_the_rule(tmp_path):
source = (
"class TestAsserting:\n"
" def _check(self):\n"
" assert compute() == 3\n"
"\n"
" def test_ok(self):\n"
" self._check()\n"
"\n"
"\n"
"class TestNotAsserting:\n"
" def _check(self):\n"
" compute()\n"
"\n"
" def test_nothing(self):\n"
" self._check()\n"
)
assert _codes(tmp_path, source) == ["TQ001"]
def test_self_call_resolves_to_the_enclosing_class(tmp_path):
source = (
"class TestOne:\n"
" def _check(self):\n"
" assert compute() == 3\n"
"\n"
" def test_ok(self):\n"
" self._check()\n"
)
assert _codes(tmp_path, source) == []
def test_a_method_named_like_a_module_helper_does_not_shadow_it(tmp_path):
source = (
"def _check():\n"
" assert compute() == 3\n"
"\n"
"\n"
"class TestThing:\n"
" def _check(self):\n"
" compute()\n"
"\n"
" def test_bare_name_uses_the_module_helper(self):\n"
" _check()\n"
"\n"
" def test_self_uses_the_method(self):\n"
" self._check()\n"
)
assert _codes(tmp_path, source) == ["TQ001"]
def test_helper_without_assertions_does_not_clear_the_rule(tmp_path):
source = (
"def _just_calls(client):\n"
" client.go()\n"
"\n"
"\n"
"def test_nothing_anywhere(client):\n"
" _just_calls(client)\n"
)
assert _codes(tmp_path, source) == ["TQ001"]
def test_mutually_recursive_helpers_terminate(tmp_path):
source = (
"def _a(x):\n"
" _b(x)\n"
"\n"
"\n"
"def _b(x):\n"
" _a(x)\n"
"\n"
"\n"
"def test_cycle():\n"
" _a(1)\n"
)
assert _codes(tmp_path, source) == ["TQ001"]
def test_non_test_function_is_not_collected(tmp_path):
assert _codes(tmp_path, "def helper_without_asserts():\n compute()\n") == []
def test_class_with_a_constructor_is_not_collected(tmp_path):
source = (
"class TestLegacy:\n"
" def __init__(self):\n"
" self.x = 1\n"
"\n"
" def test_nothing(self):\n"
" compute()\n"
)
assert _codes(tmp_path, source) == []
def test_mock_echo_is_flagged(tmp_path):
source = (
"from unittest.mock import patch\n"
"\n"
"\n"
"def test_echo():\n"
" with patch('litellm.completion') as mock_completion:\n"
" run()\n"
" mock_completion.assert_called_once()\n"
)
assert _codes(tmp_path, source) == ["TQ002"]
def test_call_args_inspection_is_mock_echo(tmp_path):
source = (
"from unittest.mock import patch\n"
"\n"
"\n"
"def test_echo():\n"
" with patch('litellm.completion') as mock_completion:\n"
" run()\n"
" assert mock_completion.call_args[1]['model'] == 'gpt-4o'\n"
)
assert _codes(tmp_path, source) == ["TQ002"]
def test_patch_decorator_counts_as_installing_a_patch(tmp_path):
source = (
"from unittest import mock\n"
"\n"
"\n"
"@mock.patch('litellm.completion')\n"
"def test_echo(mock_completion):\n"
" run()\n"
" mock_completion.assert_called_once()\n"
)
assert _codes(tmp_path, source) == ["TQ002"]
def test_patching_but_asserting_the_output_is_not_mock_echo(tmp_path):
source = (
"from unittest.mock import patch\n"
"\n"
"\n"
"def test_output():\n"
" with patch('litellm.completion') as mock_completion:\n"
" result = run()\n"
" mock_completion.assert_called_once()\n"
" assert result.choices[0].message.content == 'pong'\n"
)
assert _codes(tmp_path, source) == []
def test_asserting_without_patching_is_not_mock_echo(tmp_path):
source = "def test_plain():\n m = build()\n assert m.called\n"
assert _codes(tmp_path, source) == []
def test_a_test_with_no_assertions_is_tq001_not_tq002(tmp_path):
source = (
"from unittest.mock import patch\n"
"\n"
"\n"
"def test_nothing():\n"
" with patch('litellm.completion'):\n"
" run()\n"
)
assert _codes(tmp_path, source) == ["TQ001"]
def test_sys_path_insert_is_flagged(tmp_path):
assert _codes(tmp_path, "import sys\n\nsys.path.insert(0, '..')\n") == ["TQ003"]
def test_sys_path_read_is_not_flagged(tmp_path):
assert _codes(tmp_path, "import sys\n\nprint(sys.path)\n") == []
def test_raw_environ_write_is_flagged(tmp_path):
assert _codes(tmp_path, "import os\n\nos.environ['KEY'] = 'v'\n") == ["TQ004"]
def test_bare_environ_write_is_flagged(tmp_path):
assert _codes(tmp_path, "from os import environ\n\nenviron['KEY'] = 'v'\n") == ["TQ004"]
def test_environ_read_is_not_flagged(tmp_path):
assert _codes(tmp_path, "import os\n\nvalue = os.environ.get('KEY')\n") == []
def test_monkeypatch_setenv_is_not_flagged(tmp_path):
source = "def test_env(monkeypatch):\n monkeypatch.setenv('KEY', 'v')\n assert read() == 'v'\n"
assert _codes(tmp_path, source) == []
def test_litellm_global_write_is_flagged(tmp_path):
assert _codes(tmp_path, "import litellm\n\nlitellm.drop_params = True\n") == ["TQ005"]
def test_litellm_augmented_global_write_is_flagged(tmp_path):
assert _codes(tmp_path, "import litellm\n\nlitellm.num_retries += 1\n") == ["TQ005"]
def test_litellm_attribute_read_is_not_flagged(tmp_path):
assert _codes(tmp_path, "import litellm\n\nvalue = litellm.drop_params\n") == []
def test_unrelated_attribute_write_is_not_flagged(tmp_path):
assert _codes(tmp_path, "config.drop_params = True\n") == []
def test_suppression_with_a_reason_clears_the_violation(tmp_path):
source = "import sys\n\nsys.path.insert(0, '..') # test-quality-ok: vendored path is required here\n"
assert _codes(tmp_path, source) == []
def test_suppression_without_a_reason_does_not_suppress(tmp_path):
assert _codes(tmp_path, "import sys\n\nsys.path.insert(0, '..') # test-quality-ok:\n") == ["TQ003"]
def test_suppression_on_another_line_does_not_suppress(tmp_path):
source = "import sys # test-quality-ok: this reason sits on the wrong line\n\nsys.path.insert(0, '..')\n"
assert _codes(tmp_path, source) == ["TQ003"]
def test_unparseable_source_degrades_to_tq000(tmp_path):
assert _codes(tmp_path, "def test_broken(:\n pass\n") == ["TQ000"]
def test_every_violation_renders_as_path_line_code_message():
rendered = checker.Violation(Path("tests/test_x.py"), 7, "TQ001", "nothing asserted").render()
assert rendered == "tests/test_x.py:7: TQ001 nothing asserted"

125
tests/test_litellm/test_test_quality_gate.py Normal file
View file

@ -0,0 +1,125 @@
"""Tests for scripts/test_quality_gate.py.
The gate's whole value is that it blames a change only for what it adds, and that a
limit can never rise. Both properties live in pure functions, so they are tested
directly: `evaluate` for the blame rule, `ratcheted_budget` for the one-way ratchet,
and `parse_changed_lines` for the diff scan that turns a breach into file:line.
"""
import importlib.util
import sys
from pathlib import Path
_REPO_ROOT = Path(__file__).resolve().parents[2]
_MODULE_PATH = _REPO_ROOT / "scripts" / "test_quality_gate.py"
_spec = importlib.util.spec_from_file_location("test_quality_gate", _MODULE_PATH)
gate = importlib.util.module_from_spec(_spec)
# @dataclass(slots=True) rebuilds its class through sys.modules[__module__], so the
# module has to be registered before exec_module runs or Scope fails to construct.
sys.modules[_spec.name] = gate
_spec.loader.exec_module(gate)
_BUDGET = {"TQ001": {"limit": 10}, "TQ003": {"limit": 5}}
def test_a_rule_within_its_limit_is_not_a_breach():
assert gate.evaluate({"TQ001": 10}, {"TQ001": 10}, _BUDGET) == ()
def test_a_rule_over_its_limit_that_the_change_added_is_a_breach():
breaches = gate.evaluate({"TQ001": 12}, {"TQ001": 10}, _BUDGET)
assert [(b.rule, b.total, b.cap, b.added) for b in breaches] == [("TQ001", 12, 10, 2)]
def test_drift_already_in_the_base_is_not_blamed_on_the_change():
assert gate.evaluate({"TQ001": 14}, {"TQ001": 14}, _BUDGET) == ()
def test_a_change_that_reduces_an_over_limit_rule_is_not_blamed():
assert gate.evaluate({"TQ001": 13}, {"TQ001": 14}, _BUDGET) == ()
def test_a_rule_absent_from_head_counts_as_zero():
assert gate.evaluate({}, {}, _BUDGET) == ()
def test_over_ceiling_names_only_the_rules_above_their_limit():
assert gate.over_ceiling({"TQ001": 11, "TQ003": 5}, _BUDGET) == frozenset({"TQ001"})
def test_over_ceiling_is_empty_when_everything_fits():
assert gate.over_ceiling({"TQ001": 10, "TQ003": 4}, _BUDGET) == frozenset()
def test_ratchet_lowers_a_limit_by_what_the_branch_fixed():
updated = gate.ratcheted_budget(_BUDGET, {"TQ001": 6}, {"TQ001": 10})
assert updated["TQ001"]["limit"] == 6
def test_ratchet_never_raises_a_limit_when_violations_grew():
updated = gate.ratcheted_budget(_BUDGET, {"TQ001": 20}, {"TQ001": 10})
assert updated["TQ001"]["limit"] == 10
def test_ratchet_never_goes_below_zero():
updated = gate.ratcheted_budget({"TQ001": {"limit": 2}}, {"TQ001": 0}, {"TQ001": 100})
assert updated["TQ001"]["limit"] == 0
def test_ratchet_leaves_a_rule_seeded_on_this_branch_untouched():
updated = gate.ratcheted_budget(
_BUDGET, {"TQ001": 0}, {"TQ001": 10}, seeded=frozenset({"TQ001"})
)
assert updated["TQ001"]["limit"] == 10
def test_parse_changed_lines_groups_hunks_under_their_own_file():
diff = (
"diff --git a/tests/a.py b/tests/a.py\n"
"--- a/tests/a.py\n"
"+++ b/tests/a.py\n"
"@@ -0,0 +3,2 @@\n"
"+one\n"
"+two\n"
"diff --git a/tests/b.py b/tests/b.py\n"
"--- a/tests/b.py\n"
"+++ b/tests/b.py\n"
"@@ -0,0 +10 @@\n"
"+only\n"
)
changed = gate.parse_changed_lines(diff)
assert changed["tests/a.py"] == frozenset({3, 4})
assert changed["tests/b.py"] == frozenset({10})
def test_parse_changed_lines_handles_several_hunks_in_one_file():
diff = (
"+++ b/tests/a.py\n"
"@@ -0,0 +1,2 @@\n"
"+a\n"
"@@ -9,0 +20,1 @@\n"
"+b\n"
)
assert gate.parse_changed_lines(diff)["tests/a.py"] == frozenset({1, 2, 20})
def test_parse_changed_lines_on_an_empty_diff_is_empty():
assert dict(gate.parse_changed_lines("")) == {}
def test_introduced_keeps_only_violations_on_changed_lines():
violations = (
gate.Violation("tests/a.py", 3, "TQ001"),
gate.Violation("tests/a.py", 99, "TQ001"),
gate.Violation("tests/b.py", 3, "TQ003"),
)
kept = gate.introduced(violations, {"tests/a.py": frozenset({3})})
assert kept == (gate.Violation("tests/a.py", 3, "TQ001"),)
def test_the_shipped_budget_covers_every_rule_the_checker_can_emit():
import json
budget = json.loads((_REPO_ROOT / "test-quality-budget.json").read_text())
assert set(budget) == {"TQ001", "TQ002", "TQ003", "TQ004", "TQ005"}
assert all(spec["limit"] >= 0 for spec in budget.values())