feat(R10): implement CI Quality Gates, Contributor Recipes, E2E Smoke Tests, and update docs
This commit is contained in:
@@ -0,0 +1,130 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Lines-of-Code (LOC) Quality Guard (EPIC R10 - CASAN Gate S).
|
||||
|
||||
Enforces the Single Responsibility Principle by ensuring that no production
|
||||
Python file in Clean Architecture packages exceeds the configured limit (400 LOC).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
# Ensure stdout handles UTF-8 on Windows consoles without codec crash
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Default target directories strictly subjected to the 400 LOC constraint
|
||||
DEFAULT_TARGET_DIRS = ["domain", "application", "infrastructure", "presentation"]
|
||||
DEFAULT_MAX_LINES = 400
|
||||
|
||||
|
||||
def count_file_lines(file_path: Path) -> int:
|
||||
"""Read a python file and return total physical line count."""
|
||||
try:
|
||||
content = file_path.read_text(encoding="utf-8", errors="ignore")
|
||||
return len(content.splitlines())
|
||||
except Exception as exc:
|
||||
print(f"[WARN] Failed to read {file_path}: {exc}", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
|
||||
def scan_directories(
|
||||
root_dir: Path, target_dirs: List[str], max_lines: int, verbose: bool = False
|
||||
) -> Tuple[int, List[Tuple[str, int]]]:
|
||||
"""Recursively scan target packages for files exceeding the maximum LOC limit.
|
||||
|
||||
Returns:
|
||||
A tuple of (total_files_scanned, list_of_violations_as_(relative_path, line_count))
|
||||
"""
|
||||
total_files = 0
|
||||
violations: List[Tuple[str, int]] = []
|
||||
|
||||
for target in target_dirs:
|
||||
dir_path = root_dir / target
|
||||
if not dir_path.is_dir():
|
||||
if verbose:
|
||||
print(f"[INFO] Skipping missing directory: {target}")
|
||||
continue
|
||||
|
||||
for current_root, _, files in os.walk(dir_path):
|
||||
for file_name in files:
|
||||
if not file_name.endswith(".py"):
|
||||
continue
|
||||
|
||||
full_path = Path(current_root) / file_name
|
||||
rel_path = full_path.relative_to(root_dir).as_posix()
|
||||
lines = count_file_lines(full_path)
|
||||
total_files += 1
|
||||
|
||||
if verbose:
|
||||
print(f" {rel_path}: {lines} lines")
|
||||
|
||||
if lines > max_lines:
|
||||
violations.append((rel_path, lines))
|
||||
|
||||
return total_files, violations
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""CLI entry point for the LOC guard script."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Verify that production source files do not exceed the LOC ceiling."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-lines",
|
||||
type=int,
|
||||
default=DEFAULT_MAX_LINES,
|
||||
help=f"Maximum allowed lines per file (default: {DEFAULT_MAX_LINES})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dirs",
|
||||
nargs="+",
|
||||
default=DEFAULT_TARGET_DIRS,
|
||||
help=f"Target directories to scan (default: {' '.join(DEFAULT_TARGET_DIRS)})",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--root",
|
||||
type=str,
|
||||
default=str(Path(__file__).resolve().parent.parent),
|
||||
help="Root repository directory",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-v", "--verbose",
|
||||
action="store_true",
|
||||
help="Enable verbose output listing all scanned files",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
root_dir = Path(args.root).resolve()
|
||||
|
||||
print("=" * 70)
|
||||
print(f"CASAN Guard 'S' (Single Responsibility): Checking file length <= {args.max_lines} LOC")
|
||||
print(f"Scanning target directories: {args.dirs}")
|
||||
print("=" * 70)
|
||||
|
||||
total_files, violations = scan_directories(
|
||||
root_dir=root_dir,
|
||||
target_dirs=args.dirs,
|
||||
max_lines=args.max_lines,
|
||||
verbose=args.verbose,
|
||||
)
|
||||
|
||||
if violations:
|
||||
print(f"\n[FAIL] Found {len(violations)} oversized file(s) (> {args.max_lines} LOC):")
|
||||
for file_path, lines in sorted(violations, key=lambda x: x[1], reverse=True):
|
||||
print(f" ❌ {file_path}: {lines} lines (exceeds limit by {lines - args.max_lines})")
|
||||
print("\nAction Required: Refactor oversized files into smaller single-responsibility modules.")
|
||||
return 1
|
||||
|
||||
print(f"\n[PASS] All {total_files} production files in {args.dirs} satisfy <= {args.max_lines} LOC limit.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,143 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unified CASAN Quality Gate Orchestrator (EPIC R10 - Quality Assurance).
|
||||
|
||||
Runs all verification gates to validate architecture, security, single responsibility,
|
||||
and test suite compliance before merging PRs or cutting a release.
|
||||
|
||||
Verification Stages (CASAN):
|
||||
1. [C] Clean Architecture Guard (scripts/check_imports.py)
|
||||
2. [A] Atomic & Secrets Audit (scripts/audit_security.py)
|
||||
3. [S] Single Responsibility / LOC Guard (scripts/check_loc.py)
|
||||
4. [A/N] Automated Tests & No-Regression Suite (pytest)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
# Ensure stdout handles UTF-8 on Windows consoles without codec crash
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
try:
|
||||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def run_stage(title: str, cmd: List[str], cwd: Path) -> Tuple[bool, float, str]:
|
||||
"""Execute a single quality gate command and measure elapsed duration.
|
||||
|
||||
Returns:
|
||||
A tuple of (success_boolean, elapsed_seconds, combined_output)
|
||||
"""
|
||||
print(f"\n>> Running Gate: {title} ...")
|
||||
start_time = time.time()
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
cwd=cwd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
)
|
||||
elapsed = time.time() - start_time
|
||||
success = proc.returncode == 0
|
||||
output = proc.stdout + ("\n" + proc.stderr if proc.stderr else "")
|
||||
return success, elapsed, output
|
||||
except Exception as exc:
|
||||
elapsed = time.time() - start_time
|
||||
return False, elapsed, f"Exception occurred while running {cmd}: {exc}"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Main CLI orchestrator for CASAN quality gates."""
|
||||
parser = argparse.ArgumentParser(description="Run CASAN Quality Gates on the repository.")
|
||||
parser.add_argument(
|
||||
"--skip-tests",
|
||||
action="store_true",
|
||||
help="Skip running pytest (run static and architectural guards only)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--verbose",
|
||||
"-v",
|
||||
action="store_true",
|
||||
help="Print detailed command output for passing gates as well",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
print("=" * 75)
|
||||
print("COWORK LOCAL - CASAN QUALITY GATE RUNNER")
|
||||
print("=" * 75)
|
||||
|
||||
stages = [
|
||||
(
|
||||
"C - Clean Architecture Boundary Check",
|
||||
[sys.executable, str(REPO_ROOT / "scripts" / "check_imports.py")],
|
||||
),
|
||||
(
|
||||
"A - Secrets & Plaintext Credentials Audit",
|
||||
[sys.executable, str(REPO_ROOT / "scripts" / "audit_security.py")],
|
||||
),
|
||||
(
|
||||
"S - Single Responsibility LOC Limit (<= 400 LOC)",
|
||||
[sys.executable, str(REPO_ROOT / "scripts" / "check_loc.py"), "--max-lines", "400"],
|
||||
),
|
||||
]
|
||||
|
||||
if not args.skip_tests:
|
||||
stages.append(
|
||||
(
|
||||
"A/N - Automated Pytest Suite (No-Regression)",
|
||||
[sys.executable, "-m", "pytest", "-q"],
|
||||
)
|
||||
)
|
||||
|
||||
results = []
|
||||
all_passed = True
|
||||
total_start = time.time()
|
||||
|
||||
for title, cmd in stages:
|
||||
success, elapsed, output = run_stage(title, cmd, cwd=REPO_ROOT)
|
||||
results.append((title, success, elapsed, output))
|
||||
|
||||
if success:
|
||||
print(f" [PASS] {title} ({elapsed:.2f}s)")
|
||||
if args.verbose:
|
||||
print(output.strip())
|
||||
else:
|
||||
all_passed = False
|
||||
print(f" [FAIL] {title} ({elapsed:.2f}s)")
|
||||
print("\n--- Output ---")
|
||||
print(output.strip())
|
||||
print("--------------")
|
||||
|
||||
total_elapsed = time.time() - total_start
|
||||
|
||||
print("\n" + "=" * 75)
|
||||
print("QUALITY GATE SUMMARY REPORT")
|
||||
print("=" * 75)
|
||||
for title, success, elapsed, _ in results:
|
||||
status_str = "[PASS]" if success else "[FAIL]"
|
||||
print(f" {status_str:<8} | {elapsed:>6.2f}s | {title}")
|
||||
|
||||
print("-" * 75)
|
||||
print(f"Total Execution Time: {total_elapsed:.2f}s")
|
||||
|
||||
if all_passed:
|
||||
print("\nALL CASAN QUALITY GATES PASSED! Ready for PR merge or release.")
|
||||
return 0
|
||||
else:
|
||||
print("\nQUALITY GATE FAILED! Please resolve the issues above before proceeding.")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user