@@ -0,0 +1,107 @@
1 + from __future__ import annotations
2 +
3 + import argparse
4 + import ast
5 + import json
6 + import subprocess
7 + import sys
8 + from pathlib import Path
9 +
10 + ROOT = Path(__file__).resolve().parents[1]
11 + SCHEMA = "asw.quality-result/v1"
12 + BRANCH_NODES = (ast.If, ast.For, ast.AsyncFor, ast.While, ast.Try, ast.BoolOp)
13 + if hasattr(ast, "Match"):
14 + BRANCH_NODES += (ast.Match,)
15 +
16 +
17 + def write_result(path: str, payload: dict) -> None:
18 + destination = ROOT / path
19 + destination.parent.mkdir(parents=True, exist_ok=True)
20 + destination.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
21 +
22 +
23 + def source_files() -> list[Path]:
24 + ignored = {".git", "dist", "tools", "tests", "__pycache__"}
25 + return sorted(path for path in ROOT.rglob("*.py") if not any(part in ignored for part in path.relative_to(ROOT).parts))
26 +
27 +
28 + def static_analysis() -> dict:
29 + issues: list[dict] = []
30 + files = source_files()
31 + lines = functions = maximum_complexity = 0
32 + for path in files:
33 + relative = str(path.relative_to(ROOT))
34 + source = path.read_text(encoding="utf-8")
35 + lines += len(source.splitlines())
36 + try:
37 + tree = ast.parse(source, filename=relative)
38 + except SyntaxError as error:
39 + issues.append({"rule": "PY-SYNTAX", "severity": "CRITICAL", "path": relative, "line": error.lineno or 0, "message": error.msg})
40 + continue
41 + for node in ast.walk(tree):
42 + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
43 + functions += 1
44 + complexity = 1 + sum(isinstance(item, BRANCH_NODES) for item in ast.walk(node))
45 + maximum_complexity = max(maximum_complexity, complexity)
46 + if complexity > 12:
47 + issues.append({"rule": "PY-COMPLEXITY", "severity": "MEDIUM", "path": relative, "line": node.lineno, "message": f"{node.name} complexity {complexity} exceeds warning threshold 12"})
48 + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in {"eval", "exec"}:
49 + issues.append({"rule": "PY-DYNAMIC-EXEC", "severity": "HIGH", "path": relative, "line": node.lineno, "message": f"{node.func.id}() is not allowed in the reference policy"})
50 + blocking = [item for item in issues if item["severity"] in {"CRITICAL", "HIGH"}]
51 + return {
52 + "schemaVersion": SCHEMA,
53 + "kind": "static-analysis",
54 + "adapter": "python-ast-policy",
55 + "language": "python",
56 + "state": "PASSED" if not blocking else "FAILED",
57 + "metrics": {"files": len(files), "lines": lines, "functions": functions, "maximumComplexity": maximum_complexity, "blockingIssues": len(blocking), "warnings": len(issues) - len(blocking)},
58 + "issues": issues,
59 + }
60 +
61 +
62 + def command_result(command: list[str]) -> dict:
63 + result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False)
64 + return {"command": " ".join(command), "exitCode": result.returncode, "stdout": result.stdout[-4000:], "stderr": result.stderr[-4000:]}
65 +
66 +
67 + def dynamic_analysis() -> dict:
68 + test_cases = 0
69 + for path in (ROOT / "tests").glob("test_*.py") if (ROOT / "tests").exists() else []:
70 + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
71 + test_cases += sum(isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test_") for node in ast.walk(tree))
72 + commands = [command_result([sys.executable, "-m", "unittest", "discover", "-s", "tests", "-v"])]
73 + if (ROOT / "run.py").exists() and (ROOT / "examples" / "input.json").exists():
74 + commands.append(command_result([sys.executable, "run.py", "examples/input.json"]))
75 + passed = sum(item["exitCode"] == 0 for item in commands)
76 + rate = passed / len(commands) if commands else 0
77 + return {
78 + "schemaVersion": SCHEMA,
79 + "kind": "dynamic-analysis",
80 + "adapter": "unittest-scenario",
81 + "language": "python",
82 + "state": "PASSED" if rate == 1 else "FAILED",
83 + "metrics": {"testCases": test_cases, "commands": len(commands), "passedCommands": passed, "passRate": rate},
84 + "commands": commands,
85 + }
86 +
87 +
88 + def summary() -> dict:
89 + static = json.loads((ROOT / "dist/quality/static-analysis.json").read_text(encoding="utf-8"))
90 + dynamic = json.loads((ROOT / "dist/quality/dynamic-analysis.json").read_text(encoding="utf-8"))
91 + return {
92 + "schemaVersion": "asw.quality-summary/v1",
93 + "language": "python",
94 + "state": "PASSED" if static["state"] == dynamic["state"] == "PASSED" else "FAILED",
95 + "results": {"staticAnalysis": static["state"], "dynamicAnalysis": dynamic["state"]},
96 + "metrics": {"blockingIssues": static["metrics"]["blockingIssues"], "testCases": dynamic["metrics"]["testCases"], "passRate": dynamic["metrics"]["passRate"]},
97 + }
98 +
99 +
100 + parser = argparse.ArgumentParser()
101 + parser.add_argument("mode", choices=["static", "dynamic", "summary"])
102 + parser.add_argument("--output", required=True)
103 + arguments = parser.parse_args()
104 + result = static_analysis() if arguments.mode == "static" else dynamic_analysis() if arguments.mode == "dynamic" else summary()
105 + write_result(arguments.output, result)
106 + print(json.dumps({"kind": result.get("kind", "summary"), "state": result["state"], "metrics": result.get("metrics", {})}, ensure_ascii=False))
107 + raise SystemExit(0 if result["state"] == "PASSED" else 1)