Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 35 additions & 8 deletions scripts/budget_ratchet_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,17 @@
content at the merge-base with the target branch and fails (exits 1, red) if:

* a rule's `limit` went up,
* a rule was dropped from a budget (its ceiling effectively became infinite), or
* a rule was dropped from a budget (its ceiling effectively became infinite) while
its checker still emits it, or
* an entire budget file was deleted.

New rules and lowered/equal limits are fine. So is a rule that graduated: once a
paired config (ruff.toml for the ruff-strict budget) selects the rule outright it
hard-fails at the first violation, which is stricter than any ceiling the budget
could hold, so dropping its entry tightens the guard rather than removing it.
Likewise a retired rule: once the paired checker (check_test_quality.py for the
test-quality budget) no longer emits a code, its entry has no ceiling left to
loosen.

This is deliberately NOT a gating check. It should turn the run red so that a
loosening is impossible to miss in review, but it must stay OUT of the
Expand All @@ -29,11 +33,12 @@
from __future__ import annotations

import argparse
import importlib.util
import json
import subprocess
import sys
from pathlib import Path
from types import MappingProxyType
from types import MappingProxyType, ModuleType
from typing import Final, NamedTuple

if sys.version_info >= (3, 11):
Expand All @@ -49,6 +54,7 @@
"test-quality-budget.json",
)
GRADUATION_CONFIGS = MappingProxyType({"ruff-strict-budget.json": "ruff.toml"})
RETIREMENT_SOURCES = MappingProxyType({"test-quality-budget.json": "check_test_quality"})


class Regression(NamedTuple):
Expand Down Expand Up @@ -139,20 +145,40 @@ def graduated_selectors(rel: str) -> tuple[str, ...]:
)


def _load_script(name: str) -> ModuleType:
if name in sys.modules:
return sys.modules[name]
spec: Final = importlib.util.spec_from_file_location(name, REPO_ROOT / "scripts" / f"{name}.py")
assert spec is not None and spec.loader is not None
module: Final = importlib.util.module_from_spec(spec)
sys.modules[name] = module
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Resolved
spec.loader.exec_module(module)
return module


def retired_rules(rel: str, base: dict[str, object]) -> frozenset[str]:
"""Rules in the base budget that the paired checker can no longer emit, so there is no ceiling to loosen."""
source: Final = RETIREMENT_SOURCES.get(rel)
if source is None:
return frozenset()
return frozenset(_limits(base)) - _load_script(source).RULE_CODES


def _regression_detail(
rule: str,
base_limits: dict[str, int],
head_limits: dict[str, int],
graduated: tuple[str, ...],
retired: frozenset[str] = frozenset(),
) -> str | None:
"""Why `rule` regressed vs base, or None when it held flat, fell, or graduated.
"""Why `rule` regressed vs base, or None when it held flat, fell, or left the budget legitimately.

