feat(seedgo): fix: remove residual progress bar line after audit completion
Co-Authored-By: @seedgo <seedgo@aipass>
This commit is contained in:
@@ -47,6 +47,10 @@ Before changing a checker or standard: prove it catches the real case AND doesn'
|
||||
|
||||
When I fix my own compliance: eat my own dogfood. If seedgo can't pass its own audit, nothing else matters.
|
||||
|
||||
## Access
|
||||
|
||||
Seedgo and devpulse have **system-wide file access**. The "no cross-branch edits" rule does not apply — seedgo needs to edit system files (`.aipass/`, global prompts) and inspect any branch's code for standards enforcement.
|
||||
|
||||
## Quick Reference
|
||||
|
||||
- Pack discovery: `handlers/*_standards/` dirs with `*_check.py` files
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: commented_logger_check.py
|
||||
# Description: Commented Logger Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Commented Logger Standards Checker Handler
|
||||
|
||||
Detects commented-out logger calls in Python files. Lines like:
|
||||
|
||||
# logger.error(...)
|
||||
# logger.warning(...)
|
||||
# logger.info(...)
|
||||
# logger.debug(...)
|
||||
# logger.critical(...)
|
||||
# logger.exception(...)
|
||||
|
||||
These indicate intentionally disabled logging that should either be
|
||||
restored or removed entirely -- dead logging is noise.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
# Audit scope: scan every .py file, not just entry point
|
||||
AUDIT_SCOPE = "all_files"
|
||||
|
||||
# Regex extracted from devpulse commented_logger_scanner_v1.py
|
||||
_COMMENTED_LOGGER_RE = re.compile(
|
||||
r"#\s*logger\.(error|warning|warn|info|exception|critical|debug)\s*\("
|
||||
)
|
||||
|
||||
|
||||
def is_bypassed(file_path: str, standard: str, bypass_rules: list | None = None) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get('standard') and rule.get('standard') != standard:
|
||||
continue
|
||||
rule_file = rule.get('file', '')
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get('lines', [])
|
||||
if not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
"""
|
||||
Check a Python file for commented-out logger calls.
|
||||
|
||||
Scans every non-docstring, non-__init__.py line for patterns like
|
||||
``# logger.error(...)``. One check is produced per file: passed if
|
||||
zero violations, failed if any found (message includes count and
|
||||
first three line numbers).
|
||||
|
||||
Args:
|
||||
module_path: Path to Python module to check
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks
|
||||
|
||||
Returns:
|
||||
dict: {
|
||||
'passed': bool,
|
||||
'checks': [{'name': str, 'passed': bool, 'message': str}],
|
||||
'score': int,
|
||||
'standard': str
|
||||
}
|
||||
"""
|
||||
path = Path(module_path)
|
||||
|
||||
# --- bypass -----------------------------------------------------------
|
||||
if is_bypassed(module_path, 'commented_logger', bypass_rules=bypass_rules):
|
||||
return {
|
||||
'passed': True,
|
||||
'checks': [{'name': 'Bypassed', 'passed': True, 'message': 'Standard bypassed via .seedgo/bypass.json'}],
|
||||
'score': 100,
|
||||
'standard': 'COMMENTED_LOGGER'
|
||||
}
|
||||
|
||||
# --- skip non-.py and __init__.py -------------------------------------
|
||||
if path.suffix != '.py' or path.name == '__init__.py':
|
||||
return {
|
||||
'passed': True,
|
||||
'checks': [{'name': 'Commented logger calls', 'passed': True, 'message': 'File skipped (non-target)'}],
|
||||
'score': 100,
|
||||
'standard': 'COMMENTED_LOGGER'
|
||||
}
|
||||
|
||||
# --- file exists ------------------------------------------------------
|
||||
if not path.exists():
|
||||
return {
|
||||
'passed': False,
|
||||
'checks': [{'name': 'File exists', 'passed': False, 'message': f'File not found: {module_path}'}],
|
||||
'score': 0,
|
||||
'standard': 'COMMENTED_LOGGER'
|
||||
}
|
||||
|
||||
# --- read file --------------------------------------------------------
|
||||
try:
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
source = f.read()
|
||||
except Exception as e:
|
||||
return {
|
||||
'passed': False,
|
||||
'checks': [{'name': 'File readable', 'passed': False, 'message': f'Error reading file: {e}'}],
|
||||
'score': 0,
|
||||
'standard': 'COMMENTED_LOGGER'
|
||||
}
|
||||
|
||||
# --- scan for commented-out logger calls, skipping docstrings ---------
|
||||
violation_lines: list[int] = []
|
||||
in_docstring = False
|
||||
|
||||
for lineno, line in enumerate(source.splitlines(), start=1):
|
||||
stripped = line.strip()
|
||||
|
||||
# Track triple-quote docstring boundaries
|
||||
triple_count = stripped.count('"""') + stripped.count("'''")
|
||||
if triple_count == 1:
|
||||
in_docstring = not in_docstring
|
||||
continue
|
||||
if triple_count >= 2:
|
||||
# Opening and closing on same line -- not inside docstring
|
||||
continue
|
||||
if in_docstring:
|
||||
continue
|
||||
|
||||
if _COMMENTED_LOGGER_RE.search(line):
|
||||
violation_lines.append(lineno)
|
||||
|
||||
# --- build result -----------------------------------------------------
|
||||
checks = []
|
||||
violation_count = len(violation_lines)
|
||||
|
||||
if violation_count == 0:
|
||||
checks.append({
|
||||
'name': 'Commented logger calls',
|
||||
'passed': True,
|
||||
'message': 'No commented-out logger calls found'
|
||||
})
|
||||
else:
|
||||
first_three = violation_lines[:3]
|
||||
line_preview = ', '.join(str(ln) for ln in first_three)
|
||||
suffix = f' (and {violation_count - 3} more)' if violation_count > 3 else ''
|
||||
checks.append({
|
||||
'name': 'Commented logger calls',
|
||||
'passed': False,
|
||||
'message': f'{violation_count} commented-out logger call(s) on lines {line_preview}{suffix} -- restore or remove'
|
||||
})
|
||||
|
||||
# --- score ------------------------------------------------------------
|
||||
passed_checks = sum(1 for c in checks if c['passed'])
|
||||
total_checks = len(checks)
|
||||
score = int((passed_checks / total_checks) * 100) if total_checks > 0 else 0
|
||||
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation("check_completed", {"file": str(module_path), "score": score, "standard": "commented_logger"})
|
||||
return {
|
||||
'passed': overall_passed,
|
||||
'checks': checks,
|
||||
'score': score,
|
||||
'standard': 'COMMENTED_LOGGER'
|
||||
}
|
||||
@@ -0,0 +1,369 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: dead_code_check.py
|
||||
# Description: Dead Code Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Dead Code Standards Checker Handler
|
||||
|
||||
Detects unused Python modules and handlers within an AIPass branch.
|
||||
|
||||
Scans .py files in apps/modules/ and apps/handlers/, then checks whether
|
||||
anything in the branch's apps/ directory imports or references them.
|
||||
Files with zero references are flagged as dead code.
|
||||
|
||||
Score: referenced_files / total_files * 100, threshold 75%.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "branch_level"
|
||||
|
||||
# Directories to skip when collecting source files
|
||||
_SKIP_DIRS = {
|
||||
"__pycache__", ".archive", ".mypy_cache", ".ruff_cache",
|
||||
".pytest_cache", "json_templates", "logs", "tools",
|
||||
".venv", "venv", "node_modules", ".git", "site-packages",
|
||||
".trinity", ".aipass", ".ai_mail.local", ".spawn",
|
||||
"backups", "reports", "docs", "tests", ".sorting_unprocessed",
|
||||
}
|
||||
|
||||
|
||||
# =============================================
|
||||
# BYPASS HELPER
|
||||
# =============================================
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get("standard") and rule.get("standard") != standard:
|
||||
continue
|
||||
rule_file = rule.get("file", "")
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get("lines", [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# =============================================
|
||||
# FILE COLLECTION
|
||||
# =============================================
|
||||
|
||||
def _should_skip(path: Path) -> bool:
|
||||
"""Check whether any parent directory component is in the skip set."""
|
||||
return any(part in _SKIP_DIRS for part in path.parts)
|
||||
|
||||
|
||||
def _collect_scannable_files(apps_dir: Path) -> list[Path]:
|
||||
"""
|
||||
Collect .py files from apps/modules/ and apps/handlers/ that should be
|
||||
checked for usage. Skips __init__.py, __pycache__, .archive, etc.
|
||||
"""
|
||||
targets: list[Path] = []
|
||||
for subdir_name in ("modules", "handlers"):
|
||||
subdir = apps_dir / subdir_name
|
||||
if not subdir.is_dir():
|
||||
continue
|
||||
for py_file in subdir.rglob("*.py"):
|
||||
if py_file.name == "__init__.py":
|
||||
continue
|
||||
if _should_skip(py_file):
|
||||
continue
|
||||
targets.append(py_file)
|
||||
return sorted(targets)
|
||||
|
||||
|
||||
def _collect_source_text(apps_dir: Path) -> str:
|
||||
"""
|
||||
Read ALL .py files under apps/ into a single string for searching.
|
||||
This is the corpus we search for references.
|
||||
"""
|
||||
parts: list[str] = []
|
||||
for py_file in apps_dir.rglob("*.py"):
|
||||
if _should_skip(py_file):
|
||||
continue
|
||||
try:
|
||||
parts.append(py_file.read_text(encoding="utf-8", errors="ignore"))
|
||||
except OSError:
|
||||
continue
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
# =============================================
|
||||
# REFERENCE CHECKING
|
||||
# =============================================
|
||||
|
||||
def _build_import_path(py_file: Path, branch_path: Path, branch_name: str) -> str:
|
||||
"""
|
||||
Build the dotted import path for a file.
|
||||
|
||||
For src/aipass/prax/apps/handlers/monitoring/log_watcher.py:
|
||||
-> aipass.prax.apps.handlers.monitoring.log_watcher
|
||||
|
||||
For src/commons/apps/modules/post_manager.py:
|
||||
-> commons.apps.modules.post_manager
|
||||
"""
|
||||
try:
|
||||
rel = py_file.relative_to(branch_path)
|
||||
except ValueError:
|
||||
return py_file.stem
|
||||
|
||||
parts = list(rel.with_suffix("").parts)
|
||||
|
||||
# If the branch is under src/aipass/, prefix is aipass.{branch}
|
||||
# If under src/{name}/ (commons, skills), prefix is just {name}
|
||||
if "aipass" in branch_path.parts:
|
||||
return f"aipass.{branch_name}.{'.'.join(parts)}"
|
||||
return f"{branch_name}.{'.'.join(parts)}"
|
||||
|
||||
|
||||
def _check_file_used(
|
||||
py_file: Path,
|
||||
branch_path: Path,
|
||||
branch_name: str,
|
||||
source_text: str,
|
||||
entry_point_name: str,
|
||||
) -> bool:
|
||||
"""
|
||||
Determine if a .py file is referenced anywhere in the branch source.
|
||||
|
||||
A file counts as "used" if ANY of these hold:
|
||||
1. __init__.py (package structure) -- always used
|
||||
2. Entry point (apps/{branch}.py) -- always used
|
||||
3. Glob/discovery convention (*_check.py, *_content.py) -- always used
|
||||
4. Import by dotted path or relative path found in corpus
|
||||
5. Stem appears in an import statement in corpus
|
||||
6. Filename string reference in corpus
|
||||
"""
|
||||
stem = py_file.stem
|
||||
|
||||
# Rule 1: __init__.py is always used
|
||||
if py_file.name == "__init__.py":
|
||||
return True
|
||||
|
||||
# Rule 2: entry point
|
||||
if py_file.name == f"{entry_point_name}.py" and py_file.parent.name == "apps":
|
||||
return True
|
||||
|
||||
# Rule 3: glob/discovery convention files
|
||||
for suffix_pattern in ("_check", "_content"):
|
||||
glob_lit = f'glob("*{suffix_pattern}.py")'
|
||||
glob_lit_sq = f"glob('*{suffix_pattern}.py')"
|
||||
if stem.endswith(suffix_pattern) and (
|
||||
glob_lit in source_text or glob_lit_sq in source_text
|
||||
):
|
||||
return True
|
||||
|
||||
# Rule 4: full dotted import path
|
||||
import_path = _build_import_path(py_file, branch_path, branch_name)
|
||||
if import_path in source_text:
|
||||
return True
|
||||
|
||||
# Also check relative paths within the branch
|
||||
try:
|
||||
rel = py_file.relative_to(branch_path / "apps")
|
||||
rel_dotted = ".".join(rel.with_suffix("").parts)
|
||||
if rel_dotted in source_text:
|
||||
return True
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Rule 5: stem appears in import statements
|
||||
esc = re.escape(stem)
|
||||
import_patterns = [
|
||||
rf"from\s+\S*\.{esc}\s+import\b",
|
||||
rf"import\s+\S*\.{esc}\b",
|
||||
rf"from\s+\S+\s+import\s+[^#\n]*\b{esc}\b",
|
||||
]
|
||||
for pat in import_patterns:
|
||||
if re.search(pat, source_text):
|
||||
return True
|
||||
|
||||
# importlib.import_module with the stem
|
||||
if re.search(
|
||||
rf'import_module\([^)]*["\'].*\.{esc}["\']',
|
||||
source_text,
|
||||
):
|
||||
return True
|
||||
|
||||
# Glob-based auto-discovery for direct children of modules/
|
||||
try:
|
||||
rel_to_apps = py_file.relative_to(branch_path / "apps")
|
||||
parts = rel_to_apps.parts
|
||||
if len(parts) == 2 and parts[0] == "modules":
|
||||
if 'glob("*.py")' in source_text or "glob('*.py')" in source_text:
|
||||
return True
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Rule 6: filename string reference
|
||||
filename = py_file.name
|
||||
if re.search(rf'["\']{re.escape(filename)}["\']', source_text):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
# =============================================
|
||||
# BRANCH-LEVEL CHECK (audit pipeline entry)
|
||||
# =============================================
|
||||
|
||||
def check_branch(branch_path: str, bypass_rules: list | None = None) -> dict:
|
||||
"""
|
||||
Check a branch for dead code (unreferenced modules and handlers).
|
||||
|
||||
Args:
|
||||
branch_path: Path to branch root (e.g., src/aipass/seedgo)
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks
|
||||
|
||||
Returns:
|
||||
dict: {
|
||||
'passed': bool,
|
||||
'score': int,
|
||||
'checks': [{'name': str, 'passed': bool, 'message': str}],
|
||||
'standard': 'DEAD_CODE'
|
||||
}
|
||||
"""
|
||||
bp = Path(branch_path)
|
||||
|
||||
# Check if entire standard is bypassed
|
||||
if is_bypassed(branch_path, "dead_code", bypass_rules=bypass_rules):
|
||||
result = {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Bypassed",
|
||||
"passed": True,
|
||||
"message": "Standard bypassed via .seedgo/bypass.json",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "DEAD_CODE",
|
||||
}
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"branch": branch_path, "score": 100, "standard": "dead_code"},
|
||||
)
|
||||
return result
|
||||
|
||||
apps_dir = bp / "apps"
|
||||
if not apps_dir.is_dir():
|
||||
result = {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Dead code files",
|
||||
"passed": True,
|
||||
"message": f"No apps/ directory found in {branch_path}",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "DEAD_CODE",
|
||||
}
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"branch": branch_path, "score": 100, "standard": "dead_code"},
|
||||
)
|
||||
return result
|
||||
|
||||
# Determine branch name and entry point
|
||||
branch_name = bp.name
|
||||
entry_point_name = branch_name
|
||||
|
||||
# Collect scannable files
|
||||
targets = _collect_scannable_files(apps_dir)
|
||||
if not targets:
|
||||
result = {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Dead code files",
|
||||
"passed": True,
|
||||
"message": "No modules or handlers found to check",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "DEAD_CODE",
|
||||
}
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"branch": branch_path, "score": 100, "standard": "dead_code"},
|
||||
)
|
||||
return result
|
||||
|
||||
# Build the source corpus (all .py content from apps/)
|
||||
source_text = _collect_source_text(apps_dir)
|
||||
|
||||
# Check each target for references
|
||||
total_files = len(targets)
|
||||
dead_files: list[str] = []
|
||||
|
||||
for target in targets:
|
||||
used = _check_file_used(
|
||||
target, bp, branch_name, source_text, entry_point_name
|
||||
)
|
||||
if not used:
|
||||
# Check per-file bypass
|
||||
try:
|
||||
rel = target.relative_to(apps_dir)
|
||||
except ValueError:
|
||||
rel = target
|
||||
if not is_bypassed(str(rel), "dead_code", bypass_rules=bypass_rules):
|
||||
dead_files.append(str(rel))
|
||||
|
||||
referenced_files = total_files - len(dead_files)
|
||||
score = int(referenced_files / total_files * 100) if total_files > 0 else 100
|
||||
|
||||
# Build the single check result
|
||||
if dead_files:
|
||||
dead_list = ", ".join(dead_files[:10])
|
||||
suffix = f" (+{len(dead_files) - 10} more)" if len(dead_files) > 10 else ""
|
||||
message = (
|
||||
f"{len(dead_files)}/{total_files} files unreferenced: "
|
||||
f"{dead_list}{suffix}"
|
||||
)
|
||||
else:
|
||||
message = f"All {total_files} files referenced -- no dead code"
|
||||
|
||||
check_passed = len(dead_files) == 0
|
||||
checks = [
|
||||
{
|
||||
"name": "Dead code files",
|
||||
"passed": check_passed,
|
||||
"message": message,
|
||||
}
|
||||
]
|
||||
|
||||
overall_passed = score >= 75
|
||||
|
||||
result = {
|
||||
"passed": overall_passed,
|
||||
"checks": checks,
|
||||
"score": score,
|
||||
"standard": "DEAD_CODE",
|
||||
}
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"branch": branch_path, "score": score, "standard": "dead_code"},
|
||||
)
|
||||
return result
|
||||
@@ -0,0 +1,292 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: debug_print_check.py
|
||||
# Description: Debug Print Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Debug Print Standards Checker Handler
|
||||
|
||||
Detects bare print() calls that should use structured logging (Prax/Rich).
|
||||
Excludes lines inside docstrings, comments, doctests, and
|
||||
``if __name__ == "__main__":`` blocks.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "all_files"
|
||||
|
||||
# Matches a bare print( call: not preceded by a word char, dot, or #
|
||||
# This excludes console.print(, logger.print(, etc.
|
||||
_PRINT_RE = re.compile(r"(?<![.#\w])print\(")
|
||||
|
||||
# Doctest / interactive example lines to skip (>>> and ...)
|
||||
_DOCTEST_RE = re.compile(r"^\s*(\.\.\.|>>>)\s")
|
||||
|
||||
# Test file name patterns
|
||||
_TEST_FILE_RE = re.compile(r"^(test_.+|.+_test|conftest)\.py$")
|
||||
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
# Must match standard
|
||||
if rule.get("standard") and rule.get("standard") != standard:
|
||||
continue
|
||||
# Must match file (check if rule file path is in the full path)
|
||||
rule_file = rule.get("file", "")
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
# Check line-specific bypass
|
||||
rule_lines = rule.get("lines", [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_in_main_block(lines: list[str], lineno: int) -> bool:
|
||||
"""
|
||||
Return True if the line at *lineno* (1-based) is inside an
|
||||
``if __name__ == "__main__":`` block.
|
||||
|
||||
Uses indentation: the flagged line must have greater indentation than the
|
||||
``if __name__`` header found by scanning backwards.
|
||||
"""
|
||||
main_block_indent: int | None = None
|
||||
for i in range(lineno - 2, -1, -1): # scan backwards (0-based)
|
||||
stripped = lines[i].strip()
|
||||
if re.match(r'if\s+__name__\s*==\s*["\']__main__["\']', stripped):
|
||||
main_block_indent = len(lines[i]) - len(lines[i].lstrip())
|
||||
break
|
||||
|
||||
if main_block_indent is None:
|
||||
return False
|
||||
|
||||
target_line = lines[lineno - 1]
|
||||
if not target_line.strip():
|
||||
return False
|
||||
target_indent = len(target_line) - len(target_line.lstrip())
|
||||
return target_indent > main_block_indent
|
||||
|
||||
|
||||
def _scan_file(file_path: Path) -> tuple[list[int], str | None]:
|
||||
"""
|
||||
Scan a single .py file for bare print() calls.
|
||||
|
||||
Returns:
|
||||
(line_numbers, error_message) -- error_message is None on success.
|
||||
"""
|
||||
try:
|
||||
source = file_path.read_text(encoding="utf-8", errors="ignore")
|
||||
except OSError as exc:
|
||||
return [], f"cannot read: {exc}"
|
||||
|
||||
lines = source.splitlines()
|
||||
hit_lines: list[int] = []
|
||||
in_docstring = False
|
||||
docstring_char: str | None = None # '"""' or "'''"
|
||||
|
||||
for lineno, line in enumerate(lines, start=1):
|
||||
stripped = line.strip()
|
||||
|
||||
# -- Docstring state machine --
|
||||
for tq in ('"""', "'''"):
|
||||
count = line.count(tq)
|
||||
if count == 0:
|
||||
continue
|
||||
if not in_docstring:
|
||||
in_docstring = True
|
||||
docstring_char = tq
|
||||
# Opening and closing on the same line (one-liner)
|
||||
if count >= 2:
|
||||
in_docstring = False
|
||||
docstring_char = None
|
||||
elif docstring_char == tq:
|
||||
in_docstring = False
|
||||
docstring_char = None
|
||||
|
||||
if in_docstring:
|
||||
continue
|
||||
|
||||
# Skip comment lines
|
||||
if stripped.startswith("#"):
|
||||
continue
|
||||
|
||||
# Skip doctest / interactive example lines
|
||||
if _DOCTEST_RE.match(line):
|
||||
continue
|
||||
|
||||
# Strip inline comments before checking for print(
|
||||
code_part = line.split("#")[0]
|
||||
|
||||
if not _PRINT_RE.search(code_part):
|
||||
continue
|
||||
|
||||
# Skip if inside `if __name__ == "__main__":` block
|
||||
if _is_in_main_block(lines, lineno):
|
||||
continue
|
||||
|
||||
hit_lines.append(lineno)
|
||||
|
||||
return hit_lines, None
|
||||
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
"""
|
||||
Check a Python file for bare debug print() calls.
|
||||
|
||||
Args:
|
||||
module_path: Path to the Python file to check.
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks.
|
||||
|
||||
Returns:
|
||||
dict: {
|
||||
'passed': bool,
|
||||
'checks': [{'name': str, 'passed': bool, 'message': str}],
|
||||
'score': int,
|
||||
'standard': 'DEBUG_PRINT'
|
||||
}
|
||||
"""
|
||||
path = Path(module_path)
|
||||
|
||||
# -- Bypass entire standard for this file --
|
||||
if is_bypassed(module_path, "debug_print", bypass_rules=bypass_rules):
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Bypassed",
|
||||
"passed": True,
|
||||
"message": "Standard bypassed via .seedgo/bypass.json",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "DEBUG_PRINT",
|
||||
}
|
||||
|
||||
# -- Skip __init__.py --
|
||||
if path.name == "__init__.py":
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Debug print calls",
|
||||
"passed": True,
|
||||
"message": "__init__.py skipped",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "DEBUG_PRINT",
|
||||
}
|
||||
|
||||
# -- Skip test files --
|
||||
if _TEST_FILE_RE.match(path.name):
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Debug print calls",
|
||||
"passed": True,
|
||||
"message": "Test file skipped",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "DEBUG_PRINT",
|
||||
}
|
||||
|
||||
# -- Validate file exists --
|
||||
if not path.exists():
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "File exists",
|
||||
"passed": False,
|
||||
"message": f"File not found: {module_path}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "DEBUG_PRINT",
|
||||
}
|
||||
|
||||
# -- Scan --
|
||||
hit_lines, error = _scan_file(path)
|
||||
|
||||
if error is not None:
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "File readable",
|
||||
"passed": False,
|
||||
"message": f"Error reading file: {error}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "DEBUG_PRINT",
|
||||
}
|
||||
|
||||
# -- Filter out bypassed lines --
|
||||
non_bypassed = [
|
||||
ln
|
||||
for ln in hit_lines
|
||||
if not is_bypassed(module_path, "debug_print", ln, bypass_rules)
|
||||
]
|
||||
|
||||
# -- Build result --
|
||||
checks: list[Dict] = []
|
||||
|
||||
if not non_bypassed:
|
||||
checks.append(
|
||||
{
|
||||
"name": "Debug print calls",
|
||||
"passed": True,
|
||||
"message": "No bare print() calls found",
|
||||
}
|
||||
)
|
||||
else:
|
||||
sample = ", ".join(str(ln) for ln in non_bypassed[:3])
|
||||
suffix = f" (and {len(non_bypassed) - 3} more)" if len(non_bypassed) > 3 else ""
|
||||
checks.append(
|
||||
{
|
||||
"name": "Debug print calls",
|
||||
"passed": False,
|
||||
"message": (
|
||||
f"{len(non_bypassed)} bare print() call(s) "
|
||||
f"on lines {sample}{suffix}"
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
passed_checks = sum(1 for c in checks if c["passed"])
|
||||
total_checks = len(checks)
|
||||
score = int(passed_checks / total_checks * 100) if total_checks > 0 else 0
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": score, "standard": "debug_print"},
|
||||
)
|
||||
|
||||
return {
|
||||
"passed": overall_passed,
|
||||
"checks": checks,
|
||||
"score": score,
|
||||
"standard": "DEBUG_PRINT",
|
||||
}
|
||||
@@ -0,0 +1,194 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: deep_nesting_check.py
|
||||
# Description: Deep Nesting Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Deep Nesting Standards Checker Handler
|
||||
|
||||
Scans Python source files for functions whose control-flow nesting depth
|
||||
exceeds the allowed threshold. A nesting level is added for each:
|
||||
If / For / While / Try / With / ExceptHandler.
|
||||
|
||||
Threshold: depth > 3 is a violation. Functions that exceed this limit
|
||||
should be decomposed into smaller helpers.
|
||||
"""
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "all_files"
|
||||
|
||||
# -- Nesting node types -----------------------------------------------------
|
||||
|
||||
_NESTING_NODES = (ast.If, ast.For, ast.While, ast.Try, ast.With,
|
||||
ast.ExceptHandler)
|
||||
|
||||
DEPTH_LIMIT = 3
|
||||
|
||||
|
||||
# -- Bypass helper -----------------------------------------------------------
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get('standard') and rule.get('standard') != standard:
|
||||
continue
|
||||
rule_file = rule.get('file', '')
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get('lines', [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# -- AST depth analysis ------------------------------------------------------
|
||||
|
||||
def _max_nesting_depth(node: ast.AST, current: int = 0) -> int:
|
||||
"""
|
||||
Recursively walk a function body and return the maximum nesting depth.
|
||||
|
||||
Each If/For/While/Try/With/ExceptHandler adds one level.
|
||||
The initial call starts at depth 0 (inside the function body).
|
||||
"""
|
||||
max_depth = current
|
||||
for child in ast.iter_child_nodes(node):
|
||||
if isinstance(child, _NESTING_NODES):
|
||||
child_depth = _max_nesting_depth(child, current + 1)
|
||||
else:
|
||||
child_depth = _max_nesting_depth(child, current)
|
||||
if child_depth > max_depth:
|
||||
max_depth = child_depth
|
||||
return max_depth
|
||||
|
||||
|
||||
# -- File scanning -----------------------------------------------------------
|
||||
|
||||
def _scan_file(file_path: Path) -> list[dict]:
|
||||
"""
|
||||
Parse a single .py file and return a list of violation dicts.
|
||||
|
||||
Each violation: {'func': str, 'depth': int, 'line': int}
|
||||
"""
|
||||
violations: list[dict] = []
|
||||
|
||||
try:
|
||||
source = file_path.read_text(encoding='utf-8', errors='ignore')
|
||||
tree = ast.parse(source, filename=str(file_path))
|
||||
except SyntaxError:
|
||||
return violations
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
continue
|
||||
depth = _max_nesting_depth(node)
|
||||
if depth > DEPTH_LIMIT:
|
||||
violations.append({
|
||||
'func': node.name,
|
||||
'depth': depth,
|
||||
'line': node.lineno,
|
||||
})
|
||||
|
||||
return violations
|
||||
|
||||
|
||||
# -- Public checker entry point ----------------------------------------------
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> dict:
|
||||
"""
|
||||
Check if a module complies with deep nesting standards.
|
||||
|
||||
Args:
|
||||
module_path: Path to the Python file to check.
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks.
|
||||
|
||||
Returns:
|
||||
dict with keys: passed, score, checks, standard.
|
||||
"""
|
||||
checks: list[dict] = []
|
||||
path = Path(module_path)
|
||||
|
||||
# Check if entire standard is bypassed for this file
|
||||
if is_bypassed(module_path, 'deep_nesting', bypass_rules=bypass_rules):
|
||||
return {
|
||||
'passed': True,
|
||||
'checks': [{'name': 'Bypassed', 'passed': True, 'message': 'Standard bypassed via .seedgo/bypass.json'}],
|
||||
'score': 100,
|
||||
'standard': 'DEEP_NESTING',
|
||||
}
|
||||
|
||||
# Skip __init__.py files
|
||||
if path.name == '__init__.py':
|
||||
return {
|
||||
'passed': True,
|
||||
'checks': [{'name': 'Deep nesting', 'passed': True, 'message': '__init__.py skipped'}],
|
||||
'score': 100,
|
||||
'standard': 'DEEP_NESTING',
|
||||
}
|
||||
|
||||
# Validate file exists
|
||||
if not path.exists():
|
||||
return {
|
||||
'passed': False,
|
||||
'checks': [{'name': 'File exists', 'passed': False, 'message': f'File not found: {module_path}'}],
|
||||
'score': 0,
|
||||
'standard': 'DEEP_NESTING',
|
||||
}
|
||||
|
||||
# Scan for deep nesting violations
|
||||
violations = _scan_file(path)
|
||||
|
||||
if not violations:
|
||||
checks.append({
|
||||
'name': 'Deep nesting',
|
||||
'passed': True,
|
||||
'message': 'All functions within nesting limit (max depth 3)',
|
||||
})
|
||||
else:
|
||||
func_details = ', '.join(
|
||||
f'{v["func"]}() depth {v["depth"]} line {v["line"]}'
|
||||
for v in violations
|
||||
)
|
||||
checks.append({
|
||||
'name': 'Deep nesting',
|
||||
'passed': False,
|
||||
'message': (
|
||||
f'{len(violations)} function{"s" if len(violations) != 1 else ""} '
|
||||
f'exceed nesting limit: {func_details}'
|
||||
),
|
||||
})
|
||||
|
||||
# Calculate score
|
||||
passed_checks = sum(1 for c in checks if c['passed'])
|
||||
total_checks = len(checks)
|
||||
score = int((passed_checks / total_checks) * 100) if total_checks > 0 else 0
|
||||
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": score, "standard": "deep_nesting"},
|
||||
)
|
||||
|
||||
return {
|
||||
'passed': overall_passed,
|
||||
'score': score,
|
||||
'checks': checks,
|
||||
'standard': 'DEEP_NESTING',
|
||||
}
|
||||
@@ -10,7 +10,9 @@
|
||||
Error Handling Standards Checker Handler
|
||||
|
||||
Validates module compliance with AIPass 3-tier logging standards.
|
||||
Checks Prax imports in modules/handlers, logger calls in handlers.
|
||||
- Modules: MUST import Prax, logger.error() for system failures only
|
||||
- Handlers: MAY import Prax for info/warning, MUST NOT use logger.error()
|
||||
- stdlib logging.getLogger() prohibited everywhere
|
||||
"""
|
||||
|
||||
import re
|
||||
@@ -115,6 +117,9 @@ def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
no_stdlib_check = check_handler_no_stdlib_logging(content, module_path, bypass_rules)
|
||||
checks.append(no_stdlib_check)
|
||||
|
||||
# Note: Handlers MAY import Prax and use logger.info/warning/error (DPLAN-0040)
|
||||
# Only stdlib logging.getLogger() is prohibited (check 2 above)
|
||||
|
||||
# Check 4: Modules should have error logging
|
||||
if is_module:
|
||||
error_logging_check = check_module_error_logging(content)
|
||||
@@ -323,3 +328,5 @@ def check_error_vs_warning_usage(lines: List[str], file_path: str, bypass_rules:
|
||||
'passed': True,
|
||||
'message': 'logger.error() correctly used for system failures only'
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -40,12 +40,13 @@ def get_error_handling_standards() -> str:
|
||||
" [green]✓ Return bool to entry point[/green]",
|
||||
"",
|
||||
"[bold cyan]TIER 3: HANDLERS[/bold cyan] (apps/handlers/**/*.py)",
|
||||
" [dim]Prax Import:[/dim] [red]NO (PROHIBITED)[/red]",
|
||||
" [dim]Logging Scope:[/dim] [red]NONE[/red]",
|
||||
" [red]✗ NO Prax imports[/red]",
|
||||
" [red]✗ NO logger calls[/red]",
|
||||
" [dim]Prax Import:[/dim] [green]ALLOWED[/green]",
|
||||
" [dim]Logging Scope:[/dim] [green]info, warning, error[/green]",
|
||||
" [green]✓ logger.info() — operational visibility[/green]",
|
||||
" [green]✓ logger.warning() — non-critical issues[/green]",
|
||||
" [green]✓ logger.error() — error context at point of failure[/green]",
|
||||
" [green]✓ Return status dicts or raise exceptions[/green]",
|
||||
" [green]✓ Pure workers - testable in isolation[/green]",
|
||||
" [green]✓ Testable in isolation[/green]",
|
||||
"",
|
||||
"─" * 70,
|
||||
"",
|
||||
@@ -112,11 +113,11 @@ def get_error_handling_standards() -> str:
|
||||
"[dim]# All modules MUST import Prax[/dim]",
|
||||
"[dim]grep -r \"from aipass.prax\" apps/modules/*.py[/dim]",
|
||||
"",
|
||||
"[dim]# NO handlers can import Prax[/dim]",
|
||||
"[dim]grep -r \"from aipass.prax\" apps/handlers/**/*.py # Should find NOTHING[/dim]",
|
||||
"[dim]# Handlers MAY import Prax and use all log levels[/dim]",
|
||||
"[dim]grep -r \"from aipass.prax\" apps/handlers/**/*.py # Allowed[/dim]",
|
||||
"",
|
||||
"[dim]# NO handlers can call logger[/dim]",
|
||||
"[dim]grep -r \"logger\\.\" apps/handlers/**/*.py # Should find NOTHING[/dim]",
|
||||
"[dim]# Handlers MUST NOT use stdlib logging.getLogger()[/dim]",
|
||||
"[dim]grep -r \"logging.getLogger\" apps/handlers/**/*.py # Should find NOTHING[/dim]",
|
||||
"",
|
||||
"─" * 70,
|
||||
"",
|
||||
|
||||
@@ -0,0 +1,312 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: hardcoded_key_check.py
|
||||
# Description: Hardcoded Key Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Hardcoded Key Standards Checker Handler
|
||||
|
||||
Scans Python source files for hardcoded API keys matching known provider
|
||||
prefixes (OpenRouter, OpenAI, Anthropic, Google, AWS, GitHub, Slack, etc.).
|
||||
Placeholder values, comments, and docstrings are filtered out so only
|
||||
genuine secrets trigger a failure.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "all_files"
|
||||
|
||||
# -- Key patterns -----------------------------------------------------------
|
||||
# Each tuple: (provider_label, compiled regex)
|
||||
# Patterns require the key to appear inside quotes and be long enough to
|
||||
# exclude obvious placeholders (minimum 8 chars after prefix).
|
||||
|
||||
_KEY_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
|
||||
(
|
||||
"OpenRouter (sk-or-v1-)",
|
||||
re.compile(r"""["'`](sk-or-v1-[A-Za-z0-9\-_]{8,})["'`]"""),
|
||||
),
|
||||
(
|
||||
"OpenAI project (sk-proj-)",
|
||||
re.compile(r"""["'`](sk-proj-[A-Za-z0-9\-_]{8,})["'`]"""),
|
||||
),
|
||||
(
|
||||
"Anthropic (sk-ant-)",
|
||||
re.compile(r"""["'`](sk-ant-[A-Za-z0-9\-_]{8,})["'`]"""),
|
||||
),
|
||||
(
|
||||
"Google (AIza)",
|
||||
re.compile(r"""["'`](AIza[A-Za-z0-9\-_]{20,})["'`]"""),
|
||||
),
|
||||
(
|
||||
"AWS (AKIA)",
|
||||
re.compile(r"""["'`](AKIA[A-Za-z0-9]{12,})["'`]"""),
|
||||
),
|
||||
(
|
||||
"GitHub (ghp_/gho_/ghs_)",
|
||||
re.compile(r"""["'`](gh[pos]_[A-Za-z0-9]{8,})["'`]"""),
|
||||
),
|
||||
(
|
||||
"Slack (xoxb-/xoxp-)",
|
||||
re.compile(r"""["'`](xox[bp]-[A-Za-z0-9\-]{8,})["'`]"""),
|
||||
),
|
||||
(
|
||||
"Generic (key-)",
|
||||
re.compile(r"""["'`](key-[A-Za-z0-9\-_]{16,})["'`]"""),
|
||||
),
|
||||
]
|
||||
|
||||
# Placeholder indicators -- if the captured value matches any of these the
|
||||
# hit is treated as documentation, not a real secret.
|
||||
_PLACEHOLDER_VALUE_RE = re.compile(
|
||||
r"(?:your[_\-]?key|xxx+|example|placeholder|abc123|new\-key|"
|
||||
r"v1-\.\.\.|all[_\-]?x|test|fake|dummy|sample|\.\.\.|"
|
||||
r"your_key_here|key[-_]here|insert|changeme|<|>)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
_PLACEHOLDER_SUFFIX_RE = re.compile(
|
||||
r"[-_](?:here|example|test|xxx+|placeholder|abc|demo|key|secret|value)$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Words on the line (outside the key literal) that signal example context.
|
||||
_EXAMPLE_CONTEXT_WORDS = frozenset({
|
||||
"example", "template", "placeholder", "sample", "demo",
|
||||
})
|
||||
|
||||
# Pure comment line.
|
||||
_PAT_COMMENT = re.compile(r"^\s*#")
|
||||
|
||||
# Regex compile context -- the key-like string is a detection pattern itself.
|
||||
_PAT_REGEX_CONTEXT = re.compile(r"""re\.compile|r["']|\\[dws\^]""")
|
||||
|
||||
|
||||
# -- Helpers ----------------------------------------------------------------
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get("standard") and rule.get("standard") != standard:
|
||||
continue
|
||||
rule_file = rule.get("file", "")
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get("lines", [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_placeholder(key_value: str) -> bool:
|
||||
"""Return True if the captured key value looks like a placeholder."""
|
||||
if _PLACEHOLDER_VALUE_RE.search(key_value):
|
||||
return True
|
||||
if _PLACEHOLDER_SUFFIX_RE.search(key_value):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# -- Core detection ---------------------------------------------------------
|
||||
|
||||
def _scan_line(lineno: int, line: str) -> list[tuple[int, str]]:
|
||||
"""
|
||||
Check a single source line for hardcoded key literals.
|
||||
|
||||
Returns a list of (lineno, provider_label) tuples for each finding.
|
||||
Skips comment lines, example-context lines, regex-compile contexts,
|
||||
and placeholder values.
|
||||
"""
|
||||
if _PAT_COMMENT.match(line):
|
||||
return []
|
||||
|
||||
line_lower = line.lower()
|
||||
if any(word in line_lower for word in _EXAMPLE_CONTEXT_WORDS):
|
||||
return []
|
||||
|
||||
if _PAT_REGEX_CONTEXT.search(line):
|
||||
return []
|
||||
|
||||
findings: list[tuple[int, str]] = []
|
||||
for label, pattern in _KEY_PATTERNS:
|
||||
for match in pattern.finditer(line):
|
||||
key_val = match.group(1)
|
||||
if _is_placeholder(key_val):
|
||||
continue
|
||||
findings.append((lineno, label))
|
||||
return findings
|
||||
|
||||
|
||||
def _scan_file(file_path: Path) -> list[tuple[int, str]]:
|
||||
"""
|
||||
Scan a single Python file for hardcoded keys.
|
||||
|
||||
Skips __init__.py files, docstring regions, and comment lines.
|
||||
Returns a list of (lineno, provider_label) tuples.
|
||||
"""
|
||||
if file_path.name == "__init__.py":
|
||||
return []
|
||||
|
||||
try:
|
||||
content = file_path.read_text(encoding="utf-8", errors="ignore")
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
lines = content.splitlines()
|
||||
findings: list[tuple[int, str]] = []
|
||||
in_docstring = False
|
||||
|
||||
for lineno, line in enumerate(lines, start=1):
|
||||
stripped = line.strip()
|
||||
|
||||
# Track triple-quoted docstrings
|
||||
triple_count = stripped.count('"""') + stripped.count("'''")
|
||||
if triple_count:
|
||||
if triple_count % 2 == 1:
|
||||
in_docstring = not in_docstring
|
||||
# Skip the line itself whether it opens or closes a docstring
|
||||
continue
|
||||
|
||||
if in_docstring:
|
||||
continue
|
||||
|
||||
findings.extend(_scan_line(lineno, line))
|
||||
|
||||
return findings
|
||||
|
||||
|
||||
# -- Public entry point -----------------------------------------------------
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> dict:
|
||||
"""
|
||||
Check if a Python file contains hardcoded API keys.
|
||||
|
||||
Args:
|
||||
module_path: Path to the Python file to check.
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks.
|
||||
|
||||
Returns:
|
||||
dict with keys: passed, score, checks, standard.
|
||||
"""
|
||||
# Whole-file bypass
|
||||
if is_bypassed(module_path, "hardcoded_key", bypass_rules=bypass_rules):
|
||||
result = {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Bypassed",
|
||||
"passed": True,
|
||||
"message": "Standard bypassed via .seedgo/bypass.json",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "HARDCODED_KEY",
|
||||
}
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": 100, "standard": "hardcoded_key"},
|
||||
)
|
||||
return result
|
||||
|
||||
path = Path(module_path)
|
||||
|
||||
# File existence
|
||||
if not path.exists():
|
||||
result = {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "File exists",
|
||||
"passed": False,
|
||||
"message": f"File not found: {module_path}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "HARDCODED_KEY",
|
||||
}
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": 0, "standard": "hardcoded_key"},
|
||||
)
|
||||
return result
|
||||
|
||||
# Skip __init__.py
|
||||
if path.name == "__init__.py":
|
||||
result = {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Hardcoded API keys",
|
||||
"passed": True,
|
||||
"message": "__init__.py skipped",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "HARDCODED_KEY",
|
||||
}
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": 100, "standard": "hardcoded_key"},
|
||||
)
|
||||
return result
|
||||
|
||||
# Scan the file
|
||||
raw_findings = _scan_file(path)
|
||||
|
||||
# Filter out bypassed lines
|
||||
findings: list[tuple[int, str]] = []
|
||||
for lineno, label in raw_findings:
|
||||
if not is_bypassed(module_path, "hardcoded_key", lineno, bypass_rules):
|
||||
findings.append((lineno, label))
|
||||
|
||||
# Build the single check entry
|
||||
checks: list[dict] = []
|
||||
if findings:
|
||||
line_numbers = [f[0] for f in findings]
|
||||
preview = ", ".join(str(ln) for ln in line_numbers[:3])
|
||||
suffix = f" (and {len(line_numbers) - 3} more)" if len(line_numbers) > 3 else ""
|
||||
checks.append({
|
||||
"name": "Hardcoded API keys",
|
||||
"passed": False,
|
||||
"message": f"Found {len(findings)} hardcoded key(s) on lines {preview}{suffix}",
|
||||
})
|
||||
else:
|
||||
checks.append({
|
||||
"name": "Hardcoded API keys",
|
||||
"passed": True,
|
||||
"message": "No hardcoded API keys detected",
|
||||
})
|
||||
|
||||
# Score
|
||||
passed_checks = sum(1 for c in checks if c["passed"])
|
||||
total_checks = len(checks)
|
||||
score = int(passed_checks / total_checks * 100) if total_checks > 0 else 0
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": score, "standard": "hardcoded_key"},
|
||||
)
|
||||
|
||||
return {
|
||||
"passed": overall_passed,
|
||||
"checks": checks,
|
||||
"score": score,
|
||||
"standard": "HARDCODED_KEY",
|
||||
}
|
||||
@@ -0,0 +1,259 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: help_text_check.py
|
||||
# Description: Help Text Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Help Text Standards Checker Handler
|
||||
|
||||
Detects user-facing string literals that reference ``python3`` or ``python``
|
||||
as a command instruction (e.g. "python3 tools/scanner.py"). These should
|
||||
tell the user to run commands via ``drone @branch`` instead.
|
||||
|
||||
Shebangs, import statements, and Python API references (like "python version")
|
||||
are excluded. Only instructional command invocations are flagged.
|
||||
|
||||
One check per file -- passed if zero references found, failed with a count
|
||||
and the first three offending line numbers.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "all_files"
|
||||
|
||||
# ── Detection patterns (extracted from devpulse help_text_scanner_v1) ───
|
||||
|
||||
# Lines to always skip
|
||||
_SHEBANG_RE = re.compile(r"^\s*#!")
|
||||
_COMMENT_RE = re.compile(r"^\s*#")
|
||||
|
||||
# Single-line string literals (single or double quoted)
|
||||
_STRING_RE = re.compile(
|
||||
r'"(?:[^"\\]|\\.)*"|' # double-quoted
|
||||
r"'(?:[^'\\]|\\.)*'", # single-quoted
|
||||
)
|
||||
|
||||
# Instructional python command references inside string content
|
||||
_PYTHON3_CMD_RE = re.compile(r"\bpython3\s+\S")
|
||||
_PYTHON_CMD_RE = re.compile(r"\bpython\s+(?:-[a-zA-Z]|\S+\.py)")
|
||||
|
||||
# Triple-quote delimiters
|
||||
_TRIPLE_DOUBLE = '"""'
|
||||
_TRIPLE_SINGLE = "'''"
|
||||
|
||||
|
||||
def _has_python_instruction(text: str) -> bool:
|
||||
"""Return True if *text* contains a python3/python command reference."""
|
||||
return bool(_PYTHON3_CMD_RE.search(text) or _PYTHON_CMD_RE.search(text))
|
||||
|
||||
|
||||
def _strings_on_line(line: str) -> list[str]:
|
||||
"""Return all single-line string literal tokens found on a source line."""
|
||||
return [m.group(0) for m in _STRING_RE.finditer(line)]
|
||||
|
||||
|
||||
def _line_has_python_instruction(line: str) -> bool:
|
||||
"""Check single-line string literals on *line* for python command refs."""
|
||||
return any(_has_python_instruction(s) for s in _strings_on_line(line))
|
||||
|
||||
|
||||
# ── Bypass helper ───────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get("standard") and rule.get("standard") != standard:
|
||||
continue
|
||||
rule_file = rule.get("file", "")
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get("lines", [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# ── Main checker entry point ────────────────────────────────────────────
|
||||
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
"""
|
||||
Check a Python file for user-facing help text that references python3/python
|
||||
commands instead of ``drone @branch``.
|
||||
|
||||
Args:
|
||||
module_path: Path to the Python file to check.
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks.
|
||||
|
||||
Returns:
|
||||
dict: {
|
||||
'passed': bool,
|
||||
'checks': [{'name': str, 'passed': bool, 'message': str}],
|
||||
'score': int,
|
||||
'standard': str,
|
||||
}
|
||||
"""
|
||||
path = Path(module_path)
|
||||
|
||||
# Skip __init__.py files
|
||||
if path.name == "__init__.py":
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Help text references",
|
||||
"passed": True,
|
||||
"message": "__init__.py skipped",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "HELP_TEXT",
|
||||
}
|
||||
|
||||
# Check if entire standard is bypassed for this file
|
||||
if is_bypassed(module_path, "help_text", bypass_rules=bypass_rules):
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Bypassed",
|
||||
"passed": True,
|
||||
"message": "Standard bypassed via .seedgo/bypass.json",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "HELP_TEXT",
|
||||
}
|
||||
|
||||
# Validate file exists
|
||||
if not path.exists():
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "File exists",
|
||||
"passed": False,
|
||||
"message": f"File not found: {module_path}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "HELP_TEXT",
|
||||
}
|
||||
|
||||
# Read file
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
except Exception as e:
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "File readable",
|
||||
"passed": False,
|
||||
"message": f"Error reading file: {e}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "HELP_TEXT",
|
||||
}
|
||||
|
||||
lines = content.splitlines()
|
||||
violation_lines: list[int] = []
|
||||
|
||||
# State machine for multiline strings (mirrors scanner logic)
|
||||
in_multiline: bool = False
|
||||
multiline_delim: str = ""
|
||||
|
||||
for lineno, raw_line in enumerate(lines, start=1):
|
||||
# --- Inside a multiline string block ---
|
||||
if in_multiline:
|
||||
if multiline_delim in raw_line:
|
||||
in_multiline = False
|
||||
multiline_delim = ""
|
||||
# Check content inside multiline strings (skip comment-only lines)
|
||||
if not _COMMENT_RE.match(raw_line) and _has_python_instruction(raw_line):
|
||||
if not is_bypassed(module_path, "help_text", lineno, bypass_rules):
|
||||
violation_lines.append(lineno)
|
||||
continue
|
||||
|
||||
# --- Normal line handling ---
|
||||
|
||||
# Skip shebangs and pure-comment lines
|
||||
if _SHEBANG_RE.match(raw_line) or _COMMENT_RE.match(raw_line):
|
||||
continue
|
||||
|
||||
# Check for opening triple-quote that does NOT close on the same line
|
||||
for delim in (_TRIPLE_DOUBLE, _TRIPLE_SINGLE):
|
||||
if delim in raw_line:
|
||||
count = raw_line.count(delim)
|
||||
if count % 2 == 1:
|
||||
in_multiline = True
|
||||
multiline_delim = delim
|
||||
break
|
||||
|
||||
# Check single-line string literals for python command references
|
||||
if _line_has_python_instruction(raw_line):
|
||||
if not is_bypassed(module_path, "help_text", lineno, bypass_rules):
|
||||
violation_lines.append(lineno)
|
||||
|
||||
# Build the single check result
|
||||
total_found = len(violation_lines)
|
||||
|
||||
if total_found == 0:
|
||||
check_result = {
|
||||
"name": "Help text references",
|
||||
"passed": True,
|
||||
"message": "No python3/python command references in help text",
|
||||
}
|
||||
else:
|
||||
first_three = ", ".join(str(ln) for ln in violation_lines[:3])
|
||||
suffix = f" (and {total_found - 3} more)" if total_found > 3 else ""
|
||||
check_result = {
|
||||
"name": "Help text references",
|
||||
"passed": False,
|
||||
"message": (
|
||||
f"Found {total_found} python3/python command "
|
||||
f"reference{'s' if total_found != 1 else ''} in help text "
|
||||
f"on line{'s' if total_found != 1 else ''} {first_three}{suffix} "
|
||||
f"-- should use drone @branch instead"
|
||||
),
|
||||
}
|
||||
|
||||
checks = [check_result]
|
||||
|
||||
# Score: one check, so 100 if passed, 0 if failed
|
||||
passed_checks = sum(1 for c in checks if c["passed"])
|
||||
total_checks = len(checks)
|
||||
score = int((passed_checks / total_checks) * 100) if total_checks > 0 else 0
|
||||
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": score, "standard": "help_text"},
|
||||
)
|
||||
return {
|
||||
"passed": overall_passed,
|
||||
"checks": checks,
|
||||
"score": score,
|
||||
"standard": "HELP_TEXT",
|
||||
}
|
||||
@@ -0,0 +1,228 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: silent_catch_check.py
|
||||
# Description: Silent Catch Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Silent Catch Standards Checker Handler
|
||||
|
||||
Detects except blocks that silently swallow exceptions -- no logger call
|
||||
and no re-raise. A silent catch is an ExceptHandler whose body:
|
||||
|
||||
1. Contains no logger.<level>() call (error, warning, info, debug,
|
||||
exception, critical)
|
||||
2. Contains no ``raise`` statement
|
||||
|
||||
These blocks hide failures and make debugging impossible. Detection
|
||||
logic extracted from devpulse silent_catch_scanner_v2.
|
||||
"""
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
# Audit scope: scan every .py file, not just entry point
|
||||
AUDIT_SCOPE = "all_files"
|
||||
|
||||
# Logger attribute names that count as "logging present"
|
||||
_LOGGING_ATTRS = frozenset({
|
||||
"error", "warning", "warn", "info", "debug", "exception", "critical"
|
||||
})
|
||||
|
||||
|
||||
def is_bypassed(file_path: str, standard: str, bypass_rules: list | None = None) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get('standard') and rule.get('standard') != standard:
|
||||
continue
|
||||
rule_file = rule.get('file', '')
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get('lines', [])
|
||||
if not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# -- AST helpers (extracted from devpulse silent_catch_scanner_v2) ---------
|
||||
|
||||
def _has_logger_call(nodes: list[ast.stmt]) -> bool:
|
||||
"""
|
||||
Return True if any node in *nodes* (or its descendants) contains a
|
||||
``logger.<level>()`` call.
|
||||
"""
|
||||
for node in ast.walk(ast.Module(body=nodes, type_ignores=[])):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
func = node.func
|
||||
if (
|
||||
isinstance(func, ast.Attribute)
|
||||
and func.attr in _LOGGING_ATTRS
|
||||
and isinstance(func.value, ast.Name)
|
||||
and func.value.id == "logger"
|
||||
):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _has_raise(nodes: list[ast.stmt]) -> bool:
|
||||
"""Return True if any node in *nodes* (or its descendants) is a Raise."""
|
||||
for node in ast.walk(ast.Module(body=nodes, type_ignores=[])):
|
||||
if isinstance(node, ast.Raise):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _is_noop_body(nodes: list[ast.stmt]) -> bool:
|
||||
"""
|
||||
Return True if the except body is effectively a no-op: just ``pass``,
|
||||
just ``...`` (Ellipsis), or just a bare string constant (docstring).
|
||||
"""
|
||||
if len(nodes) != 1:
|
||||
return False
|
||||
node = nodes[0]
|
||||
if isinstance(node, ast.Pass):
|
||||
return True
|
||||
# Ellipsis literal: ...
|
||||
if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant):
|
||||
if node.value.value is ...:
|
||||
return True
|
||||
# Bare string constant (docstring-style)
|
||||
if isinstance(node.value.value, str):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
"""
|
||||
Check a Python file for silent exception catches.
|
||||
|
||||
Parses the file with ``ast.parse()`` and walks every ExceptHandler
|
||||
node. A handler is flagged when its body contains neither a logger
|
||||
call nor a raise statement.
|
||||
|
||||
Args:
|
||||
module_path: Path to Python module to check
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks
|
||||
|
||||
Returns:
|
||||
dict: {
|
||||
'passed': bool,
|
||||
'checks': [{'name': str, 'passed': bool, 'message': str}],
|
||||
'score': int,
|
||||
'standard': str
|
||||
}
|
||||
"""
|
||||
path = Path(module_path)
|
||||
|
||||
# --- bypass -----------------------------------------------------------
|
||||
if is_bypassed(module_path, 'silent_catch', bypass_rules=bypass_rules):
|
||||
return {
|
||||
'passed': True,
|
||||
'checks': [{'name': 'Bypassed', 'passed': True, 'message': 'Standard bypassed via .seedgo/bypass.json'}],
|
||||
'score': 100,
|
||||
'standard': 'SILENT_CATCH'
|
||||
}
|
||||
|
||||
# --- skip non-.py and __init__.py -------------------------------------
|
||||
if path.suffix != '.py' or path.name == '__init__.py':
|
||||
return {
|
||||
'passed': True,
|
||||
'checks': [{'name': 'Silent catch blocks', 'passed': True, 'message': 'File skipped (non-target)'}],
|
||||
'score': 100,
|
||||
'standard': 'SILENT_CATCH'
|
||||
}
|
||||
|
||||
# --- file exists ------------------------------------------------------
|
||||
if not path.exists():
|
||||
return {
|
||||
'passed': False,
|
||||
'checks': [{'name': 'File exists', 'passed': False, 'message': f'File not found: {module_path}'}],
|
||||
'score': 0,
|
||||
'standard': 'SILENT_CATCH'
|
||||
}
|
||||
|
||||
# --- read file --------------------------------------------------------
|
||||
try:
|
||||
with open(path, 'r', encoding='utf-8') as f:
|
||||
source = f.read()
|
||||
except Exception as e:
|
||||
return {
|
||||
'passed': False,
|
||||
'checks': [{'name': 'File readable', 'passed': False, 'message': f'Error reading file: {e}'}],
|
||||
'score': 0,
|
||||
'standard': 'SILENT_CATCH'
|
||||
}
|
||||
|
||||
# --- parse AST --------------------------------------------------------
|
||||
try:
|
||||
tree = ast.parse(source, filename=str(path))
|
||||
except SyntaxError as e:
|
||||
return {
|
||||
'passed': False,
|
||||
'checks': [{'name': 'File parseable', 'passed': False, 'message': f'Syntax error: {e}'}],
|
||||
'score': 0,
|
||||
'standard': 'SILENT_CATCH'
|
||||
}
|
||||
|
||||
# --- walk AST for silent ExceptHandler nodes --------------------------
|
||||
silent_lines: list[int] = []
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.ExceptHandler):
|
||||
continue
|
||||
|
||||
body = node.body
|
||||
if not body:
|
||||
continue
|
||||
|
||||
# An except block is "silent" when it has neither a logger call
|
||||
# nor a raise -- it swallows the exception without reporting it
|
||||
if _has_logger_call(body) or _has_raise(body):
|
||||
continue
|
||||
|
||||
silent_lines.append(node.lineno)
|
||||
|
||||
silent_lines.sort()
|
||||
|
||||
# --- build result -----------------------------------------------------
|
||||
checks = []
|
||||
violation_count = len(silent_lines)
|
||||
|
||||
if violation_count == 0:
|
||||
checks.append({
|
||||
'name': 'Silent catch blocks',
|
||||
'passed': True,
|
||||
'message': 'No silent exception catches found'
|
||||
})
|
||||
else:
|
||||
first_three = silent_lines[:3]
|
||||
line_preview = ', '.join(str(ln) for ln in first_three)
|
||||
suffix = f' (and {violation_count - 3} more)' if violation_count > 3 else ''
|
||||
checks.append({
|
||||
'name': 'Silent catch blocks',
|
||||
'passed': False,
|
||||
'message': f'{violation_count} silent catch(es) on lines {line_preview}{suffix} -- add logger call or re-raise'
|
||||
})
|
||||
|
||||
# --- score ------------------------------------------------------------
|
||||
passed_checks = sum(1 for c in checks if c['passed'])
|
||||
total_checks = len(checks)
|
||||
score = int((passed_checks / total_checks) * 100) if total_checks > 0 else 0
|
||||
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation("check_completed", {"file": str(module_path), "score": score, "standard": "silent_catch"})
|
||||
return {
|
||||
'passed': overall_passed,
|
||||
'checks': checks,
|
||||
'score': score,
|
||||
'standard': 'SILENT_CATCH'
|
||||
}
|
||||
@@ -0,0 +1,376 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: test_coverage_check.py
|
||||
# Description: Test Coverage Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Test Coverage Standards Checker Handler
|
||||
|
||||
Branch-level checker that evaluates test coverage for a branch by:
|
||||
- Discovering test files (tests/ directory and scattered test_*.py / *_test.py)
|
||||
- Counting pytest-style test functions (def test_*)
|
||||
- Mapping tested modules via import patterns
|
||||
- Calculating module coverage (covered / total testable modules)
|
||||
|
||||
Extracted from devpulse test_scanner_v1 and wrapped as a seedgo checker.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "branch_level"
|
||||
|
||||
# -- Directories to skip when scanning ----------------------------------------
|
||||
SKIP_DIRS: set[str] = {
|
||||
"__pycache__", ".archive", ".mypy_cache", ".ruff_cache",
|
||||
".pytest_cache", ".venv", "venv", "node_modules", ".git",
|
||||
"site-packages", "logs", "tools", ".trinity", ".aipass",
|
||||
".ai_mail.local", ".spawn", "backups", "reports", "docs",
|
||||
".sorting_unprocessed",
|
||||
}
|
||||
|
||||
# -- Test function pattern ----------------------------------------------------
|
||||
RE_TEST_FUNC = re.compile(r"^\s*(?:async\s+)?def\s+(test_\w+)", re.MULTILINE)
|
||||
|
||||
# -- Import patterns for mapping tests to modules ----------------------------
|
||||
RE_IMPORT_FROM = re.compile(
|
||||
r"from\s+(?:aipass\.)?\w+\.apps\.(?:modules|handlers)[./]?([\w.]*)\s+import"
|
||||
)
|
||||
RE_IMPORT_DIRECT = re.compile(
|
||||
r"import\s+(?:aipass\.)?\w+\.apps\.(?:modules|handlers)[./]?([\w.]*)"
|
||||
)
|
||||
|
||||
|
||||
# =============================================
|
||||
# BYPASS HELPER
|
||||
# =============================================
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get("standard") and rule.get("standard") != standard:
|
||||
continue
|
||||
rule_file = rule.get("file", "")
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get("lines", [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# =============================================
|
||||
# FILE HELPERS
|
||||
# =============================================
|
||||
|
||||
def _read_file_safe(path: Path) -> str:
|
||||
"""Read a file, returning empty string on any error."""
|
||||
try:
|
||||
return path.read_text(encoding="utf-8")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
return ""
|
||||
|
||||
|
||||
def _should_skip_dir(name: str) -> bool:
|
||||
"""Check if a directory name should be skipped."""
|
||||
return name in SKIP_DIRS or name.startswith(".")
|
||||
|
||||
|
||||
# =============================================
|
||||
# PHASE 1: DISCOVERY
|
||||
# =============================================
|
||||
|
||||
def _find_test_files(branch_path: Path) -> list[Path]:
|
||||
"""Find all test files for a branch.
|
||||
|
||||
Looks in:
|
||||
- {branch_path}/tests/ (recursive)
|
||||
- Any file matching test_*.py or *_test.py elsewhere in the branch
|
||||
"""
|
||||
test_files: list[Path] = []
|
||||
seen: set[Path] = set()
|
||||
|
||||
# 1. Standard tests/ directory
|
||||
tests_dir = branch_path / "tests"
|
||||
if tests_dir.is_dir():
|
||||
for py_file in sorted(tests_dir.rglob("*.py")):
|
||||
if py_file.name in ("__init__.py", "conftest.py"):
|
||||
continue
|
||||
if "__pycache__" in py_file.parts:
|
||||
continue
|
||||
resolved = py_file.resolve()
|
||||
if resolved not in seen:
|
||||
seen.add(resolved)
|
||||
test_files.append(py_file)
|
||||
|
||||
# 2. Scattered test_*.py or *_test.py anywhere in the branch
|
||||
for py_file in sorted(branch_path.rglob("*.py")):
|
||||
if any(_should_skip_dir(part) for part in py_file.relative_to(branch_path).parts):
|
||||
continue
|
||||
if py_file.name in ("__init__.py", "conftest.py"):
|
||||
continue
|
||||
if py_file.name.startswith("test_") or py_file.name.endswith("_test.py"):
|
||||
resolved = py_file.resolve()
|
||||
if resolved not in seen:
|
||||
seen.add(resolved)
|
||||
test_files.append(py_file)
|
||||
|
||||
return test_files
|
||||
|
||||
|
||||
# =============================================
|
||||
# PHASE 2: ANALYZE TEST FILES
|
||||
# =============================================
|
||||
|
||||
def _analyze_test_file(test_file: Path) -> dict:
|
||||
"""Analyze a single test file for test functions and module coverage.
|
||||
|
||||
Returns:
|
||||
dict with keys: path, test_count, test_names, tested_modules
|
||||
"""
|
||||
source = _read_file_safe(test_file)
|
||||
info: dict = {
|
||||
"path": test_file,
|
||||
"test_count": 0,
|
||||
"test_names": [],
|
||||
"tested_modules": set(),
|
||||
}
|
||||
|
||||
if not source:
|
||||
return info
|
||||
|
||||
# Count test functions
|
||||
for match in RE_TEST_FUNC.finditer(source):
|
||||
info["test_names"].append(match.group(1))
|
||||
info["test_count"] = len(info["test_names"])
|
||||
|
||||
# Find which modules this test file covers via imports
|
||||
for match in RE_IMPORT_FROM.finditer(source):
|
||||
sub_path = match.group(1)
|
||||
if sub_path:
|
||||
first_segment = sub_path.split(".")[0]
|
||||
info["tested_modules"].add(first_segment)
|
||||
|
||||
for match in RE_IMPORT_DIRECT.finditer(source):
|
||||
sub_path = match.group(1)
|
||||
if sub_path:
|
||||
first_segment = sub_path.split(".")[0]
|
||||
info["tested_modules"].add(first_segment)
|
||||
|
||||
return info
|
||||
|
||||
|
||||
# =============================================
|
||||
# PHASE 3: COLLECT TESTABLE MODULES
|
||||
# =============================================
|
||||
|
||||
def _collect_testable_modules(branch_path: Path) -> set[str]:
|
||||
"""Collect module names from apps/modules/ and apps/handlers/.
|
||||
|
||||
Returns set of module names:
|
||||
- apps/modules/*.py -> file stem (e.g. "runner")
|
||||
- apps/handlers/*.py -> file stem (e.g. "audit")
|
||||
- apps/handlers/subdir/ -> directory name if it contains .py files
|
||||
"""
|
||||
modules: set[str] = set()
|
||||
apps_dir = branch_path / "apps"
|
||||
if not apps_dir.is_dir():
|
||||
return modules
|
||||
|
||||
# apps/modules/ -- flat .py files
|
||||
modules_dir = apps_dir / "modules"
|
||||
if modules_dir.is_dir():
|
||||
for item in sorted(modules_dir.iterdir()):
|
||||
if item.is_file() and item.suffix == ".py" and item.name != "__init__.py":
|
||||
modules.add(item.stem)
|
||||
|
||||
# apps/handlers/ -- flat .py files OR subdirectories with .py files
|
||||
handlers_dir = apps_dir / "handlers"
|
||||
if handlers_dir.is_dir():
|
||||
for item in sorted(handlers_dir.iterdir()):
|
||||
if _should_skip_dir(item.name):
|
||||
continue
|
||||
if item.is_dir() and item.name != "__pycache__":
|
||||
has_py = any(
|
||||
f.suffix == ".py" and f.name != "__init__.py"
|
||||
for f in item.iterdir()
|
||||
if f.is_file()
|
||||
)
|
||||
if has_py:
|
||||
modules.add(item.name)
|
||||
elif item.is_file() and item.suffix == ".py" and item.name != "__init__.py":
|
||||
modules.add(item.stem)
|
||||
|
||||
return modules
|
||||
|
||||
|
||||
# =============================================
|
||||
# PHASE 4: BRANCH-LEVEL CHECK
|
||||
# =============================================
|
||||
|
||||
def check_branch(branch_path: str, bypass_rules: list | None = None) -> dict:
|
||||
"""Run test coverage analysis on a branch.
|
||||
|
||||
Args:
|
||||
branch_path: Path to branch root directory
|
||||
bypass_rules: Optional list of bypass rules
|
||||
|
||||
Returns:
|
||||
dict: {passed, score, checks, standard: 'TEST_COVERAGE'}
|
||||
"""
|
||||
checks: list[dict] = []
|
||||
bp = Path(branch_path)
|
||||
|
||||
# Check if entire standard is bypassed
|
||||
if is_bypassed(branch_path, "test_coverage", bypass_rules=bypass_rules):
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Bypassed",
|
||||
"passed": True,
|
||||
"message": "Standard bypassed via .seedgo/bypass.json",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "TEST_COVERAGE",
|
||||
}
|
||||
|
||||
# Validate branch path exists
|
||||
if not bp.is_dir():
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Branch exists",
|
||||
"passed": False,
|
||||
"message": f"Branch directory not found: {branch_path}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "TEST_COVERAGE",
|
||||
}
|
||||
|
||||
# Phase 1: Find test files
|
||||
test_files = _find_test_files(bp)
|
||||
|
||||
# Phase 2: Analyze each test file
|
||||
total_tests = 0
|
||||
tested_modules: set[str] = set()
|
||||
for tf in test_files:
|
||||
info = _analyze_test_file(tf)
|
||||
total_tests += info["test_count"]
|
||||
tested_modules.update(info["tested_modules"])
|
||||
|
||||
# Clear tested modules if no actual test functions found
|
||||
if total_tests == 0:
|
||||
tested_modules = set()
|
||||
|
||||
# Phase 3: Collect all testable modules
|
||||
all_modules = _collect_testable_modules(bp)
|
||||
total_modules = len(all_modules)
|
||||
|
||||
# Phase 4: Calculate coverage
|
||||
if total_modules > 0:
|
||||
covered_count = len(tested_modules & all_modules)
|
||||
coverage_pct = (covered_count / total_modules) * 100
|
||||
else:
|
||||
covered_count = 0
|
||||
coverage_pct = 0.0
|
||||
|
||||
# -- Check 1: Test files exist --
|
||||
if test_files:
|
||||
checks.append({
|
||||
"name": "Test files",
|
||||
"passed": True,
|
||||
"message": f"Found {len(test_files)} test file(s)",
|
||||
})
|
||||
else:
|
||||
checks.append({
|
||||
"name": "Test files",
|
||||
"passed": False,
|
||||
"message": "No test files found (expected tests/ dir or test_*.py files)",
|
||||
})
|
||||
|
||||
# -- Check 2: Test functions --
|
||||
if total_tests > 0:
|
||||
checks.append({
|
||||
"name": "Test functions",
|
||||
"passed": True,
|
||||
"message": f"Found {total_tests} test function(s)",
|
||||
})
|
||||
else:
|
||||
checks.append({
|
||||
"name": "Test functions",
|
||||
"passed": False,
|
||||
"message": "No test functions found (expected def test_* functions)",
|
||||
})
|
||||
|
||||
# -- Check 3: Module coverage --
|
||||
# Lenient threshold: 25% -- most branches have no tests yet
|
||||
coverage_threshold = 25
|
||||
if total_modules == 0:
|
||||
checks.append({
|
||||
"name": "Module coverage",
|
||||
"passed": True,
|
||||
"message": "No testable modules found (nothing to test)",
|
||||
})
|
||||
elif coverage_pct >= coverage_threshold:
|
||||
checks.append({
|
||||
"name": "Module coverage",
|
||||
"passed": True,
|
||||
"message": f"{covered_count}/{total_modules} modules covered ({coverage_pct:.0f}%)",
|
||||
})
|
||||
else:
|
||||
checks.append({
|
||||
"name": "Module coverage",
|
||||
"passed": False,
|
||||
"message": (
|
||||
f"{covered_count}/{total_modules} modules covered ({coverage_pct:.0f}%) "
|
||||
f"-- below {coverage_threshold}% threshold"
|
||||
),
|
||||
})
|
||||
|
||||
# Calculate score
|
||||
# If branch has 0 testable modules, score = 100 (nothing to test)
|
||||
if total_modules == 0:
|
||||
score = 100
|
||||
else:
|
||||
score = int((covered_count / total_modules) * 100)
|
||||
|
||||
# Overall pass at 75% score threshold
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{
|
||||
"branch": branch_path,
|
||||
"score": score,
|
||||
"standard": "test_coverage",
|
||||
"total_tests": total_tests,
|
||||
"covered_modules": covered_count,
|
||||
"total_modules": total_modules,
|
||||
},
|
||||
)
|
||||
|
||||
return {
|
||||
"passed": overall_passed,
|
||||
"score": score,
|
||||
"checks": checks,
|
||||
"standard": "TEST_COVERAGE",
|
||||
}
|
||||
@@ -0,0 +1,212 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: todo_check.py
|
||||
# Description: TODO/FIXME Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
TODO/FIXME Standards Checker Handler
|
||||
|
||||
Detects TODO, FIXME, HACK, and XXX comments in Python source files.
|
||||
These tags indicate incomplete work, known hacks, or code needing attention.
|
||||
Each file is scored: passed if zero tags found, failed otherwise with
|
||||
a count and tag breakdown in the message.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "all_files"
|
||||
|
||||
# Tags to detect, case-insensitive
|
||||
_TAGS = ("TODO", "FIXME", "HACK", "XXX")
|
||||
|
||||
# Regex: matches # TODO: text, # FIXME(user): text, inline # HACK ..., etc.
|
||||
# Captures (tag, comment_text). Handles inline and standalone comments.
|
||||
_TAG_RE = re.compile(
|
||||
r"#\s*(TODO|FIXME|HACK|XXX)\b[:\s]*(.*)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
# Must match standard
|
||||
if rule.get("standard") and rule.get("standard") != standard:
|
||||
continue
|
||||
# Must match file (check if rule file path is in the full path)
|
||||
rule_file = rule.get("file", "")
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
# Check line-specific bypass
|
||||
rule_lines = rule.get("lines", [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
|
||||
"""
|
||||
Check a Python file for TODO/FIXME/HACK/XXX comments.
|
||||
|
||||
Args:
|
||||
module_path: Path to the Python file to check.
|
||||
bypass_rules: Optional list of bypass rules to skip certain checks.
|
||||
|
||||
Returns:
|
||||
dict: {
|
||||
'passed': bool,
|
||||
'checks': [{'name': str, 'passed': bool, 'message': str}],
|
||||
'score': int,
|
||||
'standard': str,
|
||||
}
|
||||
"""
|
||||
path = Path(module_path)
|
||||
|
||||
# Skip __init__.py files
|
||||
if path.name == "__init__.py":
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "TODO/FIXME comments",
|
||||
"passed": True,
|
||||
"message": "__init__.py skipped",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "TODO",
|
||||
}
|
||||
|
||||
# Check if entire standard is bypassed for this file
|
||||
if is_bypassed(module_path, "todo", bypass_rules=bypass_rules):
|
||||
return {
|
||||
"passed": True,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Bypassed",
|
||||
"passed": True,
|
||||
"message": "Standard bypassed via .seedgo/bypass.json",
|
||||
}
|
||||
],
|
||||
"score": 100,
|
||||
"standard": "TODO",
|
||||
}
|
||||
|
||||
# Validate file exists
|
||||
if not path.exists():
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "File exists",
|
||||
"passed": False,
|
||||
"message": f"File not found: {module_path}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "TODO",
|
||||
}
|
||||
|
||||
# Read file
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
except Exception as e:
|
||||
return {
|
||||
"passed": False,
|
||||
"checks": [
|
||||
{
|
||||
"name": "File readable",
|
||||
"passed": False,
|
||||
"message": f"Error reading file: {e}",
|
||||
}
|
||||
],
|
||||
"score": 0,
|
||||
"standard": "TODO",
|
||||
}
|
||||
|
||||
# Scan for TODO/FIXME/HACK/XXX tags in comment lines
|
||||
tag_counts: dict[str, int] = {}
|
||||
total_found = 0
|
||||
in_docstring = False
|
||||
docstring_char: str | None = None
|
||||
|
||||
for line in content.splitlines():
|
||||
stripped = line.strip()
|
||||
|
||||
# Track docstring boundaries
|
||||
if not in_docstring:
|
||||
if stripped.startswith('"""') or stripped.startswith("'''"):
|
||||
docstring_char = stripped[:3]
|
||||
# Single-line docstring: opens and closes on the same line
|
||||
if stripped.count(docstring_char) >= 2:
|
||||
continue
|
||||
in_docstring = True
|
||||
continue
|
||||
else:
|
||||
if docstring_char and docstring_char in stripped:
|
||||
in_docstring = False
|
||||
continue
|
||||
|
||||
# Only match in comment portions (the regex already requires #)
|
||||
match = _TAG_RE.search(line)
|
||||
if match:
|
||||
tag = match.group(1).upper()
|
||||
tag_counts[tag] = tag_counts.get(tag, 0) + 1
|
||||
total_found += 1
|
||||
|
||||
# Build the single check result
|
||||
if total_found == 0:
|
||||
check_result = {
|
||||
"name": "TODO/FIXME comments",
|
||||
"passed": True,
|
||||
"message": "No TODO/FIXME/HACK/XXX comments found",
|
||||
}
|
||||
else:
|
||||
breakdown = ", ".join(
|
||||
f"{tag}: {count}"
|
||||
for tag in _TAGS
|
||||
if (count := tag_counts.get(tag, 0)) > 0
|
||||
)
|
||||
check_result = {
|
||||
"name": "TODO/FIXME comments",
|
||||
"passed": False,
|
||||
"message": f"Found {total_found} TODO-type comment{'s' if total_found != 1 else ''} ({breakdown})",
|
||||
}
|
||||
|
||||
checks = [check_result]
|
||||
|
||||
# Score: one check, so 100 if passed, 0 if failed
|
||||
passed_checks = sum(1 for c in checks if c["passed"])
|
||||
total_checks = len(checks)
|
||||
score = int((passed_checks / total_checks) * 100) if total_checks > 0 else 0
|
||||
|
||||
overall_passed = score >= 75
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"file": str(module_path), "score": score, "standard": "todo"},
|
||||
)
|
||||
return {
|
||||
"passed": overall_passed,
|
||||
"checks": checks,
|
||||
"score": score,
|
||||
"standard": "TODO",
|
||||
}
|
||||
@@ -0,0 +1,355 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: unused_function_check.py
|
||||
# Description: Unused Function Standards Checker Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-22
|
||||
# Modified: 2026-03-22
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Unused Function Standards Checker Handler
|
||||
|
||||
Branch-level checker that detects functions defined but never referenced
|
||||
elsewhere in the branch. Uses AST parsing to extract function definitions
|
||||
and corpus-level text search to count references.
|
||||
|
||||
A function is flagged as unused when its name appears only in its own
|
||||
definition line (i.e., it is never called or imported elsewhere in the
|
||||
branch source).
|
||||
|
||||
Excluded from flagging:
|
||||
- Dunder methods (__init__, __str__, __repr__, etc.)
|
||||
- main(), handle_command() -- framework/entry-point conventions
|
||||
- Any function with a decorator (@property, @staticmethod, etc.)
|
||||
"""
|
||||
|
||||
import ast
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
AUDIT_SCOPE = "branch_level"
|
||||
|
||||
# -- Directories to skip when collecting source files -------------------------
|
||||
SKIP_DIRS = {
|
||||
"__pycache__", ".archive", "logs", "tests",
|
||||
"json_templates", "tools", ".trinity", ".aipass", ".ai_mail.local",
|
||||
".venv", "venv", "node_modules", ".git", "site-packages",
|
||||
".mypy_cache", ".ruff_cache", ".pytest_cache", ".spawn",
|
||||
"backups", "reports", "docs", ".sorting_unprocessed",
|
||||
}
|
||||
|
||||
# -- Function names excluded from analysis ------------------------------------
|
||||
EXCLUDED_NAMES = {
|
||||
"main",
|
||||
"handle_command",
|
||||
}
|
||||
|
||||
# -- Regex helpers for corpus stripping ---------------------------------------
|
||||
|
||||
# Matches triple-quoted string literals (both ''' and """), including content.
|
||||
_TRIPLE_QUOTED_RE = re.compile(
|
||||
r'""".*?"""|\'\'\'.*?\'\'\'',
|
||||
re.DOTALL,
|
||||
)
|
||||
|
||||
# Matches single-line comments.
|
||||
_COMMENT_RE = re.compile(r"#[^\n]*")
|
||||
|
||||
# Matches `if __name__ == "__main__":` through end of file.
|
||||
_MAIN_BLOCK_RE = re.compile(
|
||||
r"""^if\s+__name__\s*==\s*["']__main__["']\s*:.*""",
|
||||
re.MULTILINE | re.DOTALL,
|
||||
)
|
||||
|
||||
|
||||
# -- Bypass helper ------------------------------------------------------------
|
||||
|
||||
def is_bypassed(
|
||||
file_path: str,
|
||||
standard: str,
|
||||
line: int | None = None,
|
||||
bypass_rules: list | None = None,
|
||||
) -> bool:
|
||||
"""Check if a violation should be bypassed."""
|
||||
if not bypass_rules:
|
||||
return False
|
||||
for rule in bypass_rules:
|
||||
if rule.get("standard") and rule.get("standard") != standard:
|
||||
continue
|
||||
rule_file = rule.get("file", "")
|
||||
if rule_file and rule_file not in file_path:
|
||||
continue
|
||||
rule_lines = rule.get("lines", [])
|
||||
if rule_lines and line is not None:
|
||||
if line in rule_lines:
|
||||
return True
|
||||
elif not rule_lines:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# -- File collection ----------------------------------------------------------
|
||||
|
||||
def _should_skip(path: Path) -> bool:
|
||||
"""Return True if any path component is in the skip set."""
|
||||
return any(part in SKIP_DIRS for part in path.parts)
|
||||
|
||||
|
||||
def _collect_python_files(branch_path: Path) -> list[Path]:
|
||||
"""Collect all .py files in the branch, skipping irrelevant dirs."""
|
||||
files: list[Path] = []
|
||||
if not branch_path.is_dir():
|
||||
return files
|
||||
for py_file in branch_path.rglob("*.py"):
|
||||
if _should_skip(py_file):
|
||||
continue
|
||||
files.append(py_file)
|
||||
return sorted(files)
|
||||
|
||||
|
||||
# -- Corpus preparation ------------------------------------------------------
|
||||
|
||||
def _strip_non_code(source: str) -> str:
|
||||
"""
|
||||
Remove triple-quoted strings, comments, and __main__ blocks.
|
||||
|
||||
Prevents doctest lines, commented-out code, and demo invocations
|
||||
from inflating reference counts.
|
||||
"""
|
||||
source = _TRIPLE_QUOTED_RE.sub("", source)
|
||||
source = _COMMENT_RE.sub("", source)
|
||||
source = _MAIN_BLOCK_RE.sub("", source)
|
||||
return source
|
||||
|
||||
|
||||
# -- AST function extraction --------------------------------------------------
|
||||
|
||||
def _is_excluded(name: str) -> bool:
|
||||
"""Return True if this function name should never be flagged."""
|
||||
if name.startswith("__") and name.endswith("__"):
|
||||
return True
|
||||
if name in EXCLUDED_NAMES:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _has_decorator(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
|
||||
"""Return True if the function has any decorator."""
|
||||
return len(node.decorator_list) > 0
|
||||
|
||||
|
||||
def _extract_functions(py_file: Path) -> list[tuple[str, int]]:
|
||||
"""
|
||||
Parse a .py file with AST and return (function_name, line_number) for each
|
||||
FunctionDef / AsyncFunctionDef that is not excluded and has no decorators.
|
||||
"""
|
||||
try:
|
||||
source = py_file.read_text(encoding="utf-8", errors="ignore")
|
||||
tree = ast.parse(source, filename=str(py_file))
|
||||
except SyntaxError:
|
||||
return []
|
||||
|
||||
results: list[tuple[str, int]] = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
if _is_excluded(node.name):
|
||||
continue
|
||||
if _has_decorator(node):
|
||||
continue
|
||||
results.append((node.name, node.lineno))
|
||||
return results
|
||||
|
||||
|
||||
# -- Reference counting -------------------------------------------------------
|
||||
|
||||
def _count_references_in_corpus(func_name: str, corpus: str) -> int:
|
||||
"""
|
||||
Count how many times func_name appears as a word-bounded identifier
|
||||
in the corpus (docstrings and comments already stripped).
|
||||
|
||||
Counts ALL occurrences including `def func_name(` lines.
|
||||
"""
|
||||
pattern = re.compile(rf"\b{re.escape(func_name)}\b")
|
||||
return len(pattern.findall(corpus))
|
||||
|
||||
|
||||
def _count_def_lines(func_name: str, corpus: str) -> int:
|
||||
"""
|
||||
Count definition lines (def func_name / async def func_name) in corpus.
|
||||
"""
|
||||
pattern = re.compile(
|
||||
rf"\basync\s+def\s+{re.escape(func_name)}\b"
|
||||
rf"|\bdef\s+{re.escape(func_name)}\b"
|
||||
)
|
||||
return len(pattern.findall(corpus))
|
||||
|
||||
|
||||
# -- Branch-level check (audit pipeline entry) --------------------------------
|
||||
|
||||
def check_branch(branch_path: str, bypass_rules: list | None = None) -> dict:
|
||||
"""
|
||||
Check a branch for unused function definitions.
|
||||
|
||||
Scans all .py files, builds a text corpus, extracts function definitions
|
||||
via AST, and flags any function whose name appears only in its own
|
||||
definition (no call references elsewhere).
|
||||
|
||||
Args:
|
||||
branch_path: Path to branch root directory.
|
||||
bypass_rules: Optional list of bypass rules from .seedgo/bypass.json.
|
||||
|
||||
Returns:
|
||||
dict with keys: passed, score, checks, standard.
|
||||
"""
|
||||
branch = Path(branch_path)
|
||||
|
||||
# Check if entire standard is bypassed
|
||||
if is_bypassed(branch_path, "unused_function", bypass_rules=bypass_rules):
|
||||
return {
|
||||
"passed": True,
|
||||
"score": 100,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Bypassed",
|
||||
"passed": True,
|
||||
"message": "Standard bypassed via .seedgo/bypass.json",
|
||||
}
|
||||
],
|
||||
"standard": "UNUSED_FUNCTION",
|
||||
}
|
||||
|
||||
# Phase 1: Collect all .py files
|
||||
py_files = _collect_python_files(branch)
|
||||
if not py_files:
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"branch": branch_path, "score": 100, "standard": "unused_function"},
|
||||
)
|
||||
return {
|
||||
"passed": True,
|
||||
"score": 100,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Unused functions",
|
||||
"passed": True,
|
||||
"message": "No .py files found in branch",
|
||||
}
|
||||
],
|
||||
"standard": "UNUSED_FUNCTION",
|
||||
}
|
||||
|
||||
# Phase 2: Build cleaned text corpus from all files
|
||||
file_sources: dict[Path, str] = {}
|
||||
for py_file in py_files:
|
||||
try:
|
||||
raw = py_file.read_text(encoding="utf-8", errors="ignore")
|
||||
except OSError:
|
||||
continue
|
||||
file_sources[py_file] = _strip_non_code(raw)
|
||||
|
||||
corpus = "\n".join(file_sources.values())
|
||||
|
||||
# Phase 3: Extract function definitions from all files
|
||||
all_functions: list[tuple[str, int, Path]] = []
|
||||
for py_file in py_files:
|
||||
for func_name, lineno in _extract_functions(py_file):
|
||||
all_functions.append((func_name, lineno, py_file))
|
||||
|
||||
total_functions = len(all_functions)
|
||||
|
||||
if total_functions == 0:
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{"branch": branch_path, "score": 100, "standard": "unused_function"},
|
||||
)
|
||||
return {
|
||||
"passed": True,
|
||||
"score": 100,
|
||||
"checks": [
|
||||
{
|
||||
"name": "Unused functions",
|
||||
"passed": True,
|
||||
"message": "No eligible functions found",
|
||||
}
|
||||
],
|
||||
"standard": "UNUSED_FUNCTION",
|
||||
}
|
||||
|
||||
# Phase 4: For each function, check if it has references beyond its definition
|
||||
unused_functions: list[dict] = []
|
||||
|
||||
for func_name, lineno, py_file in all_functions:
|
||||
# Check bypass at file+line level
|
||||
if is_bypassed(str(py_file), "unused_function", lineno, bypass_rules):
|
||||
continue
|
||||
|
||||
total_refs = _count_references_in_corpus(func_name, corpus)
|
||||
def_count = _count_def_lines(func_name, corpus)
|
||||
call_refs = total_refs - def_count
|
||||
|
||||
if call_refs <= 0:
|
||||
try:
|
||||
rel_path = py_file.relative_to(branch)
|
||||
except ValueError:
|
||||
rel_path = py_file
|
||||
unused_functions.append({
|
||||
"name": func_name,
|
||||
"file": str(rel_path),
|
||||
"line": lineno,
|
||||
})
|
||||
|
||||
# Score: clean_functions / total_functions * 100
|
||||
clean_count = total_functions - len(unused_functions)
|
||||
score = int(clean_count / total_functions * 100) if total_functions > 0 else 100
|
||||
passed = score >= 75
|
||||
|
||||
# Build check entry
|
||||
if unused_functions:
|
||||
# Build a summary of unused functions (cap at 15 for readability)
|
||||
details = [
|
||||
f" {uf['name']} ({uf['file']}:{uf['line']})"
|
||||
for uf in unused_functions[:15]
|
||||
]
|
||||
if len(unused_functions) > 15:
|
||||
details.append(f" ... and {len(unused_functions) - 15} more")
|
||||
detail_text = "\n".join(details)
|
||||
|
||||
checks = [
|
||||
{
|
||||
"name": "Unused functions",
|
||||
"passed": passed,
|
||||
"message": (
|
||||
f"{len(unused_functions)} unused out of {total_functions} "
|
||||
f"functions ({score}% clean)\n{detail_text}"
|
||||
),
|
||||
"unused": unused_functions,
|
||||
}
|
||||
]
|
||||
else:
|
||||
checks = [
|
||||
{
|
||||
"name": "Unused functions",
|
||||
"passed": True,
|
||||
"message": f"All {total_functions} functions are referenced",
|
||||
}
|
||||
]
|
||||
|
||||
json_handler.log_operation(
|
||||
"check_completed",
|
||||
{
|
||||
"branch": branch_path,
|
||||
"score": score,
|
||||
"standard": "unused_function",
|
||||
"total_functions": total_functions,
|
||||
"unused_count": len(unused_functions),
|
||||
},
|
||||
)
|
||||
|
||||
return {
|
||||
"passed": passed,
|
||||
"score": score,
|
||||
"checks": checks,
|
||||
"standard": "UNUSED_FUNCTION",
|
||||
}
|
||||
@@ -142,11 +142,22 @@ def print_branch_summary(audit_result: Dict, system_averages: Dict[str, int] | N
|
||||
for check in other_failures:
|
||||
console.print(f" [dim]• {check.get('message', '')}[/dim]")
|
||||
else:
|
||||
# Non-architecture failures - show as before
|
||||
console.print(f" [red]└─ {standard_name.title()} issues:[/red]")
|
||||
for check in result.get('checks', []):
|
||||
if not check.get('passed', False):
|
||||
console.print(f" [dim]• {check.get('message', 'Unknown error')}[/dim]")
|
||||
# Non-architecture failures — check for per-file violations first
|
||||
per_file_violations = audit_result.get(f'{standard_name}_violations', [])
|
||||
if per_file_violations:
|
||||
console.print(f" [red]└─ {standard_name.replace('_', ' ').title()} issues:[/red]")
|
||||
for violation in per_file_violations:
|
||||
file_path = violation.get('path', violation.get('file', ''))
|
||||
score = violation.get('score', 0)
|
||||
console.print(f" [red]✗[/red] [magenta]{file_path}[/magenta] [dim](score: {score}%)[/dim]")
|
||||
for issue in violation.get('issues', []):
|
||||
console.print(f" [dim]• {issue}[/dim]")
|
||||
else:
|
||||
# Fallback: show check messages directly
|
||||
console.print(f" [red]└─ {standard_name.replace('_', ' ').title()} issues:[/red]")
|
||||
for check in result.get('checks', []):
|
||||
if not check.get('passed', False):
|
||||
console.print(f" [dim]• {check.get('message', 'Unknown error')}[/dim]")
|
||||
|
||||
# Always show CLI violations (audit = comprehensive)
|
||||
# Use absolute paths for reliable VS Code clickability
|
||||
|
||||
@@ -16,6 +16,7 @@ Run: seedgo audit
|
||||
"""
|
||||
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
from collections import defaultdict
|
||||
@@ -204,7 +205,7 @@ def handle_command(command: str, args: List[str]) -> bool:
|
||||
|
||||
# Validate pack name
|
||||
packs = _discover_packs()
|
||||
if pack_name not in packs:
|
||||
if pack_name is None or pack_name not in packs:
|
||||
available = ", ".join(packs.keys())
|
||||
error(
|
||||
f"Unknown pack: '{pack_name}'",
|
||||
@@ -254,20 +255,55 @@ def handle_command(command: str, args: List[str]) -> bool:
|
||||
console.print(f"[red]Branch '{specific_branch}' not found[/red]")
|
||||
return True
|
||||
|
||||
console.print(f"[dim]Discovered {len(branches)} branches to audit...[/dim]")
|
||||
from rich.progress import Progress, BarColumn, TextColumn, TimeRemainingColumn, SpinnerColumn
|
||||
|
||||
# Audit all branches (always full - checks all files)
|
||||
is_compact = specific_branch is None # Full audit = compact, single branch = detailed
|
||||
|
||||
total_branches = len(branches)
|
||||
console.print(f"[dim]Discovered {total_branches} branch{'es' if total_branches != 1 else ''} to audit...[/dim]")
|
||||
console.print()
|
||||
|
||||
# Audit all branches with live progress
|
||||
audit_results = []
|
||||
for branch in branches:
|
||||
console.print(f"[dim]Auditing {branch['name']}...[/dim]", end="\r")
|
||||
audit_start = time.monotonic()
|
||||
|
||||
# Load bypass rules for this branch
|
||||
bypass_rules = load_bypass_rules(branch['path'])
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
BarColumn(),
|
||||
TextColumn("{task.completed}/{task.total}"),
|
||||
TimeRemainingColumn(),
|
||||
console=console,
|
||||
transient=True,
|
||||
) as progress:
|
||||
task = progress.add_task("Scanning...", total=total_branches)
|
||||
|
||||
result = audit_branch(branch, bypass_rules, pack_path=pack_path)
|
||||
audit_results.append(result)
|
||||
for idx, branch in enumerate(branches, 1):
|
||||
branch_name = branch['name']
|
||||
progress.update(task, description=f"[cyan]{branch_name}[/cyan]")
|
||||
|
||||
console.print(" " * 50, end="\r") # Clear progress line
|
||||
# Load bypass rules for this branch
|
||||
bypass_rules = load_bypass_rules(branch['path'])
|
||||
|
||||
branch_start = time.monotonic()
|
||||
result = audit_branch(branch, bypass_rules, pack_path=pack_path)
|
||||
branch_elapsed = time.monotonic() - branch_start
|
||||
|
||||
result['elapsed'] = branch_elapsed
|
||||
audit_results.append(result)
|
||||
|
||||
# Print completed branch result (persists above progress bar)
|
||||
avg = result.get('average', 0)
|
||||
style = "green" if avg >= 90 else "yellow" if avg >= 75 else "red"
|
||||
progress.console.print(
|
||||
f" [dim][{idx}/{total_branches}][/dim] [cyan]{branch_name:<12}[/cyan] [{style}]{avg:>3}%[/{style}] [dim]({branch_elapsed:.1f}s)[/dim]"
|
||||
)
|
||||
progress.advance(task)
|
||||
|
||||
total_elapsed = time.monotonic() - audit_start
|
||||
console.print()
|
||||
console.print(f"[dim]Audit complete — {total_branches} branches in {total_elapsed:.1f}s[/dim]")
|
||||
console.print()
|
||||
|
||||
# Calculate system-wide averages for each standard
|
||||
standard_scores = defaultdict(list)
|
||||
@@ -282,12 +318,13 @@ def handle_command(command: str, args: List[str]) -> bool:
|
||||
|
||||
overall_system_avg = int(sum(r['average'] for r in audit_results) / len(audit_results)) if audit_results else 0
|
||||
|
||||
# Print results with system averages
|
||||
for result in audit_results:
|
||||
print_branch_summary(result, system_averages, overall_system_avg)
|
||||
# Print results — detailed for single branch, skip for full audit
|
||||
if not is_compact:
|
||||
for result in audit_results:
|
||||
print_branch_summary(result, system_averages, overall_system_avg)
|
||||
|
||||
# Print system summary (unless specific branch)
|
||||
if not specific_branch:
|
||||
# Print system summary (full audit only)
|
||||
if is_compact:
|
||||
print_system_summary(audit_results)
|
||||
|
||||
# Log completion
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script for audit progress display.
|
||||
|
||||
Run this through drone to test if Rich Progress renders in Patrick's terminal:
|
||||
drone @seedgo test_progress
|
||||
|
||||
Or run directly:
|
||||
python3 src/aipass/seedgo/tests/test_progress_display.py
|
||||
"""
|
||||
import time
|
||||
from rich.console import Console
|
||||
from rich.progress import Progress, BarColumn, TextColumn, TimeRemainingColumn, SpinnerColumn
|
||||
|
||||
console = Console()
|
||||
|
||||
# Fake branch names to simulate audit
|
||||
branches = [
|
||||
"AI_MAIL", "API", "BACKUP", "CLI", "COMMONS",
|
||||
"DAEMON", "DRONE", "FLOW", "MEMORY", "PRAX",
|
||||
"SEEDGO", "SKILLS", "SPAWN", "TRIGGER"
|
||||
]
|
||||
|
||||
|
||||
def test_rich_progress():
|
||||
"""Test 1: Rich Progress bar (same pattern backup uses)"""
|
||||
console.print("\n[bold cyan]Test 1: Rich Progress Bar[/bold cyan]")
|
||||
console.print("[dim]This is what backup uses — should show a moving bar[/dim]\n")
|
||||
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
BarColumn(),
|
||||
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
|
||||
TimeRemainingColumn(),
|
||||
console=console,
|
||||
) as progress:
|
||||
task = progress.add_task("Auditing branches...", total=len(branches))
|
||||
for branch in branches:
|
||||
progress.update(task, description=f"Auditing {branch}...")
|
||||
time.sleep(0.3) # Simulate work
|
||||
progress.advance(task)
|
||||
|
||||
console.print("[green]Done![/green]\n")
|
||||
|
||||
|
||||
def test_rich_progress_with_results():
|
||||
"""Test 2: Progress bar + print completed lines"""
|
||||
console.print("[bold cyan]Test 2: Progress + Per-Branch Results[/bold cyan]")
|
||||
console.print("[dim]Shows progress bar while processing, prints results as they complete[/dim]\n")
|
||||
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
BarColumn(),
|
||||
TextColumn("{task.completed}/{task.total}"),
|
||||
TimeRemainingColumn(),
|
||||
console=console,
|
||||
) as progress:
|
||||
task = progress.add_task("Scanning...", total=len(branches))
|
||||
for idx, branch in enumerate(branches, 1):
|
||||
progress.update(task, description=f"[cyan]{branch}[/cyan]")
|
||||
time.sleep(0.3) # Simulate work
|
||||
|
||||
# Fake score
|
||||
score = 90 + (idx % 5)
|
||||
elapsed = 0.3
|
||||
|
||||
# Print result line (persists above progress bar)
|
||||
style = "green" if score >= 90 else "yellow"
|
||||
progress.console.print(
|
||||
f" [{idx}/{len(branches)}] {branch:<12} [{style}]{score}%[/{style}] ({elapsed:.1f}s)"
|
||||
)
|
||||
progress.advance(task)
|
||||
|
||||
console.print()
|
||||
console.print("[dim]Audit complete[/dim]")
|
||||
console.print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
console.print("\n[bold]Audit Progress Display Tests[/bold]")
|
||||
console.print("[dim]Testing which Rich display method works in this terminal[/dim]\n")
|
||||
|
||||
test_rich_progress()
|
||||
test_rich_progress_with_results()
|
||||
|
||||
console.print("[bold green]All tests complete[/bold green]\n")
|
||||
Reference in New Issue
Block a user