christianmanivong
c3151d62fa
napalm-hpe-aruba-procurve ships napalm_procurve; guessed from its name, the ratchet measured nothing and wrote an empty baseline without a word.
237 lines
9.3 KiB
Python
237 lines
9.3 KiB
Python
"""Measure a driver against the limits of STANDARDS.md §5, and hold the line (§6).
|
|
|
|
The limits:
|
|
|
|
- function length 50 lines, cyclomatic complexity 10
|
|
- file length 1000 lines (hard maximum)
|
|
- public functions annotated
|
|
- no module exempt from strict mypy
|
|
- the coverage floor in ``[tool.coverage.report] fail_under``
|
|
|
|
Existing code exceeds them in places. That is recorded in
|
|
``tests/quality_baseline.json``: an entry may shrink but not grow, and
|
|
everything else keeps the limit. A baseline entry that was fixed is reported as
|
|
stale, so progress is written down and holds from then on.
|
|
|
|
Code is measured from the AST, not with ruff or radon, so the numbers do not
|
|
move with a tool's version. Complexity is mccabe-style: 1 plus one per
|
|
``if``/``elif``, loop, ``except``, ternary, ``match`` case, comprehension
|
|
``for`` and ``if``, and each extra operand of ``and``/``or``. A nested
|
|
function counts for itself, not for the function around it.
|
|
|
|
Ported from netOrk's scripts/quality_ratchet.py (#402).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
import pathlib
|
|
import tomllib
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
LIMITS = {
|
|
"function_lines": 50,
|
|
"function_complexity": 10,
|
|
"python_file_lines": 1000,
|
|
}
|
|
|
|
BASELINE = pathlib.Path("tests") / "quality_baseline.json"
|
|
|
|
#: mypy settings that strict = true turns on; an override setting one to False loosens it.
|
|
_STRICT_FLAGS = frozenset(
|
|
{
|
|
"strict",
|
|
"check_untyped_defs",
|
|
"disallow_any_generics",
|
|
"disallow_incomplete_defs",
|
|
"disallow_untyped_calls",
|
|
"disallow_untyped_decorators",
|
|
"disallow_untyped_defs",
|
|
"extra_checks",
|
|
"no_implicit_reexport",
|
|
"strict_equality",
|
|
"warn_return_any",
|
|
}
|
|
)
|
|
|
|
_FUNCTION = (ast.FunctionDef, ast.AsyncFunctionDef)
|
|
FunctionNode = ast.FunctionDef | ast.AsyncFunctionDef
|
|
|
|
|
|
def complexity(func: ast.AST) -> int:
|
|
score = 1
|
|
stack = list(ast.iter_child_nodes(func))
|
|
while stack:
|
|
node = stack.pop()
|
|
if isinstance(node, (*_FUNCTION, ast.ClassDef, ast.Lambda)):
|
|
continue # counts for itself
|
|
if isinstance(
|
|
node, (ast.If, ast.For, ast.AsyncFor, ast.While, ast.IfExp, ast.ExceptHandler)
|
|
):
|
|
score += 1
|
|
elif isinstance(node, ast.BoolOp):
|
|
score += len(node.values) - 1
|
|
elif isinstance(node, ast.comprehension):
|
|
score += 1 + len(node.ifs)
|
|
elif isinstance(node, ast.match_case):
|
|
score += 1
|
|
stack.extend(ast.iter_child_nodes(node))
|
|
return score
|
|
|
|
|
|
def _typed(func: FunctionNode, is_method: bool) -> bool:
|
|
args = [*func.args.posonlyargs, *func.args.args, *func.args.kwonlyargs]
|
|
if is_method and args and args[0].arg in ("self", "cls"):
|
|
args = args[1:]
|
|
extras = [a for a in (func.args.vararg, func.args.kwarg) if a is not None]
|
|
return func.returns is not None and all(a.annotation is not None for a in [*args, *extras])
|
|
|
|
|
|
def iter_python_functions(tree: ast.AST) -> Iterator[tuple[str, FunctionNode, bool, bool]]:
|
|
"""Every function in *tree*, outermost first: its qualified name, its node,
|
|
whether it sits directly in a class, and whether it is public."""
|
|
|
|
def visit(
|
|
node: ast.AST, prefix: str, in_class: bool, public_scope: bool
|
|
) -> Iterator[tuple[str, FunctionNode, bool, bool]]:
|
|
for child in ast.iter_child_nodes(node):
|
|
if isinstance(child, ast.ClassDef):
|
|
yield from visit(
|
|
child,
|
|
f"{prefix}{child.name}.",
|
|
True,
|
|
public_scope and not child.name.startswith("_"),
|
|
)
|
|
elif isinstance(child, _FUNCTION):
|
|
name = f"{prefix}{child.name}"
|
|
yield name, child, in_class, public_scope and not child.name.startswith("_")
|
|
yield from visit(child, f"{name}.", False, False)
|
|
elif isinstance(child, ast.stmt):
|
|
# a def under if/with/try/for belongs to the same scope
|
|
yield from visit(child, prefix, in_class, public_scope)
|
|
|
|
yield from visit(tree, "", False, True)
|
|
|
|
|
|
def python_functions_in(source: str, relpath: str) -> dict[str, dict[str, Any]]:
|
|
"""Every function in *source*: lines, complexity, whether it is typed and public."""
|
|
return {
|
|
f"{relpath}::{name}": {
|
|
"lines": (node.end_lineno or node.lineno) - node.lineno + 1,
|
|
"complexity": complexity(node),
|
|
"typed": _typed(node, in_class),
|
|
"public": public,
|
|
}
|
|
for name, node, in_class, public in iter_python_functions(ast.parse(source))
|
|
}
|
|
|
|
|
|
def relaxed_mypy_modules(pyproject: dict[str, Any]) -> list[str]:
|
|
"""Modules a ``[[tool.mypy.overrides]]`` entry exempts from strict checking."""
|
|
out: set[str] = set()
|
|
for override in pyproject.get("tool", {}).get("mypy", {}).get("overrides", []):
|
|
loosens = override.get("ignore_errors") is True or any(
|
|
override.get(flag) is False for flag in _STRICT_FLAGS
|
|
)
|
|
if loosens:
|
|
modules = override.get("module", [])
|
|
out.update([modules] if isinstance(modules, str) else modules)
|
|
return sorted(out)
|
|
|
|
|
|
def coverage_floor(pyproject: dict[str, Any]) -> float:
|
|
floor = pyproject.get("tool", {}).get("coverage", {}).get("report", {}).get("fail_under", 0)
|
|
return float(floor)
|
|
|
|
|
|
def _lines(source: str) -> int:
|
|
return source.count("\n") + (0 if source.endswith("\n") else 1)
|
|
|
|
|
|
def measure(repo: pathlib.Path, package: str) -> dict[str, Any]:
|
|
"""Today's excesses, keyed the way the baseline stores them."""
|
|
if not (repo / package).is_dir():
|
|
raise FileNotFoundError(
|
|
f"no package directory {package}/ to measure; name it in [tool.napalm-standard] package"
|
|
)
|
|
function_lines: dict[str, int] = {}
|
|
function_complexity: dict[str, int] = {}
|
|
python_file_lines: dict[str, int] = {}
|
|
untyped_public: list[str] = []
|
|
for path in sorted((repo / package).rglob("*.py")):
|
|
rel = path.relative_to(repo).as_posix()
|
|
source = path.read_text()
|
|
if _lines(source) > LIMITS["python_file_lines"]:
|
|
python_file_lines[rel] = _lines(source)
|
|
for key, f in python_functions_in(source, rel).items():
|
|
if f["lines"] > LIMITS["function_lines"]:
|
|
function_lines[key] = f["lines"]
|
|
if f["complexity"] > LIMITS["function_complexity"]:
|
|
function_complexity[key] = f["complexity"]
|
|
if f["public"] and not f["typed"]:
|
|
untyped_public.append(key)
|
|
pyproject = tomllib.loads((repo / "pyproject.toml").read_text())
|
|
return {
|
|
"function_lines": function_lines,
|
|
"function_complexity": function_complexity,
|
|
"python_file_lines": python_file_lines,
|
|
"untyped_public": sorted(untyped_public),
|
|
"mypy_relaxed": relaxed_mypy_modules(pyproject),
|
|
"coverage_floor": coverage_floor(pyproject),
|
|
}
|
|
|
|
|
|
def problems(measured: dict[str, Any], baseline: dict[str, Any]) -> list[str]:
|
|
"""What is new over a limit, or grew past what the baseline allows."""
|
|
out = []
|
|
for metric in LIMITS:
|
|
allowed = baseline.get(metric, {})
|
|
for key, value in sorted(measured[metric].items()):
|
|
if key not in allowed:
|
|
out.append(f"{metric}: {key} = {value} (new)")
|
|
elif value > allowed[key]:
|
|
out.append(f"{metric}: {key} = {value} (recorded {allowed[key]})")
|
|
for metric in ("untyped_public", "mypy_relaxed"):
|
|
new = sorted(set(measured[metric]) - set(baseline.get(metric, [])))
|
|
out.extend(f"{metric}: {key} (new)" for key in new)
|
|
recorded_floor = baseline.get("coverage_floor", 0)
|
|
if measured["coverage_floor"] < recorded_floor:
|
|
out.append(f"coverage_floor: {measured['coverage_floor']:g} (recorded {recorded_floor:g})")
|
|
return out
|
|
|
|
|
|
def stale_entries(measured: dict[str, Any], baseline: dict[str, Any]) -> list[str]:
|
|
"""Baseline entries that are fixed, shrunk or gone: the ratchet can tighten."""
|
|
out = []
|
|
for metric in LIMITS:
|
|
for key, value in sorted(baseline.get(metric, {}).items()):
|
|
now = measured[metric].get(key)
|
|
if now is None or now < value:
|
|
out.append(f"{metric}: {key}")
|
|
for metric in ("untyped_public", "mypy_relaxed"):
|
|
gone = sorted(set(baseline.get(metric, [])) - set(measured[metric]))
|
|
out.extend(f"{metric}: {key}" for key in gone)
|
|
recorded_floor = baseline.get("coverage_floor")
|
|
if recorded_floor is not None and measured["coverage_floor"] > recorded_floor:
|
|
out.append(f"coverage_floor: {recorded_floor:g} -> {measured['coverage_floor']:g}")
|
|
return out
|
|
|
|
|
|
def run(repo: pathlib.Path, package: str, write: bool = False) -> tuple[int, str]:
|
|
"""Check the repo against its baseline, or (*write*) record today's state as it."""
|
|
measured = measure(repo, package)
|
|
path = repo / BASELINE
|
|
if write:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(measured, indent=1, sort_keys=True) + "\n")
|
|
return 0, f"wrote {BASELINE}"
|
|
baseline = json.loads(path.read_text()) if path.is_file() else {}
|
|
lines = problems(measured, baseline)
|
|
lines += [
|
|
f"stale: {entry} (run napalm-standard ratchet --write)"
|
|
for entry in stale_entries(measured, baseline)
|
|
]
|
|
return (1, "\n".join(lines)) if lines else (0, "within the baseline")
|