@@ -1,14 +1,18 @@
1 1 from __future__ import annotations
2 2
3 3 import argparse
4 − import ast
5 4 import json
6 5 import subprocess
7 − import sys
8 6 from pathlib import Path
9 7
10 8 ROOT = Path(__file__).resolve().parents[1]
11 − SCHEMA = "asw.quality-result/v1"
9 + LANGUAGE = "cpp"
10 + STATIC_ADAPTER = "g++-warning-policy"
11 + DYNAMIC_ADAPTER = "g++-test-scenario"
12 + COMPILER = "g++"
13 + STANDARD = "c++17"
14 + EXTENSION = "cpp"
15 + BANNED = [["NATIVE-GETS","gets("],["NATIVE-STRCPY","strcpy("],["NATIVE-SPRINTF","sprintf("],["NATIVE-SYSTEM","system("]]
12 16
13 17
14 18 def write_result(path: str, payload: dict) -> None:
@@ -17,81 +21,60 @@ def write_result(path: str, payload: dict) -> None:
17 21 destination.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
18 22
19 23
24 + def run(command: list[str]) -> dict:
25 + completed = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False)
26 + return {"command": " ".join(command), "exitCode": completed.returncode, "stdout": completed.stdout[-4000:], "stderr": completed.stderr[-4000:]}
27 +
28 +
20 29 def source_files() -> list[Path]:
21 − ignored = {".git", "dist", "tools", "tests", "__pycache__"}
22 − return sorted(path for path in ROOT.rglob("*.py") if not any(part in ignored for part in path.relative_to(ROOT).parts))
30 + patterns = ["*.ts", "*.mjs"] if LANGUAGE == "typescript" else [f"*.{EXTENSION}", "*.h", "*.hpp"]
31 + files: set[Path] = set()
32 + for pattern in patterns:
33 + files.update((ROOT / "src").rglob(pattern))
34 + if (ROOT / "include").exists():
35 + files.update((ROOT / "include").rglob(pattern))
36 + return sorted(files)
23 37
24 38
25 39 def static_analysis() -> dict:
26 40 issues: list[dict] = []
27 41 files = source_files()
28 − lines = functions = maximum_complexity = 0
42 + lines = 0
29 43 for path in files:
30 − relative = str(path.relative_to(ROOT))
31 44 source = path.read_text(encoding="utf-8")
32 45 lines += len(source.splitlines())
33 − try:
34 − tree = ast.parse(source, filename=relative)
35 − except SyntaxError as error:
36 − issues.append({"rule": "PY-SYNTAX", "severity": "CRITICAL", "path": relative, "line": error.lineno or 0, "message": error.msg})
37 − continue
38 − for node in ast.walk(tree):
39 − if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
40 − functions += 1
41 − complexity = 1 + sum(isinstance(item, (ast.If, ast.For, ast.AsyncFor, ast.While, ast.Try, ast.BoolOp, ast.Match)) for item in ast.walk(node))
42 − maximum_complexity = max(maximum_complexity, complexity)
43 − if complexity > 12:
44 − issues.append({"rule": "PY-COMPLEXITY", "severity": "MEDIUM", "path": relative, "line": node.lineno, "message": f"{node.name} complexity {complexity} exceeds warning threshold 12"})
45 − if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in {"eval", "exec"}:
46 − 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"})
47 − blocking = [item for item in issues if item["severity"] in {"CRITICAL", "HIGH"}]
48 − return {
49 − "schemaVersion": SCHEMA,
50 − "kind": "static-analysis",
51 − "adapter": "python-ast-policy",
52 − "language": "python",
53 − "state": "PASSED" if not blocking else "FAILED",
54 − "metrics": {"files": len(files), "lines": lines, "functions": functions, "maximumComplexity": maximum_complexity, "blockingIssues": len(blocking), "warnings": len(issues) - len(blocking)},
55 − "issues": issues,
56 − }
57 −
58 −
59 − def command_result(command: list[str]) -> dict:
60 − result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False)
61 − return {"command": " ".join(command), "exitCode": result.returncode, "stdout": result.stdout[-4000:], "stderr": result.stderr[-4000:]}
46 + for rule, token in BANNED:
47 + if token in source:
48 + issues.append({"rule": rule, "severity": "HIGH", "path": str(path.relative_to(ROOT)), "line": source[:source.index(token)].count("\n") + 1, "message": f"forbidden token: {token}"})
49 + if LANGUAGE == "typescript":
50 + command = run(["node", "--check", "src/component.mjs"])
51 + else:
52 + command = run([COMPILER, f"-std={STANDARD}", "-Wall", "-Wextra", "-Wpedantic", "-Werror", "-Iinclude", "-fsyntax-only", f"src/component.{EXTENSION}"])
53 + if command["exitCode"] != 0:
54 + issues.append({"rule": "COMPILE-SYNTAX", "severity": "CRITICAL", "path": f"src/component.{EXTENSION}", "line": 0, "message": command["stderr"][-1000:]})
55 + blocking = [issue for issue in issues if issue["severity"] in {"CRITICAL", "HIGH"}]
56 + return {"schemaVersion": "asw.quality-result/v1", "kind": "static-analysis", "adapter": STATIC_ADAPTER, "language": LANGUAGE, "state": "PASSED" if not blocking else "FAILED", "metrics": {"files": len(files), "lines": lines, "blockingIssues": len(blocking), "warnings": len(issues) - len(blocking)}, "issues": issues, "commands": [command]}
62 57
63 58
64 59 def dynamic_analysis() -> dict:
65 − test_cases = 0
66 − for path in (ROOT / "tests").glob("test_*.py") if (ROOT / "tests").exists() else []:
67 − tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
68 − test_cases += sum(isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test_") for node in ast.walk(tree))
69 − commands = [command_result([sys.executable, "-m", "unittest", "discover", "-s", "tests", "-v"])]
70 − if (ROOT / "run.py").exists() and (ROOT / "examples" / "input.json").exists():
71 − commands.append(command_result([sys.executable, "run.py", "examples/input.json"]))
60 + output = ROOT / "dist" / "component_test"
61 + output.parent.mkdir(parents=True, exist_ok=True)
62 + if LANGUAGE == "typescript":
63 + commands = [run(["node", "--test", "tests/component.test.mjs"])]
64 + else:
65 + compile_command = [COMPILER, f"-std={STANDARD}", "-Wall", "-Wextra", "-Wpedantic", "-Werror", "-Iinclude", f"src/component.{EXTENSION}", f"tests/component_test.{EXTENSION}", "-o", str(output)]
66 + commands = [run(compile_command)]
67 + if commands[0]["exitCode"] == 0:
68 + commands.append(run([str(output)]))
72 69 passed = sum(item["exitCode"] == 0 for item in commands)
73 70 rate = passed / len(commands) if commands else 0
74 − return {
75 − "schemaVersion": SCHEMA,
76 − "kind": "dynamic-analysis",
77 − "adapter": "unittest-scenario",
78 − "language": "python",
79 − "state": "PASSED" if rate == 1 else "FAILED",
80 − "metrics": {"testCases": test_cases, "commands": len(commands), "passedCommands": passed, "passRate": rate},
81 − "commands": commands,
82 − }
71 + return {"schemaVersion": "asw.quality-result/v1", "kind": "dynamic-analysis", "adapter": DYNAMIC_ADAPTER, "language": LANGUAGE, "state": "PASSED" if rate == 1 else "FAILED", "metrics": {"testCases": 1, "commands": len(commands), "passedCommands": passed, "passRate": rate}, "commands": commands}
83 72
84 73
85 74 def summary() -> dict:
86 75 static = json.loads((ROOT / "dist/quality/static-analysis.json").read_text(encoding="utf-8"))
87 76 dynamic = json.loads((ROOT / "dist/quality/dynamic-analysis.json").read_text(encoding="utf-8"))
88 − return {
89 − "schemaVersion": "asw.quality-summary/v1",
90 − "language": "python",
91 − "state": "PASSED" if static["state"] == dynamic["state"] == "PASSED" else "FAILED",
92 − "results": {"staticAnalysis": static["state"], "dynamicAnalysis": dynamic["state"]},
93 − "metrics": {"blockingIssues": static["metrics"]["blockingIssues"], "testCases": dynamic["metrics"]["testCases"], "passRate": dynamic["metrics"]["passRate"]},
94 − }
77 + return {"schemaVersion": "asw.quality-summary/v1", "language": LANGUAGE, "state": "PASSED" if static["state"] == dynamic["state"] == "PASSED" else "FAILED", "results": {"staticAnalysis": static["state"], "dynamicAnalysis": dynamic["state"]}, "metrics": {"blockingIssues": static["metrics"]["blockingIssues"], "testCases": dynamic["metrics"]["testCases"], "passRate": dynamic["metrics"]["passRate"]}}
95 78
96 79
97 80 parser = argparse.ArgumentParser()