A dropped rule is terminal unless it graduated; otherwise the only loosening
left is a raised limit.
A dropped rule is terminal unless it graduated or retired; otherwise the only
loosening left is a raised limit.
"""
base_limit = base_limits[rule]
if rule not in head_limits:
if graduated and rule.startswith(graduated):
if rule in retired or (graduated and rule.startswith(graduated)):
return None
return f"rule dropped (limit {base_limit} -> removed)"
if head_limits[rule] > base_limit:
Expand All @@ -165,6 +191,7 @@ def regressions_for(
base: dict | None,
head: dict | None,
graduated: tuple[str, ...] = (),
retired: frozenset[str] = frozenset(),
) -> list[Regression]:
if base is None:
return [] # new budget file: nothing to ratchet against yet
Expand All @@ -175,7 +202,7 @@ def regressions_for(
return [
Regression(rel, rule, detail)
for rule in sorted(base_limits)
if (detail := _regression_detail(rule, base_limits, head_limits, graduated)) is not None
if (detail := _regression_detail(rule, base_limits, head_limits, graduated, retired)) is not None
]


Expand Down Expand Up @@ -209,7 +236,7 @@ def main() -> int:
print(f"skip {rel}: new file (no base at {base_ref} to ratchet against)")
continue
checked.append(rel)
regressions.extend(regressions_for(rel, base, head, graduated_selectors(rel)))
regressions.extend(regressions_for(rel, base, head, graduated_selectors(rel), retired_rules(rel, base)))

if regressions:
print(
Expand Down
4 changes: 4 additions & 0 deletions scripts/check_test_quality.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,10 @@ class that defines no `__init__`.
SUBPROCESS_SPAWNS: Final = frozenset(("run", "Popen", "check_output", "check_call", "call"))
INTERPRETER_ISOLATION_FLAGS: Final = frozenset(("-I", "-P"))

RULE_CODES: Final = frozenset((
"TQ000", "TQ001", "TQ002", "TQ003", "TQ004", "TQ005", "TQ006", "TQ007", "TQ009",
))

CREDENTIAL_NAME_RE: Final = re.compile(
r"(?:API_KEY|_KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|DATABASE_URL|ACCESS_KEY_ID)$"
)
Expand Down
30 changes: 30 additions & 0 deletions tests/test_litellm/test_budget_ratchet_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,36 @@ def test_graduation_never_excuses_a_raised_limit():
assert "0 -> 7" in regs[0].detail


def test_dropped_rule_the_checker_retired_is_clean():
base = {"TQ008": _spec_of(10993)}
assert ratchet.regressions_for("b.json", base, {}, retired=frozenset({"TQ008"})) == []


def test_dropped_rule_the_checker_still_emits_is_a_regression():
base = {"TQ001": _spec_of(5), "TQ008": _spec_of(10993)}
regs = ratchet.regressions_for("b.json", base, {}, retired=frozenset({"TQ008"}))
assert [r.rule for r in regs] == ["TQ001"]
assert "dropped" in regs[0].detail


def test_retirement_never_excuses_a_raised_limit():
base = {"TQ008": _spec_of(0)}
regs = ratchet.regressions_for("b.json", base, {"TQ008": _spec_of(7)}, retired=frozenset({"TQ008"}))
assert [r.rule for r in regs] == ["TQ008"]
assert "0 -> 7" in regs[0].detail


def test_retired_rules_come_from_the_paired_checker():
base = {"TQ001": _spec_of(5), "TQ008": _spec_of(10993)}
assert ratchet.retired_rules("test-quality-budget.json", base) == frozenset({"TQ008"})


def test_budgets_without_a_paired_checker_never_retire():
base = {"TQ008": _spec_of(1)}
for rel in ("ruff-strict-budget.json", "type-discipline-budget.json", "basedpyright-code-budget.json"):
assert ratchet.retired_rules(rel, base) == frozenset()


def test_graduated_selectors_come_from_the_paired_ruff_config():
selectors = ratchet.graduated_selectors("ruff-strict-budget.json")
assert "UP006" in selectors
Expand Down
40 changes: 40 additions & 0 deletions tests/test_litellm/test_check_test_quality.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
import subprocess
import sys
from pathlib import Path
from types import MappingProxyType
from typing import Final

import pytest

Expand Down Expand Up @@ -612,6 +614,44 @@ def test_a_fanned_out_run_reports_each_generated_file_exactly_once(tmp_path):
assert all(" TQ001 " in line for line in reported)


_VIOLATING_SNIPPETS: Final = MappingProxyType(
{
"TQ000": ("test_snippet.py", "def test_broken(:\n pass\n"),
"TQ001": ("test_snippet.py", "def test_nothing():\n compute()\n"),
"TQ002": (
"test_snippet.py",
"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",
),
"TQ003": ("test_snippet.py", "import sys\n\nsys.path.insert(0, '..')\n"),
"TQ004": ("test_snippet.py", "import os\n\nos.environ['KEY'] = 'v'\n"),
"TQ005": ("test_snippet.py", "import litellm\n\nlitellm.drop_params = True\n"),
"TQ006": ("test_snippet.py", _DIRECT_GATE),
"TQ007": ("conftest.py", _SNAPSHOT_CONFTEST),
"TQ009": (
"test_snippet.py",
'import subprocess, sys\nsubprocess.run([sys.executable, "-c", "pass"])\n',
),
}
)


def test_rule_codes_match_every_code_the_checker_emits(tmp_path):
emitted = frozenset(
v.code
for name, source in _VIOLATING_SNIPPETS.values()
for v in checker.check_file(_written(tmp_path, source, name))
)
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
Resolved
for code, (name, source) in _VIOLATING_SNIPPETS.items():
assert code in [v.code for v in checker.check_file(_written(tmp_path, source, name))], code
assert emitted == checker.RULE_CODES


def test_sys_executable_child_without_isolation_flag_is_flagged(tmp_path):
source = 'import subprocess, sys\nsubprocess.run([sys.executable, "-c", "pass"])\n'
assert _codes(tmp_path, source) == ["TQ009"]
Expand Down
Loading