mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-11 12:17:45 +02:00
[hAIk]: add ruff lint check for per-file scope-aware rules (F401 unused imports, F811 redefinitions, F841 unused locals, ARG001/ARG002 unused args) and narrow vulture to whole-program dead-code reachability only — the two are complementary, not redundant; introduce CheckError exception so eslint/knip/ruff/vulture surface tool failures loudly instead of silently returning []; refactor run_checks() from a 7-tuple to a LintResult dataclass with sections + incomplete tracking, and exit non-zero on incomplete runs so CI never mistakes a partial pass for a clean one; update print_errors.sh with ruff section; pin ruff==0.15.17 in requirements-dev.txt; add missing @pytest.mark.asyncio decorators to test_outputs_runtime_cleanup.py and remove the pytest_collection_modifyitems auto-marker from test_phase1_stress.py; set tasks.json reveal to always
This commit is contained in:
Vendored
+5
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"python.defaultInterpreterPath": "${workspaceFolder}/backend/.venv/bin/python",
|
||||
"python.terminal.activateEnvironment": true,
|
||||
"terminal.integrated.enablePersistentSessions": false
|
||||
}
|
||||
Vendored
+1
-1
@@ -16,7 +16,7 @@
|
||||
"runOn": "folderOpen"
|
||||
},
|
||||
"presentation": {
|
||||
"reveal": "never",
|
||||
"reveal": "always",
|
||||
"panel": "dedicated",
|
||||
"close": true
|
||||
},
|
||||
|
||||
@@ -10,6 +10,12 @@
|
||||
pytest==8.3.4
|
||||
pytest-asyncio==0.25.2
|
||||
|
||||
# Used by linter/lint.py (the vulture dead-code check). watchfiles, also
|
||||
# needed by lint.py, already comes in transitively via uvicorn[standard].
|
||||
# Used by linter/lint.py. watchfiles, also needed by lint.py, already comes in
|
||||
# transitively via uvicorn[standard].
|
||||
# - vulture: whole-program dead code (dead functions/methods/classes/attrs).
|
||||
# - ruff: per-file/per-scope checks (F401 unused imports, F811 redefinitions,
|
||||
# F841 unused locals, ARG001/ARG002 unused args) that ruff does AST-accurately
|
||||
# and vulture either gets wrong or rates as noise. vulture is narrowed to defer
|
||||
# imports/locals to ruff (see linter/checks/vulture.py).
|
||||
vulture==2.16
|
||||
ruff==0.15.17
|
||||
|
||||
@@ -20,6 +20,8 @@ import sys
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
|
||||
|
||||
from backend.apps.outputs.runtime import (
|
||||
@@ -115,6 +117,7 @@ def test_write_env_value():
|
||||
|
||||
|
||||
# --- Test 2: stop_all reaps an active runtime (real spawn). ---
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_all_kills_active():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
port = _find_free_port()
|
||||
@@ -151,6 +154,7 @@ async def test_stop_all_kills_active():
|
||||
|
||||
|
||||
# --- Test 3: stop_all reaps an idle (LRU + SIGSTOP'd) runtime. ---
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_all_kills_idle():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
port = _find_free_port()
|
||||
@@ -182,6 +186,7 @@ async def test_stop_all_kills_idle():
|
||||
|
||||
|
||||
# --- Test 4: persisted port collision triggers .env rewrite + new spawn. ---
|
||||
@pytest.mark.asyncio
|
||||
async def test_port_collision_reallocates_env():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
squatted_port = _find_free_port()
|
||||
@@ -215,6 +220,7 @@ async def test_port_collision_reallocates_env():
|
||||
|
||||
|
||||
# --- Test 5: stop_all is idempotent. ---
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_all_idempotent():
|
||||
m = AppRuntimeManager()
|
||||
n = await m.stop_all()
|
||||
@@ -225,6 +231,7 @@ async def test_stop_all_idempotent():
|
||||
|
||||
|
||||
# --- Test 6: vite-like grandchild dies even with EXIT-only trap. ---
|
||||
@pytest.mark.asyncio
|
||||
async def test_descendant_tree_killed_despite_exit_only_trap():
|
||||
"""Regression for the actual prod bug: webapp_template run.sh has only
|
||||
`trap cleanup EXIT` (no TERM), so a flat SIGTERM to bash exits bash
|
||||
|
||||
@@ -279,15 +279,3 @@ async def test_concurrent_send_message_unique_client_ids():
|
||||
actual = [p[1] for p in pairs]
|
||||
assert expected == actual, "client_message_id must round-trip exactly"
|
||||
assert len(set(actual)) == 100, "all unique"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pytest config: register asyncio mode so we don't need the plugin.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(config, items):
|
||||
"""Auto-mark async tests so they run under pytest-asyncio."""
|
||||
for item in items:
|
||||
if asyncio.iscoroutinefunction(getattr(item, "function", None)):
|
||||
item.add_marker(pytest.mark.asyncio)
|
||||
|
||||
@@ -9,6 +9,21 @@ from pathlib import Path
|
||||
LINTIGNORE_PREFIX = ".lintignore"
|
||||
|
||||
|
||||
class CheckError(Exception):
|
||||
"""Raised when a check could not complete (missing tool, timeout, crash).
|
||||
|
||||
This is deliberately distinct from "the check ran and found zero problems":
|
||||
the orchestrator catches it and surfaces the failure loudly so a partial run
|
||||
is never silently reported as a clean one. Returning ``[]`` on failure (the
|
||||
old behavior) made a timed-out or crashed tool look identical to a passing
|
||||
one, which let the reported error count silently undercount the real total.
|
||||
"""
|
||||
|
||||
def __init__(self, reason: str) -> None:
|
||||
super().__init__(reason)
|
||||
self.reason = reason
|
||||
|
||||
|
||||
def _matches_any(text: str, patterns: list[str]) -> bool:
|
||||
return any(fnmatch.fnmatch(text, p) for p in patterns)
|
||||
|
||||
|
||||
+12
-7
@@ -6,7 +6,9 @@ import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_lintignored
|
||||
from . import CheckError, is_lintignored
|
||||
|
||||
_TIMEOUT = 120
|
||||
|
||||
|
||||
def run_eslint(root: Path, ignores: dict[Path, set[str]] | None = None) -> list[str]:
|
||||
@@ -14,21 +16,24 @@ def run_eslint(root: Path, ignores: dict[Path, set[str]] | None = None) -> list[
|
||||
frontend_dir = root / "frontend"
|
||||
eslint_bin = frontend_dir / "node_modules" / ".bin" / "eslint"
|
||||
if not eslint_bin.exists():
|
||||
return []
|
||||
raise CheckError("eslint binary not found (run npm install in frontend/)")
|
||||
|
||||
cmd = [str(eslint_bin), "src/", "--format", "json", "--no-warn-ignored"]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True,
|
||||
cwd=str(frontend_dir), timeout=60,
|
||||
cwd=str(frontend_dir), timeout=_TIMEOUT,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return []
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise CheckError(f"timed out after {_TIMEOUT}s") from e
|
||||
except OSError as e:
|
||||
raise CheckError(f"failed to launch eslint ({e})") from e
|
||||
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return []
|
||||
except (json.JSONDecodeError, ValueError) as e:
|
||||
detail = result.stderr.strip()[:300] or "non-JSON output"
|
||||
raise CheckError(f"eslint produced unparseable output: {detail}") from e
|
||||
|
||||
errors: list[str] = []
|
||||
for entry in data:
|
||||
|
||||
+12
-7
@@ -6,7 +6,9 @@ import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_lintignored
|
||||
from . import CheckError, is_lintignored
|
||||
|
||||
_TIMEOUT = 120
|
||||
|
||||
KIND_LABELS = {
|
||||
"dependencies": "Unused dependency",
|
||||
@@ -25,21 +27,24 @@ def run_knip(root: Path, ignores: dict[Path, set[str]] | None = None) -> list[st
|
||||
frontend_dir = root / "frontend"
|
||||
knip_bin = frontend_dir / "node_modules" / ".bin" / "knip"
|
||||
if not knip_bin.exists():
|
||||
return []
|
||||
raise CheckError("knip binary not found (run npm install in frontend/)")
|
||||
|
||||
cmd = [str(knip_bin), "--reporter", "json"]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True,
|
||||
cwd=str(frontend_dir), timeout=60,
|
||||
cwd=str(frontend_dir), timeout=_TIMEOUT,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return []
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise CheckError(f"timed out after {_TIMEOUT}s") from e
|
||||
except OSError as e:
|
||||
raise CheckError(f"failed to launch knip ({e})") from e
|
||||
|
||||
try:
|
||||
data = json.loads(result.stdout)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
return []
|
||||
except (json.JSONDecodeError, ValueError) as e:
|
||||
detail = result.stderr.strip()[:300] or "non-JSON output"
|
||||
raise CheckError(f"knip produced unparseable output: {detail}") from e
|
||||
|
||||
errors: list[str] = []
|
||||
for entry in data.get("issues", []):
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Ruff runner: per-file, scope-aware lint that vulture's global name-set can't do.
|
||||
|
||||
Owns the checks ruff is strictly better at than vulture:
|
||||
- F401 unused imports (real per-file scope; honors __all__ / redundant-alias re-exports)
|
||||
- F811 redefinition of an unused name
|
||||
- F841 unused local variable assignment
|
||||
- ARG001/ARG002 unused function/method arguments
|
||||
|
||||
Vulture is narrowed (see checks/vulture.py) to only emit dead functions, methods,
|
||||
classes, and attributes, the whole-program reachability that ruff structurally
|
||||
does not attempt. The two are complementary, not redundant.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from . import CheckError, is_excepted, is_lintignored
|
||||
|
||||
# Generous because the first run after a window reload races the editor's
|
||||
# startup load with a cold ruff cache; a tight limit there is exactly what made
|
||||
# the check time out and silently report zero.
|
||||
_TIMEOUT = 120
|
||||
|
||||
# Ruff emits ANSI color into its concise output even on a non-tty / NO_COLOR;
|
||||
# strip the escapes before parsing so the regex sees plain text.
|
||||
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
|
||||
|
||||
# `path:line:col: CODE message`. The optional `[*]` fixable marker is dropped.
|
||||
_LINE_RE = re.compile(r"^(?P<path>.+?):(?P<line>\d+):(?P<col>\d+): (?P<code>[A-Z]+\d+) (?P<msg>.+)$")
|
||||
|
||||
|
||||
def run_ruff(
|
||||
root: Path,
|
||||
select: str,
|
||||
exceptions: dict[str, list[str]],
|
||||
ignores: dict[Path, set[str]] | None = None,
|
||||
) -> list[str]:
|
||||
"""Run ruff on the Python backend and return errors."""
|
||||
ruff_bin = root / "backend" / ".venv" / "bin" / "ruff"
|
||||
if not ruff_bin.exists():
|
||||
found = shutil.which("ruff")
|
||||
if not found:
|
||||
raise CheckError("ruff executable not found in backend/.venv/bin or PATH")
|
||||
ruff_bin = Path(found)
|
||||
|
||||
targets = ["backend"]
|
||||
if (root / "debug.py").exists():
|
||||
targets.append("debug.py")
|
||||
cmd = [
|
||||
str(ruff_bin), "check", *targets,
|
||||
"--isolated", # ignore any stray pyproject/ruff.toml so the linter is hermetic
|
||||
"--select", select,
|
||||
"--output-format", "concise",
|
||||
"--no-fix",
|
||||
"--exclude", ".venv,__pycache__,data,uv-bin,webapp_template",
|
||||
]
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True, cwd=str(root), timeout=_TIMEOUT,
|
||||
)
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise CheckError(f"timed out after {_TIMEOUT}s (machine under load or cold cache)") from e
|
||||
except OSError as e:
|
||||
raise CheckError(f"failed to launch ruff ({e})") from e
|
||||
|
||||
# ruff exits 0 (no findings) or 1 (findings) on success; anything else with
|
||||
# no parseable output means ruff itself errored (e.g. could not write its
|
||||
# cache) — surface it instead of treating the empty stdout as "clean".
|
||||
if result.returncode not in (0, 1) and not result.stdout.strip():
|
||||
detail = _ANSI_RE.sub("", result.stderr).strip()[:300] or "no output"
|
||||
raise CheckError(f"ruff exited with code {result.returncode}: {detail}")
|
||||
|
||||
errors: list[str] = []
|
||||
for line in result.stdout.strip().splitlines():
|
||||
m = _LINE_RE.match(_ANSI_RE.sub("", line).strip())
|
||||
if not m:
|
||||
continue
|
||||
filepath = m.group("path")
|
||||
if is_excepted(filepath, "ruff", exceptions):
|
||||
continue
|
||||
if ignores and is_lintignored(root / filepath, root, "ruff", ignores):
|
||||
continue
|
||||
errors.append(
|
||||
f"{filepath}:{m.group('line')}:{m.group('col')}: "
|
||||
f"error: [ruff] {m.group('code')} {m.group('msg')}"
|
||||
)
|
||||
return errors
|
||||
@@ -1,5 +1,11 @@
|
||||
"""Vulture dead-code detection runner.
|
||||
|
||||
Scoped to whole-program reachability only: dead functions, methods, classes,
|
||||
and attributes. Unused imports and unused local variables are dropped here and
|
||||
owned by ruff (F401/F841), whose per-file scope analysis is strictly more
|
||||
accurate than vulture's global name-set matching (which hides an import that is
|
||||
dead in one file whenever the same name is used in any other file).
|
||||
|
||||
Class-body findings (fields, methods inside a class) are filtered out here
|
||||
and handled separately by checks/classes.py which understands Pydantic.
|
||||
"""
|
||||
@@ -13,14 +19,22 @@ import subprocess
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
|
||||
from . import is_excepted, is_lintignored
|
||||
from . import CheckError, is_excepted, is_lintignored
|
||||
|
||||
CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"
|
||||
|
||||
# Generous so the first run after a window reload (editor startup load) does not
|
||||
# time out and get reported as zero findings.
|
||||
_TIMEOUT = 120
|
||||
|
||||
@lru_cache(maxsize=64)
|
||||
def _class_line_ranges(filepath: str) -> list[tuple[int, int]]:
|
||||
"""Return (start, end) line ranges for all class bodies in *filepath*."""
|
||||
|
||||
@lru_cache(maxsize=256)
|
||||
def _class_line_ranges_cached(filepath: str, _mtime: float) -> list[tuple[int, int]]:
|
||||
"""Return (start, end) line ranges for all class bodies in *filepath*.
|
||||
|
||||
Keyed on *(filepath, mtime)* so the long-lived watch process re-parses a file
|
||||
after it is edited instead of returning stale ranges from an earlier version.
|
||||
"""
|
||||
try:
|
||||
tree = ast.parse(Path(filepath).read_text())
|
||||
except (OSError, SyntaxError):
|
||||
@@ -33,6 +47,14 @@ def _class_line_ranges(filepath: str) -> list[tuple[int, int]]:
|
||||
return ranges
|
||||
|
||||
|
||||
def _class_line_ranges(filepath: str) -> list[tuple[int, int]]:
|
||||
try:
|
||||
mtime = Path(filepath).stat().st_mtime
|
||||
except OSError:
|
||||
return []
|
||||
return _class_line_ranges_cached(filepath, mtime)
|
||||
|
||||
|
||||
def _is_inside_class(filepath: str, lineno: int) -> bool:
|
||||
"""True when *lineno* is strictly inside a class body.
|
||||
|
||||
@@ -52,7 +74,7 @@ def run_vulture(
|
||||
if not vulture_bin.exists():
|
||||
found = shutil.which("vulture")
|
||||
if not found:
|
||||
return []
|
||||
raise CheckError("vulture executable not found in backend/.venv/bin or PATH")
|
||||
vulture_bin = Path(found)
|
||||
|
||||
whitelist = CONFIG_DIR / "vulture_whitelist.py"
|
||||
@@ -71,10 +93,19 @@ def run_vulture(
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True, cwd=str(root), timeout=30,
|
||||
cmd, capture_output=True, text=True, cwd=str(root), timeout=_TIMEOUT,
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return []
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise CheckError(f"timed out after {_TIMEOUT}s (machine under load or large tree)") from e
|
||||
except OSError as e:
|
||||
raise CheckError(f"failed to launch vulture ({e})") from e
|
||||
|
||||
# vulture exits 0 (clean) or 1 (findings) on success; a higher code with no
|
||||
# findings on stdout means vulture errored — surface it rather than reporting
|
||||
# an empty result as "clean".
|
||||
if result.returncode not in (0, 1) and not result.stdout.strip():
|
||||
detail = result.stderr.strip()[:300] or "no output"
|
||||
raise CheckError(f"vulture exited with code {result.returncode}: {detail}")
|
||||
|
||||
errors: list[str] = []
|
||||
for line in result.stdout.strip().splitlines():
|
||||
@@ -82,6 +113,11 @@ def run_vulture(
|
||||
if not m:
|
||||
continue
|
||||
filepath, lineno, message = m.groups()
|
||||
# Imports and local variables are ruff's domain (F401/F841); skip them
|
||||
# here so we don't double-report and so vulture's weaker import logic
|
||||
# never gets a vote.
|
||||
if re.search(r"unused (import|variable)", message):
|
||||
continue
|
||||
if is_excepted(filepath, "vulture", exceptions):
|
||||
continue
|
||||
if ignores and is_lintignored(root / filepath, root, "vulture", ignores):
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
"import-cycles": false,
|
||||
"no-nested-imports": false,
|
||||
"vulture": true,
|
||||
"ruff": true,
|
||||
"eslint": false,
|
||||
"knip": false,
|
||||
"endpoints": false,
|
||||
@@ -14,6 +15,7 @@
|
||||
"no-nested-imports": "Off on purpose: this codebase uses function-level/lazy imports to break import cycles (400+ sites). Flagging them all is wrong for us.",
|
||||
"eslint-knip": "Node tooling deferred to a later pass.",
|
||||
"classes": "Placeholder check, not wired up. endpoints: orphaned-endpoint triage deferred.",
|
||||
"ruff-vulture-split": "ruff owns per-file/per-scope checks (F401 unused imports, F811 redefinitions, F841 unused locals, ARG001/ARG002 unused args), which it does AST-accurately and which vulture either gets wrong (global name-set hides per-file dead imports) or rates as noisy 60% findings. vulture is narrowed in checks/vulture.py to whole-program reachability only (dead functions/methods/classes/attributes), the one thing ruff structurally cannot do since it never builds a cross-module symbol graph. F401 honors __all__ and the redundant-alias form (import x as x) for intentional re-exports; add one of those if F401 flags a deliberate re-export.",
|
||||
"max-file-lines-exceptions": "Grandfather list of pre-existing >300-line files (existing debt, not new). Paths updated after the folder-tree restructure moved several of them.",
|
||||
"max-folder-items-exceptions": "Exact-path allow for folders intentionally over the cap. The rule trips at >7 (7 items is fine, the 8th tips it), so only genuinely 8+ folders are listed. backend/ and backend/apps are FastAPI feature-package registries (each child is an app mounted in main.py); agents/ aggregates agent subsystems; core/, tools_lib/, tests/ are conventionally flat. Frontend: app/pages is the page registry, AgentChat/ChatInput/Settings-sections/Onboarding are organizational parents, and shared/state (Redux slices) plus hooks/steps/mcp-cards/Views are flat peer collections. scripts/, electron/, linter/checks/ are flat tool dirs. These replaced blanket .lintignore-max-folder-items sentinels (backend, frontend, scripts, electron, linter/checks) so the rule still catches NEW unplanned bloat everywhere else. Kept as whole-subtree sentinels on purpose: debugger/ (self-contained injected sub-tool with its own Vite GUI), webapp_template (Vite scaffold payload), and vendored mcp-bundles.",
|
||||
"import-cycles": "Flags RUNTIME circular imports only (SCC>1). Skips type-only imports (import type / export type) and dynamic import() since neither runs at module init, which is why the idiomatic Redux store<->hooks type cycle is not flagged. Frontend alias resolution comes from import-cycle-aliases. Zero cycles today; the check keeps it that way."
|
||||
@@ -24,6 +26,7 @@
|
||||
"import-cycle-aliases": { "@/": "frontend/src/" },
|
||||
"vulture-min-confidence": 1,
|
||||
"vulture-error-threshold": 1,
|
||||
"ruff-select": "F401,F811,F841,ARG001,ARG002",
|
||||
"no-nested-imports": true,
|
||||
"endpoint-ignore-routes": ["*/callback", "*/callback/*"]
|
||||
},
|
||||
@@ -59,6 +62,7 @@
|
||||
"no-nested-imports": [],
|
||||
"import-cycles": [],
|
||||
"vulture": [],
|
||||
"ruff": [],
|
||||
"endpoints": [],
|
||||
"classes": []
|
||||
}
|
||||
|
||||
+110
-49
@@ -7,12 +7,14 @@ import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Any, Callable
|
||||
|
||||
from checks import is_excluded, is_excepted, is_lintignored, collect_lintignores
|
||||
from checks import CheckError, is_excluded, is_excepted, is_lintignored, collect_lintignores
|
||||
from checks.structural import check_file_lines, check_folder_items, check_nested_imports
|
||||
from checks.vulture import run_vulture
|
||||
from checks.ruff import run_ruff
|
||||
from checks.eslint import run_eslint
|
||||
from checks.knip import run_knip
|
||||
from checks.endpoints import run_endpoint_check
|
||||
@@ -23,21 +25,44 @@ from watchfiles import watch, DefaultFilter
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
CONFIG_FILE = SCRIPT_DIR / "config" / "config.json"
|
||||
|
||||
# Print order for the sections; also the order they run in.
|
||||
SECTION_ORDER = [
|
||||
"structural", "vulture", "ruff", "eslint",
|
||||
"knip", "endpoints", "classes", "import-cycles",
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class LintResult:
|
||||
"""Outcome of one full lint pass.
|
||||
|
||||
``sections`` maps each section name to its (sorted) error lines. ``incomplete``
|
||||
maps a section name to the reason it could not run; such a section is *not*
|
||||
the same as a clean one, so callers must surface it rather than trusting its
|
||||
empty error list as a zero count.
|
||||
"""
|
||||
|
||||
sections: dict[str, list[str]] = field(default_factory=dict)
|
||||
incomplete: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
def has_findings(self) -> bool:
|
||||
return any(self.sections.values())
|
||||
|
||||
|
||||
def load_config() -> dict[str, Any]:
|
||||
with open(CONFIG_FILE) as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str], list[str], list[str], list[str]]:
|
||||
config = load_config()
|
||||
enabled: dict[str, bool] = config.get("enabled", {})
|
||||
rules: dict[str, int] = config["rules"]
|
||||
excludes: list[str] = config["exclude"]
|
||||
exceptions: dict[str, list[str]] = config["exceptions"]
|
||||
extensions: list[str] = config["include_extensions"]
|
||||
ignores = collect_lintignores(root, excludes)
|
||||
|
||||
def _structural_checks(
|
||||
root: Path,
|
||||
rules: dict[str, Any],
|
||||
enabled: dict[str, bool],
|
||||
excludes: list[str],
|
||||
exceptions: dict[str, list[str]],
|
||||
extensions: list[str],
|
||||
ignores: dict[Path, set[str]],
|
||||
) -> list[str]:
|
||||
max_lines: int = rules["max-file-lines"]
|
||||
max_items: int = rules["max-folder-items"]
|
||||
check_imports: bool = rules.get("no-nested-imports", False)
|
||||
@@ -88,53 +113,87 @@ def run_checks(root: Path) -> tuple[list[str], list[str], list[str], list[str],
|
||||
):
|
||||
structural_errors.extend(check_nested_imports(fp, root))
|
||||
|
||||
vulture_errors: list[str] = []
|
||||
if enabled.get("vulture", True):
|
||||
vulture_confidence = rules.get("vulture-min-confidence")
|
||||
if vulture_confidence is not None:
|
||||
vulture_error_threshold = rules.get("vulture-error-threshold", 100)
|
||||
vulture_errors = run_vulture(
|
||||
root, vulture_confidence, vulture_error_threshold, exceptions, ignores,
|
||||
)
|
||||
|
||||
eslint_errors = run_eslint(root, ignores) if enabled.get("eslint", True) else []
|
||||
knip_errors = run_knip(root, ignores) if enabled.get("knip", True) else []
|
||||
endpoint_ignore_routes: list[str] = rules.get("endpoint-ignore-routes", [])
|
||||
endpoint_errors = run_endpoint_check(root, exceptions, endpoint_ignore_routes, ignores) if enabled.get("endpoints", True) else []
|
||||
class_errors = run_class_check(root, exceptions, excludes, ignores) if enabled.get("classes", True) else []
|
||||
aliases: dict[str, str] = rules.get("import-cycle-aliases", {})
|
||||
cycle_errors = run_cycle_check(root, excludes, aliases, exceptions, ignores) if enabled.get("import-cycles", True) else []
|
||||
|
||||
return sorted(structural_errors), sorted(vulture_errors), sorted(eslint_errors), sorted(knip_errors), sorted(endpoint_errors), sorted(class_errors), sorted(cycle_errors)
|
||||
return structural_errors
|
||||
|
||||
|
||||
def _print_section(name: str, errors: list[str]) -> None:
|
||||
def run_checks(root: Path) -> LintResult:
|
||||
config = load_config()
|
||||
enabled: dict[str, bool] = config.get("enabled", {})
|
||||
rules: dict[str, Any] = config["rules"]
|
||||
excludes: list[str] = config["exclude"]
|
||||
exceptions: dict[str, list[str]] = config["exceptions"]
|
||||
extensions: list[str] = config["include_extensions"]
|
||||
ignores = collect_lintignores(root, excludes)
|
||||
|
||||
result = LintResult()
|
||||
|
||||
def run_section(name: str, fn: Callable[[], list[str]]) -> None:
|
||||
"""Run one section, recording a failure as *incomplete* rather than
|
||||
letting an empty list masquerade as a clean result."""
|
||||
try:
|
||||
result.sections[name] = sorted(fn())
|
||||
except CheckError as e:
|
||||
result.sections[name] = []
|
||||
result.incomplete[name] = e.reason
|
||||
|
||||
def _vulture() -> list[str]:
|
||||
if not enabled.get("vulture", True):
|
||||
return []
|
||||
confidence = rules.get("vulture-min-confidence")
|
||||
if confidence is None:
|
||||
return []
|
||||
threshold = rules.get("vulture-error-threshold", 100)
|
||||
return run_vulture(root, confidence, threshold, exceptions, ignores)
|
||||
|
||||
run_section("structural", lambda: _structural_checks(
|
||||
root, rules, enabled, excludes, exceptions, extensions, ignores,
|
||||
))
|
||||
run_section("vulture", _vulture)
|
||||
run_section("ruff", lambda: run_ruff(
|
||||
root, rules.get("ruff-select", "F401,F811,F841,ARG001,ARG002"), exceptions, ignores,
|
||||
) if enabled.get("ruff", True) else [])
|
||||
run_section("eslint", lambda: run_eslint(root, ignores) if enabled.get("eslint", True) else [])
|
||||
run_section("knip", lambda: run_knip(root, ignores) if enabled.get("knip", True) else [])
|
||||
run_section("endpoints", lambda: run_endpoint_check(
|
||||
root, exceptions, rules.get("endpoint-ignore-routes", []), ignores,
|
||||
) if enabled.get("endpoints", True) else [])
|
||||
run_section("classes", lambda: run_class_check(
|
||||
root, exceptions, excludes, ignores,
|
||||
) if enabled.get("classes", True) else [])
|
||||
run_section("import-cycles", lambda: run_cycle_check(
|
||||
root, excludes, rules.get("import-cycle-aliases", {}), exceptions, ignores,
|
||||
) if enabled.get("import-cycles", True) else [])
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _print_section(name: str, errors: list[str], reason: str | None) -> None:
|
||||
print(f"{name}: checking...", flush=True)
|
||||
for e in errors:
|
||||
print(e, flush=True)
|
||||
print(f"{name}: done. {len(errors)} error(s) found.", flush=True)
|
||||
if reason is not None:
|
||||
# Emit a problemMatcher-catchable line so the IDE Problems panel shows the
|
||||
# run was partial, then a human-readable status line for the terminal.
|
||||
print(
|
||||
f"linter/lint.py:1:1: error: [linter] '{name}' check INCOMPLETE: "
|
||||
f"{reason} — error counts are unreliable until this is resolved",
|
||||
flush=True,
|
||||
)
|
||||
print(f"{name}: done. INCOMPLETE — {reason}.", flush=True)
|
||||
else:
|
||||
print(f"{name}: done. {len(errors)} error(s) found.", flush=True)
|
||||
|
||||
|
||||
def print_results(
|
||||
structural_errors: list[str], vulture_errors: list[str],
|
||||
eslint_errors: list[str], knip_errors: list[str],
|
||||
endpoint_errors: list[str], class_errors: list[str],
|
||||
cycle_errors: list[str],
|
||||
) -> None:
|
||||
_print_section("structural", structural_errors)
|
||||
_print_section("vulture", vulture_errors)
|
||||
_print_section("eslint", eslint_errors)
|
||||
_print_section("knip", knip_errors)
|
||||
_print_section("endpoints", endpoint_errors)
|
||||
_print_section("classes", class_errors)
|
||||
_print_section("import-cycles", cycle_errors)
|
||||
def print_results(result: LintResult) -> None:
|
||||
for name in SECTION_ORDER:
|
||||
_print_section(name, result.sections.get(name, []), result.incomplete.get(name))
|
||||
|
||||
|
||||
def watch_loop(root: Path) -> None:
|
||||
|
||||
config_dir = SCRIPT_DIR / "config"
|
||||
|
||||
print_results(*run_checks(root))
|
||||
print_results(run_checks(root))
|
||||
|
||||
class SourceFilter(DefaultFilter):
|
||||
allowed_extensions = (".py", ".ts", ".tsx", ".js", ".jsx")
|
||||
@@ -152,7 +211,7 @@ def watch_loop(root: Path) -> None:
|
||||
return Path(path).is_dir()
|
||||
|
||||
for _changes in watch(root, watch_filter=SourceFilter()):
|
||||
print_results(*run_checks(root))
|
||||
print_results(run_checks(root))
|
||||
|
||||
|
||||
def main() -> None:
|
||||
@@ -166,9 +225,11 @@ def main() -> None:
|
||||
if args.watch:
|
||||
watch_loop(root)
|
||||
else:
|
||||
results = run_checks(root)
|
||||
print_results(*results)
|
||||
sys.exit(1 if any(results) else 0)
|
||||
result = run_checks(root)
|
||||
print_results(result)
|
||||
# Exit non-zero on findings OR on an incomplete run, so a partial pass is
|
||||
# never mistaken for a clean one by CI or print_errors.sh.
|
||||
sys.exit(1 if result.has_findings() or result.incomplete else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+12
-1
@@ -14,13 +14,15 @@ LINT_OUTPUT=$(python3 "$SCRIPT_DIR/lint.py" --root "$ROOT_DIR" 2>&1)
|
||||
LINT_EXIT=$?
|
||||
|
||||
if [ $LINT_EXIT -ne 0 ]; then
|
||||
STRUCT_LINES=$(echo "$LINT_OUTPUT" | grep -v "^structural:" | grep -v "^vulture:" | grep -v "^eslint:" | grep -v "^knip:" | grep -v '\[vulture\]' | grep -v '\[eslint\]' | grep -v '\[knip\]')
|
||||
STRUCT_LINES=$(echo "$LINT_OUTPUT" | grep -v "^structural:" | grep -v "^vulture:" | grep -v "^ruff:" | grep -v "^eslint:" | grep -v "^knip:" | grep -v '\[vulture\]' | grep -v '\[ruff\]' | grep -v '\[eslint\]' | grep -v '\[knip\]')
|
||||
VULTURE_LINES=$(echo "$LINT_OUTPUT" | grep '\[vulture\]')
|
||||
RUFF_LINES=$(echo "$LINT_OUTPUT" | grep '\[ruff\]')
|
||||
ESLINT_LINES=$(echo "$LINT_OUTPUT" | grep '\[eslint\]')
|
||||
KNIP_LINES=$(echo "$LINT_OUTPUT" | grep '\[knip\]')
|
||||
|
||||
STRUCT_COUNT=$(echo "$STRUCT_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
VULTURE_COUNT=$(echo "$VULTURE_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
RUFF_COUNT=$(echo "$RUFF_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
ESLINT_COUNT=$(echo "$ESLINT_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
KNIP_COUNT=$(echo "$KNIP_LINES" | grep -cE ':\s+(error|warning):\s+')
|
||||
|
||||
@@ -42,6 +44,15 @@ if [ $LINT_EXIT -ne 0 ]; then
|
||||
echo -e "${CYAN}${BOLD} ${VULTURE_COUNT} finding(s) — fix or add to linter/config/vulture_whitelist.py${RESET}"
|
||||
fi
|
||||
|
||||
if [ "$RUFF_COUNT" -gt 0 ]; then
|
||||
echo ""
|
||||
echo -e "${CYAN}${BOLD}[ruff] Lint errors found:${RESET}"
|
||||
echo "$RUFF_LINES" | while IFS= read -r line; do
|
||||
[ -n "$line" ] && echo -e "${CYAN} $line${RESET}"
|
||||
done
|
||||
echo -e "${CYAN}${BOLD} ${RUFF_COUNT} finding(s) — fix, add a noqa, or except in linter/config/config.json${RESET}"
|
||||
fi
|
||||
|
||||
if [ "$ESLINT_COUNT" -gt 0 ]; then
|
||||
echo ""
|
||||
echo -e "${YELLOW}${BOLD}[eslint] Lint errors found:${RESET}"
|
||||
|
||||
Reference in New Issue
Block a user