94 verbose: bool =
False,
95 exclude_dirs: Optional[set] =
None,
96 wildcard_exclude: Optional[list] =
None,
97 active_rules: Optional[dict] =
None,
100 Run linter checks on all Python files and aggregate pass/fail statistics.
102 Returns a dict with total_files, fully_compliant count, per-check pass
103 counts, and compliance percentage.
105 func =
"check_lint_health"
107 if exclude_dirs
is None:
108 exclude_dirs = set(LINTER_EXCLUDE_DIRS)
109 if wildcard_exclude
is None:
110 wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
111 if active_rules
is None:
112 active_rules = dict(_DEFAULT_RULES)
114 python_files = list(find_python_files(project_root, exclude_dirs, wildcard_exclude))
118 "header",
"docstring",
"dependencies_section",
"diagnostics_profile",
119 "dbg_usage",
"dbg_import",
"func_pattern",
"syntax_warnings",
121 active_checks = [c
for c
in health_checks
if active_rules.get(c,
True)]
123 total = len(python_files)
124 counters = {c: 0
for c
in active_checks}
129 "header":
lambda text, lines, py_file, tree, sw: validate_header_block(
130 lines[:4]
if len(lines) >= 4
else lines, py_file, ctx),
131 "docstring":
lambda text, lines, py_file, tree, sw: check_docstring(text),
132 "dependencies_section":
lambda text, lines, py_file, tree, sw: check_dependencies_section(text),
133 "diagnostics_profile":
lambda text, lines, py_file, tree, sw: check_diagnostics_profile(text),
134 "dbg_usage":
lambda text, lines, py_file, tree, sw: check_dbg_usage(text, tree=tree),
135 "dbg_import":
lambda text, lines, py_file, tree, sw: check_dbg_import(text, tree=tree),
136 "func_pattern":
lambda text, lines, py_file, tree, sw: check_func_pattern(text, tree=tree),
137 "syntax_warnings":
lambda text, lines, py_file, tree, sw: check_syntax_warnings(text, warnings_list=sw),
140 for py_file
in python_files:
142 text = py_file.read_text(encoding=
"utf-8")
146 lines = text.split(
'\n')
147 tree, syntax_warnings_list = safe_parse(text)
150 for check_name
in active_checks:
151 ok, _ = check_funcs[check_name](text, lines, py_file, tree, syntax_warnings_list)
152 checks[check_name] = ok
154 counters[check_name] += 1
156 if all(checks.values()):
160 rel_path = to_project_relative(py_file, project_root)
161 file_details.append({
164 "compliant": all(checks.values()),
167 compliance_pct = round((fully_compliant / total * 100), 1)
if total > 0
else 0.0
170 "total_files": total,
171 "fully_compliant": fully_compliant,
172 "compliance_pct": compliance_pct,
176 for check_name, count
in counters.items():
177 pct = round((count / total * 100), 1)
if total > 0
else 0.0
178 result[
"checks"][check_name] = {
185 result[
"files"] = file_details
193 exclude_dirs: Optional[set] =
None,
194 wildcard_exclude: Optional[list] =
None,
195 active_rules: Optional[dict] =
None,
196 verbose: bool =
False,
199 Count violations that oct format could fix automatically.
201 Runs formatter checks in dry-run mode (no modifications).
202 Only counts fixes for rules that are active.
204 When *verbose* is True the result includes a ``"files"`` key mapping
205 to a list of ``{"path": <rel_path>, "fixes": [<fix_name>, ...]}``
206 entries so the caller can show which files have fixable violations.
208 func =
"check_fixable_health"
218 if active_rules
is None:
219 active_rules = dict(_DEFAULT_RULES)
221 fmt_ctx = FormatterContext(
222 project_root=project_root,
223 project_name=project_root.name,
224 diagnostics_dir=ctx.diagnostics_dir,
225 tests_dir=ctx.tests_dir,
226 docs_dir=ctx.docs_dir,
233 if exclude_dirs
is None:
234 exclude_dirs = set(LINTER_EXCLUDE_DIRS)
235 if wildcard_exclude
is None:
236 wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
237 python_files = list(find_python_files(project_root, exclude_dirs, wildcard_exclude))
242 "docstring":
"docstring",
243 "dbg_import":
"dbg_import",
244 "func_pattern":
"func_pattern",
246 fixable = {k: 0
for k
in fix_rule_map
if active_rules.get(fix_rule_map[k],
True)}
249 "header": fix_header_block,
250 "docstring": fix_docstring,
251 "dbg_import": fix_dbg_import,
252 "func_pattern": fix_func_pattern,
255 file_details: list[dict] = []
257 for py_file
in python_files:
259 text = py_file.read_text(encoding=
"utf-8")
263 file_fixes: list[str] = []
264 for fix_name
in fixable:
265 changed, _, _ = fix_funcs[fix_name](py_file, text, fmt_ctx)
267 fixable[fix_name] += 1
268 file_fixes.append(fix_name)
270 if verbose
and file_fixes:
271 rel_path = to_project_relative(py_file, project_root)
272 file_details.append({
"path": rel_path,
"fixes": file_fixes})
274 fixable[
"total"] = sum(fixable.values())
276 fixable[
"files"] = file_details
337 coverage_threshold: Optional[float] =
None,
338 production_mode: bool =
False,
339 coverage_target: Optional[str] =
None,
342 Run project tests and capture pass/fail status.
344 Uses pytest subprocess to run tests and captures the exit code
350 Path to the project root containing a ``tests/`` directory.
352 Maximum seconds to wait for pytest to complete. Default 300 (OI-413).
354 FS-C3 / Drift item C3 — when set, run pytest with ``--cov`` and
355 compare the resulting TOTAL coverage percentage against this
356 threshold. ``None`` (default) skips the coverage measurement
357 entirely so existing CI keeps its prior behaviour.
359 When ``True`` together with a ``coverage_threshold``, a coverage
360 result below the threshold flips ``status`` to ``"fail"``.
361 Outside production mode, below-threshold coverage is reported
362 but never blocks the dashboard.
364 Module / package name to measure (passed as ``--cov=<target>``).
365 Defaults to the project root directory name when ``None``.
367 func =
"check_test_health"
369 import importlib.util
371 tests_dir = project_root /
"tests"
372 if not tests_dir.is_dir():
376 "summary":
"No tests/ directory found",
378 "coverage_pct":
None,
379 "coverage_threshold": coverage_threshold,
380 "coverage_blocking": production_mode,
384 if importlib.util.find_spec(
"pytest")
is None:
388 "summary":
"pytest not installed — run: pip install pytest",
390 "coverage_pct":
None,
391 "coverage_threshold": coverage_threshold,
392 "coverage_blocking": production_mode,
397 coverage_requested = coverage_threshold
is not None
398 coverage_available = (
400 and importlib.util.find_spec(
"pytest_cov")
is not None
404 sys.executable,
"-m",
"pytest", str(tests_dir),
405 "-n",
"auto",
"--dist=worksteal",
"--tb=line",
"--no-header",
407 if coverage_available:
408 target = coverage_target
or project_root.name
409 pytest_args.extend([f
"--cov={target}",
"--cov-report=term"])
412 result = subprocess.run(
419 cwd=str(project_root),
425 _SUMMARY_RE = re.compile(
r"\b\d+ \w+ in [\d.]+s")
426 _DECORATION_RE = re.compile(
r"^=+\s*|\s*=+$")
428 for line
in reversed(result.stdout.strip().split(
'\n')):
430 if "passed" in line
or "failed" in line
or "error" in line
or _SUMMARY_RE.search(line):
431 summary = _DECORATION_RE.sub(
"", line).strip()
439 if result.returncode != 0:
440 for line
in result.stdout.strip().split(
'\n'):
442 if not line
or line.startswith(
"=")
or line.startswith(
"."):
444 if (
"Error" in line
or "Exception" in line
445 or line.startswith(
"FAILED ")
446 or line.startswith(
"ERROR ")):
447 detail_lines.append(line)
448 if len(detail_lines) >= 5:
452 if not summary
and result.stderr.strip():
453 err_lines = result.stderr.strip().split(
'\n')
454 summary = err_lines[-1].strip()
458 summary = f
"pytest exited with code {result.returncode}"
460 status =
"pass" if result.returncode == 0
else "fail"
463 coverage_pct: Optional[float] =
None
464 if coverage_available:
465 cov_match = _COVERAGE_TOTAL_RE.search(result.stdout)
468 coverage_pct = float(cov_match.group(1))
477 and coverage_pct
is not None
478 and coverage_threshold
is not None
479 and coverage_pct < coverage_threshold
484 f
"{summary} | coverage {coverage_pct:.1f}% below "
485 f
"production threshold {coverage_threshold:.1f}%"
490 coverage_msg: Optional[str] =
None
491 if coverage_requested
and not coverage_available:
493 "pytest-cov not installed — install with: "
494 "pip install -e .[test]"
499 "exit_code": result.returncode,
501 "details": detail_lines,
502 "coverage_pct": coverage_pct,
503 "coverage_threshold": coverage_threshold,
504 "coverage_blocking": production_mode,
505 "coverage_message": coverage_msg,
508 except subprocess.TimeoutExpired:
512 "summary": f
"Tests timed out after {timeout} seconds",
514 "coverage_pct":
None,
515 "coverage_threshold": coverage_threshold,
516 "coverage_blocking": production_mode,
518 except Exception
as e:
524 "coverage_pct":
None,
525 "coverage_threshold": coverage_threshold,
526 "coverage_blocking": production_mode,
1088 json_mode: bool =
False,
1089 verbose: bool =
False,
1090 test_timeout: int |
None =
None,
1091 log_output: bool =
False,
1094 Run the project health dashboard.
1099 The detected Option C project root directory.
1101 Output machine-readable JSON instead of terminal report.
1103 Show per-file details in addition to summary.
1104 test_timeout : int | None
1105 Maximum seconds to wait for pytest. Overrides ``health.test_timeout``
1106 in ``.octrc.json``. Defaults to 300 when neither is set (OI-413).
1111 diagnostics_dir = project_root /
"oc_diagnostics"
1112 if not diagnostics_dir.is_dir():
1113 diagnostics_dir = project_root /
"diagnostics"
1114 if diagnostics_dir.is_dir():
1116 "Warning: 'diagnostics/' is deprecated and will be removed in v1.0.0. "
1117 "Rename to 'oc_diagnostics/'.",
1122 project_root=project_root,
1123 project_name=project_root.name,
1124 diagnostics_dir=diagnostics_dir,
1125 tests_dir=project_root /
"tests",
1126 docs_dir=project_root /
"docs",
1130 exclude_dirs = set(LINTER_EXCLUDE_DIRS)
1131 wildcard_exclude = list(WILDCARD_EXCLUDE_DIRS)
1132 excluded_by_config: List[str] = []
1133 active_rules = dict(_DEFAULT_RULES)
1134 disabled_rules: List[str] = []
1139 effective_test_timeout: int = 300
1140 from oct.core.octrc
import load_octrc_safe
1141 octrc = load_octrc_safe(project_root)
1142 linter_cfg = octrc.get(
"linter", {})
1143 if isinstance(linter_cfg, dict):
1144 for d
in linter_cfg.get(
"exclude_dirs_add", []):
1146 excluded_by_config.append(d)
1147 for d
in linter_cfg.get(
"exclude_dirs_remove", []):
1148 exclude_dirs.discard(d)
1149 for d
in linter_cfg.get(
"wildcard_exclude_add", []):
1150 if d
not in wildcard_exclude:
1151 wildcard_exclude.append(d)
1153 profile_name = linter_cfg.get(
"profile",
"default")
1154 active_rules = dict(_PROFILES.get(profile_name, _DEFAULT_RULES))
1155 rule_overrides = linter_cfg.get(
"rules", {})
1156 if isinstance(rule_overrides, dict):
1157 active_rules.update(rule_overrides)
1158 disabled_rules = [r
for r, v
in active_rules.items()
if not v]
1159 health_cfg = octrc.get(
"health", {})
1160 if isinstance(health_cfg, dict):
1161 cfg_timeout = health_cfg.get(
"test_timeout")
1162 if isinstance(cfg_timeout, int)
and cfg_timeout > 0:
1163 effective_test_timeout = cfg_timeout
1166 if test_timeout
is not None:
1167 effective_test_timeout = test_timeout
1170 active_checks = [c
for c
in _HEALTH_CHECKS
if active_rules.get(c,
True)]
1172 "header":
"header",
"docstring":
"docstring",
1173 "dbg_import":
"dbg_import",
"func_pattern":
"func_pattern",
1175 fixable_names = [k
for k
in _fix_rule_map
if active_rules.get(_fix_rule_map[k],
True)]
1177 from oct.core.terminal
import supports_ansi_on
1178 use_color = supports_ansi_on(sys.stdout)
and not json_mode
1179 G =
"\033[92m" if use_color
else ""
1180 R =
"\033[91m" if use_color
else ""
1181 Y =
"\033[93m" if use_color
else ""
1182 X =
"\033[0m" if use_color
else ""
1185 project_name=project_root.name,
1186 excluded_dirs=excluded_by_config,
1187 disabled_rules=disabled_rules,
1188 active_checks=active_checks,
1189 fixable_names=fixable_names,
1190 use_color=use_color,
1193 dashboard = Dashboard(quiet=json_mode)
1194 dashboard.render(template_lines, footer_line=lm.footer, tasks_total=7)
1197 dashboard.set_task(
"Lint compliance")
1198 dashboard.update_line(lm.lint, f
"{Y}[>]{X} Lint Compliance: running...")
1200 project_root, ctx, verbose=verbose,
1201 exclude_dirs=exclude_dirs, wildcard_exclude=wildcard_exclude,
1202 active_rules=active_rules,
1204 pct = lint_result[
"compliance_pct"]
1205 pc = G
if pct == 100.0
else R
1206 dashboard.update_line(
1208 f
"{G}[x]{X} Lint Compliance: "
1209 f
"{pc}{pct}%{X} ({lint_result['fully_compliant']}/{lint_result['total_files']}"
1210 f
" files fully compliant)",
1212 for ck, idx
in lm.lint_checks.items():
1213 cd = lint_result[
"checks"].get(ck)
1214 label = ck.replace(
"_",
" ").title()
1216 c = G
if cd[
"pass"] == cd[
"total"]
else R
1217 dashboard.update_line(
1218 idx, f
" {G}[x]{X} {label + ':':<20}{c}{_pct_str(cd['pass'], cd['total'])}{X}",
1221 dashboard.update_line(idx, f
" {Y}[-]{X} {label + ':':<20}disabled")
1222 dashboard.complete_task()
1225 dashboard.set_task(
"Auto-fixable checks")
1226 dashboard.update_line(lm.fixable, f
"{Y}[>]{X} Fixable Violations: running...")
1229 exclude_dirs=exclude_dirs, wildcard_exclude=wildcard_exclude,
1230 active_rules=active_rules,
1233 fix_total = fixable_result.get(
"total", 0)
1234 fc = G
if fix_total == 0
else Y
1235 dashboard.update_line(lm.fixable, f
"{G}[x]{X} Fixable Violations: {fc}{fix_total} total{X}")
1236 for fn, idx
in lm.fixable_items.items():
1237 val = fixable_result.get(fn, 0)
1238 vc = G
if val == 0
else Y
1239 dashboard.update_line(
1240 idx, f
" {G}[x]{X} {_FIX_LABELS.get(fn, fn) + ':':<20}{vc}{val}{X}",
1242 dashboard.complete_task()
1245 dashboard.set_task(
"Documentation")
1246 dashboard.update_line(lm.docs, f
"{Y}[>]{X} Documentation: running...")
1247 _octrc_cfg = load_octrc_safe(project_root)
1248 _production_mode = bool(
1249 (_octrc_cfg.get(
"health")
or {}).get(
"production_mode",
False)
1252 dc = G
if docs_result[
"total_present"] == docs_result[
"total_required"]
else R
1253 dashboard.update_line(
1255 f
"{G}[x]{X} Documentation: "
1256 f
"{dc}{docs_result['total_present']}/{docs_result['total_required']} present{X}",
1258 for fi
in docs_result[
"files"]:
1259 idx = lm.docs_files.get(fi[
"name"])
1263 dashboard.update_line(idx, f
" {G}[x]{X} {fi['name']:<20}{G}[OK]{X}")
1264 elif fi.get(
"required",
True):
1265 dashboard.update_line(idx, f
" {R}[!]{X} {fi['name']:<20}{R}[MISSING]{X}")
1267 dashboard.update_line(idx, f
" {Y}[-]{X} {fi['name']:<20}{Y}[OPTIONAL]{X}")
1268 dashboard.complete_task()
1271 dashboard.set_task(
"Test suite")
1272 dashboard.update_line(lm.tests, f
"{Y}[>]{X} Tests: running...")
1274 _health_cfg = _octrc_cfg.get(
"health")
or {}
1275 _coverage_threshold_raw = _health_cfg.get(
"coverage_threshold")
1276 _coverage_threshold: Optional[float] =
None
1277 if isinstance(_coverage_threshold_raw, (int, float)):
1278 _coverage_threshold = float(_coverage_threshold_raw)
1281 timeout=effective_test_timeout,
1282 coverage_threshold=_coverage_threshold,
1283 production_mode=_production_mode,
1285 ts = test_result[
"status"]
1286 cov_pct = test_result.get(
"coverage_pct")
1287 cov_thr = test_result.get(
"coverage_threshold")
1288 cov_msg = test_result.get(
"coverage_message")
1290 if cov_pct
is not None and cov_thr
is not None:
1291 cov_state =
"OK" if cov_pct >= cov_thr
else "below"
1292 cov_suffix = f
" | cov {cov_pct:.1f}%/{cov_thr:.0f}% [{cov_state}]"
1294 cov_suffix = f
" | {cov_msg}"
1295 tsm = f
" ({test_result['summary']}{cov_suffix})" if test_result[
"summary"]
else cov_suffix
1297 dashboard.update_line(lm.tests, f
"{G}[x]{X} Tests: {G}PASS{X}{tsm}")
1298 elif ts ==
"not_run":
1299 dashboard.update_line(lm.tests, f
"{Y}[-]{X} Tests: {Y}NOT RUN{X}{tsm}")
1301 dashboard.update_line(lm.tests, f
"{R}[!]{X} Tests: {R}{ts.upper()}{X}{tsm}")
1302 dashboard.complete_task()
1305 dashboard.set_task(
"Compatibility check")
1306 dashboard.update_line(lm.compat, f
"{Y}[>]{X} oc_diagnostics: checking...")
1308 if compat_result[
"installed"]:
1309 if compat_result[
"compatible"]:
1310 dashboard.update_line(
1312 f
"{G}[x]{X} oc_diagnostics: v{compat_result['version']} -- {G}compatible{X}",
1315 dashboard.update_line(
1317 f
"{R}[!]{X} oc_diagnostics: v{compat_result['version']} -- {R}INCOMPATIBLE{X}",
1320 dashboard.update_line(lm.compat, f
"{R}[!]{X} oc_diagnostics: {R}NOT INSTALLED{X}")
1321 dashboard.complete_task()
1324 dashboard.set_task(
"Virtual environment")
1325 dashboard.update_line(lm.venv, f
"{Y}[>]{X} Virtual Env: checking...")
1327 if venv_result[
"status"] ==
"OK":
1328 dashboard.update_line(
1329 lm.venv, f
"{G}[x]{X} Virtual Env: {G}[OK]{X} {venv_result['message']}",
1332 dashboard.update_line(
1333 lm.venv, f
"{R}[!]{X} Virtual Env: {Y}[WARN]{X} {venv_result['message']}",
1335 dashboard.complete_task()
1338 dashboard.set_task(
"Git integration")
1339 dashboard.update_line(lm.git, f
"{Y}[>]{X} Git Integration: checking...")
1341 _GC = {
"OK": G,
"WARN": Y,
"FAIL": R,
"SKIP": Y}
1344 for gk, gl
in _GIT_LABELS.items():
1345 idx = lm.git_checks.get(gk)
1346 check = git_result.get(gk)
1347 if idx
is None or check
is None:
1350 status = check[
"status"]
1351 color = _GC.get(status, X)
1355 elif status
in (
"WARN",
"SKIP"):
1359 dashboard.update_line(idx, f
" {mk} {gl + ':':<20}{color}[{status}]{X} {check['detail']}")
1360 gc = G
if git_ok == git_total
else Y
1361 dashboard.update_line(lm.git, f
"{G}[x]{X} Git Integration: {gc}{git_ok}/{git_total} checks OK{X}")
1362 dashboard.complete_task()
1368 "project": project_root.name,
1369 "oct_version": __version__,
1370 "timestamp": datetime.now().isoformat(),
1371 "excluded_dirs": excluded_by_config,
1372 "disabled_rules": disabled_rules,
1373 "lint": lint_result,
1374 "fixable": fixable_result,
1375 "docs": docs_result,
1376 "tests": test_result,
1377 "compat": compat_result,
1378 "venv": venv_result,
1386 from oct.core.log_writer
import write_command_log
1387 log_path = write_command_log(project_root,
"health", report)
1388 print(f
"Log written to: {log_path}", file=sys.stderr)
1390 if verbose
and not json_mode
and "files" in lint_result:
1391 print(f
"\n{Y}{'-' * 70}{X}")
1392 print(f
"{Y}Per-file breakdown:{X}")
1393 print(f
"{Y}{'-' * 70}{X}\n")
1394 for file_info
in lint_result[
"files"]:
1395 if not file_info[
"compliant"]:
1396 failed = [k
for k, v
in file_info[
"checks"].items()
if not v]
1397 print(f
" {file_info['path']}")
1398 print(f
" Failed: {R}{', '.join(failed)}{X}")
1400 if verbose
and not json_mode
and fixable_result.get(
"files"):
1401 print(f
"\n{Y}{'-' * 70}{X}")
1402 print(f
"{Y}Fixable violations per file:{X}")
1403 print(f
"{Y}{'-' * 70}{X}\n")
1404 for file_info
in fixable_result[
"files"]:
1405 print(f
" {file_info['path']}")
1406 print(f
" Fixable: {Y}{', '.join(file_info['fixes'])}{X}")