maintainability_ratchet.py (11520B)
1 #!/usr/bin/env python3 2 """Fail-closed Swift/Python size and Python complexity ratchet.""" 3 4 from __future__ import annotations 5 6 import argparse 7 import ast 8 import json 9 import sys 10 from pathlib import Path 11 from typing import Any 12 13 MAX_BASELINE_BYTES = 256 * 1024 14 BASELINE_PATH = Path("test-fixtures/maintainability-baseline.v1.json") 15 SWIFT_ROOTS = (Path("Tera"), Path("TeraTests"), Path("TeraUITests")) 16 PYTHON_ROOT = Path("scripts") 17 EXCLUDED_SWIFT_ROOT = Path("Tera/Generated") 18 19 20 class MaintainabilityError(Exception): 21 """Stable maintainability-policy rejection.""" 22 23 24 class _FunctionComplexity(ast.NodeVisitor): 25 def __init__(self) -> None: 26 self.value = 1 27 28 def visit_FunctionDef(self, node: ast.FunctionDef) -> None: 29 del node 30 31 def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: 32 del node 33 34 def visit_Lambda(self, node: ast.Lambda) -> None: 35 del node 36 37 def visit_If(self, node: ast.If) -> None: 38 self.value += 1 39 self.generic_visit(node) 40 41 def visit_IfExp(self, node: ast.IfExp) -> None: 42 self.value += 1 43 self.generic_visit(node) 44 45 def visit_For(self, node: ast.For) -> None: 46 self.value += 1 47 self.generic_visit(node) 48 49 def visit_AsyncFor(self, node: ast.AsyncFor) -> None: 50 self.value += 1 51 self.generic_visit(node) 52 53 def visit_While(self, node: ast.While) -> None: 54 self.value += 1 55 self.generic_visit(node) 56 57 def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None: 58 self.value += 1 59 self.generic_visit(node) 60 61 def visit_BoolOp(self, node: ast.BoolOp) -> None: 62 self.value += max(0, len(node.values) - 1) 63 self.generic_visit(node) 64 65 def visit_Match(self, node: ast.Match) -> None: 66 self.value += len(node.cases) 67 self.generic_visit(node) 68 69 def visit_comprehension(self, node: ast.comprehension) -> None: 70 self.value += 1 + len(node.ifs) 71 self.generic_visit(node) 72 73 74 class _DefinitionCollector(ast.NodeVisitor): 75 def __init__(self, path: str) -> None: 76 self.path = path 77 self.scope: list[str] = [] 78 self.values: dict[str, int] = {} 79 80 def visit_ClassDef(self, node: ast.ClassDef) -> None: 81 self.scope.append(node.name) 82 self.generic_visit(node) 83 self.scope.pop() 84 85 def visit_FunctionDef(self, node: ast.FunctionDef) -> None: 86 self._record(node) 87 88 def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: 89 self._record(node) 90 91 def _record(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None: 92 name = ".".join((*self.scope, node.name)) 93 key = f"{self.path}:{name}" 94 if key in self.values: 95 raise MaintainabilityError("Python function identity is duplicated") 96 counter = _FunctionComplexity() 97 for statement in node.body: 98 counter.visit(statement) 99 self.values[key] = counter.value 100 self.scope.append(node.name) 101 for statement in node.body: 102 self.visit(statement) 103 self.scope.pop() 104 105 106 def _read_regular(path: Path, maximum: int = MAX_BASELINE_BYTES) -> bytes: 107 try: 108 if path.is_symlink() or not path.is_file(): 109 raise MaintainabilityError("maintainability input is not a regular file") 110 size = path.stat().st_size 111 if size < 1 or size > maximum: 112 raise MaintainabilityError("maintainability input exceeds its byte bound") 113 value = path.read_bytes() 114 except OSError as error: 115 raise MaintainabilityError("maintainability input cannot be read") from error 116 if len(value) != size: 117 raise MaintainabilityError("maintainability input changed while reading") 118 return value 119 120 121 def _load_baseline(repo_root: Path) -> dict[str, Any]: 122 try: 123 value = json.loads(_read_regular(repo_root / BASELINE_PATH)) 124 except (UnicodeDecodeError, json.JSONDecodeError) as error: 125 raise MaintainabilityError("maintainability baseline is malformed") from error 126 expected = { 127 "schema", 128 "schema_version", 129 "source_revision", 130 "thresholds", 131 "swift_file_exception", 132 "python_file_exception", 133 "python_complexity_exception", 134 "bounded_module", 135 } 136 if not isinstance(value, dict) or set(value) != expected: 137 raise MaintainabilityError("maintainability baseline fields differ") 138 if ( 139 value["schema"] != "radroots.ios.maintainability-baseline.v1" 140 or value["schema_version"] != 1 141 or value["source_revision"] != "c63002bcc4d3f6656e93aabe4fca6bd771376629" 142 ): 143 raise MaintainabilityError("maintainability baseline identity differs") 144 return value 145 146 147 def _source_files(repo_root: Path) -> tuple[list[Path], list[Path]]: 148 swift = sorted( 149 path 150 for root in SWIFT_ROOTS 151 for path in (repo_root / root).rglob("*.swift") 152 if not path.is_symlink() 153 and not path.relative_to(repo_root).is_relative_to(EXCLUDED_SWIFT_ROOT) 154 ) 155 python = sorted( 156 path for path in (repo_root / PYTHON_ROOT).glob("*.py") if not path.is_symlink() 157 ) 158 if not swift or not python: 159 raise MaintainabilityError("maintainability source inventory is empty") 160 return swift, python 161 162 163 def _line_inventory(repo_root: Path, paths: list[Path]) -> dict[str, int]: 164 values: dict[str, int] = {} 165 for path in paths: 166 raw = _read_regular(path, 4 * 1024 * 1024) 167 try: 168 text = raw.decode("utf-8") 169 except UnicodeDecodeError as error: 170 raise MaintainabilityError("maintainability source is not UTF-8") from error 171 relative = path.relative_to(repo_root).as_posix() 172 values[relative] = len(text.splitlines()) 173 return values 174 175 176 def _python_complexity(repo_root: Path, paths: list[Path]) -> dict[str, int]: 177 values: dict[str, int] = {} 178 for path in paths: 179 relative = path.relative_to(repo_root).as_posix() 180 try: 181 tree = ast.parse(_read_regular(path, 4 * 1024 * 1024), filename=relative) 182 except (SyntaxError, ValueError) as error: 183 raise MaintainabilityError("Python source cannot be parsed") from error 184 collector = _DefinitionCollector(relative) 185 collector.visit(tree) 186 overlap = set(values) & set(collector.values) 187 if overlap: 188 raise MaintainabilityError("Python function identity is duplicated") 189 values.update(collector.values) 190 return values 191 192 193 def snapshot(repo_root: Path) -> dict[str, dict[str, int]]: 194 swift, python = _source_files(repo_root) 195 return { 196 "swift_lines": _line_inventory(repo_root, swift), 197 "python_lines": _line_inventory(repo_root, python), 198 "python_complexity": _python_complexity(repo_root, python), 199 } 200 201 202 def _closed_exception_map( 203 value: object, 204 *, 205 key_name: str, 206 ceiling_name: str, 207 ) -> dict[str, int]: 208 if not isinstance(value, list): 209 raise MaintainabilityError("maintainability exception inventory is invalid") 210 result: dict[str, int] = {} 211 for row in value: 212 if ( 213 not isinstance(row, dict) 214 or set(row) != {key_name, ceiling_name} 215 or not isinstance(row[key_name], str) 216 or not isinstance(row[ceiling_name], int) 217 or row[ceiling_name] < 1 218 or row[key_name] in result 219 ): 220 raise MaintainabilityError("maintainability exception row is invalid") 221 result[row[key_name]] = row[ceiling_name] 222 if list(result) != sorted(result): 223 raise MaintainabilityError("maintainability exceptions are not ordered") 224 return result 225 226 227 def _verify_metric( 228 observed: dict[str, int], 229 exceptions: dict[str, int], 230 threshold: int, 231 label: str, 232 ) -> None: 233 expected_exceptions = {key for key, value in observed.items() if value > threshold} 234 if set(exceptions) != expected_exceptions: 235 raise MaintainabilityError(f"{label} exception inventory differs") 236 for key, ceiling in exceptions.items(): 237 if ceiling <= threshold or observed[key] > ceiling: 238 raise MaintainabilityError(f"{label} metric regressed") 239 240 241 def _thresholds(baseline: dict[str, Any]) -> dict[str, int]: 242 value = baseline["thresholds"] 243 if ( 244 not isinstance(value, dict) 245 or set(value) 246 != {"swift_file_lines", "python_file_lines", "python_function_complexity"} 247 or value["swift_file_lines"] != 600 248 or value["python_file_lines"] != 800 249 or value["python_function_complexity"] != 10 250 ): 251 raise MaintainabilityError("maintainability thresholds differ") 252 return value 253 254 255 def _verify_observed_metrics( 256 baseline: dict[str, Any], 257 observed: dict[str, dict[str, int]], 258 thresholds: dict[str, int], 259 ) -> None: 260 specifications = ( 261 ( 262 "swift_file_exception", 263 "path", 264 "maximum_lines", 265 "swift_lines", 266 "swift_file_lines", 267 "Swift file", 268 ), 269 ( 270 "python_file_exception", 271 "path", 272 "maximum_lines", 273 "python_lines", 274 "python_file_lines", 275 "Python file", 276 ), 277 ( 278 "python_complexity_exception", 279 "function", 280 "maximum_complexity", 281 "python_complexity", 282 "python_function_complexity", 283 "Python complexity", 284 ), 285 ) 286 for baseline_key, identity, ceiling, metric, threshold, label in specifications: 287 exceptions = _closed_exception_map( 288 baseline[baseline_key], key_name=identity, ceiling_name=ceiling 289 ) 290 _verify_metric(observed[metric], exceptions, thresholds[threshold], label) 291 292 293 def _verify_bounded_modules( 294 baseline: dict[str, Any], 295 observed: dict[str, dict[str, int]], 296 thresholds: dict[str, int], 297 ) -> None: 298 modules = baseline["bounded_module"] 299 if not isinstance(modules, list) or modules != sorted(set(modules)): 300 raise MaintainabilityError("bounded module inventory differs") 301 all_lines = observed["swift_lines"] | observed["python_lines"] 302 for path in modules: 303 if path not in all_lines: 304 raise MaintainabilityError("bounded module is absent") 305 limit = thresholds[ 306 "swift_file_lines" if path.endswith(".swift") else "python_file_lines" 307 ] 308 if all_lines[path] > limit: 309 raise MaintainabilityError("bounded module exceeds its threshold") 310 311 312 def verify(repo_root: Path) -> None: 313 baseline = _load_baseline(repo_root) 314 thresholds = _thresholds(baseline) 315 observed = snapshot(repo_root) 316 _verify_observed_metrics(baseline, observed, thresholds) 317 _verify_bounded_modules(baseline, observed, thresholds) 318 319 320 def main() -> int: 321 parser = argparse.ArgumentParser() 322 parser.add_argument("command", choices=("verify", "snapshot")) 323 parser.add_argument( 324 "--repo-root", type=Path, default=Path(__file__).resolve().parent.parent 325 ) 326 arguments = parser.parse_args() 327 try: 328 if arguments.command == "snapshot": 329 print(json.dumps(snapshot(arguments.repo_root.resolve()), sort_keys=True)) 330 else: 331 verify(arguments.repo_root.resolve()) 332 print("maintainability ratchet verified") 333 except MaintainabilityError as error: 334 print(f"maintainability: {error}", file=sys.stderr) 335 return 1 336 return 0 337 338 339 if __name__ == "__main__": 340 raise SystemExit(main())