feat(scripts): add heuristics to audit_exception_handling for review pass patterns (10 new heuristics + tests)

2026-06-17 16:15:16 -04:00
parent 8b954ee180
commit f26091941c
2 changed files with 500 additions and 0 deletions
@@ -453,11 +453,205 @@ class ExceptionVisitor(ast.NodeVisitor):
    f"Compliant: stdlib I/O exception {exc_name} caught in our own code is acceptable (per convention, file/network errors are converted to ErrorInfo).",
   )

+  # 11-17. Heuristics added by result_migration_review_pass_20260617
+  # These cover the 7 most common compliant patterns the review pass found.
+  # Each heuristic inspects the try body + except body together.
+  compliant = self._try_compliant_pattern(try_node, handler, exc_name)
+  if compliant is not None:
+   return compliant
+
  return (
   "UNCLEAR",
   f"Manual review: catches {exc_name}; not obviously boundary or violation. Check whether the except site is converting to ErrorInfo (good) or hiding the error (bad).",
  )

+ def _has_call_with_attr(self, stmts: list[ast.stmt], attr_name: str) -> bool:
+  """True if any statement contains a call to `.attr_name(...)` (e.g. list.index, dict.get)."""
+  for s in stmts:
+   for node in ast.walk(s):
+    if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == attr_name:
+     return True
+  return False
+
+ def _has_keyword_true_call(self, stmts: list[ast.stmt], attr_name: str, kw_name: str) -> bool:
+  """True if any statement contains a call `.attr_name(..., kw_name=True)`."""
+  for s in stmts:
+   for node in ast.walk(s):
+    if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == attr_name:
+     for kw in node.keywords:
+      if kw.arg == kw_name and isinstance(kw.value, ast.Constant) and kw.value.value is True:
+       return True
+  return False
+
+ def _has_print_call(self, stmts: list[ast.stmt]) -> bool:
+  """True if any statement is an `Expr(Call(Name('print'), ...))`."""
+  for s in stmts:
+   if isinstance(s, ast.Expr) and isinstance(s.value, ast.Call):
+    f = s.value.func
+    if isinstance(f, ast.Name) and f.id == "print":
+     return True
+  return False
+
+ def _has_import_stmt(self, stmts: list[ast.stmt]) -> bool:
+  """True if any statement is an `Import` or `ImportFrom`."""
+  for s in stmts:
+   if isinstance(s, (ast.Import, ast.ImportFrom)):
+    return True
+  return False
+
+ def _try_compliant_pattern(self, try_node: ast.Try, handler: ast.ExceptHandler, exc_name: str) -> tuple[str, str] | None:
+  """Detect one of the 7 common compliant patterns found by the review pass.
+
+  Returns (category, hint) if the pattern is compliant, else None.
+  """
+  try_body = try_node.body
+  except_body = handler.body
+  exc_set = {e.strip() for e in exc_name.replace("(", "").replace(")", "").split(",") if e.strip()}
+
+  # 11. list.index(x) with ValueError fallback to default index
+  if exc_set & {"ValueError"} and self._has_call_with_attr(try_body, "index") and len(except_body) > 0:
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: list.index(x); except ({', '.join(sorted(exc_set))}): ...` is the canonical combo-box fallback pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 12. dict[x] or get_capabilities(...) with KeyError fallback to default
+  if exc_set == {"KeyError"} and len(except_body) > 0 and len(try_body) > 0:
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: <lookup>; except KeyError: ...` is the canonical lookup-miss-with-default pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 13. datetime.fromisoformat(s) with ValueError: None
+  if exc_set == {"ValueError"} and self._has_call_with_attr(try_body, "fromisoformat"):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: datetime.fromisoformat(s); except ValueError: ...` is the canonical lenient-deserialization pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 14. Path.resolve(strict=True) with (OSError, ValueError) fallback
+  if exc_set == {"OSError", "ValueError"} and self._has_keyword_true_call(try_body, "resolve", "strict"):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: Path(p).resolve(strict=True); except (OSError, ValueError): ...` is the canonical graceful-path-resolution pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 15. Path.relative_to with ValueError: pass / return False
+  if exc_set == {"ValueError"} and self._has_call_with_attr(try_body, "relative_to"):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: rp.relative_to(base); except ValueError: ...` is the canonical subpath-check pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 16. asyncio.get_running_loop() with RuntimeError: asyncio.run(...)
+  if exc_set == {"RuntimeError"} and self._has_call_with_attr(try_body, "get_running_loop") and self._has_call_with_attr(except_body, "run"):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: get_running_loop(); except RuntimeError: asyncio.run(...)` is the canonical sync/async bridge pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 17. import with (ImportError, ModuleNotFoundError, AttributeError) + fallback stub
+  if exc_set & {"ImportError", "ModuleNotFoundError", "AttributeError"} and self._has_import_stmt(try_body) and len(except_body) > 0:
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: import ...; except ({', '.join(sorted(exc_set))}): <stub>` is the canonical graceful-degradation pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 18. JSON parse with (json.JSONDecodeError, KeyError) and print() for CLI-style input
+  if "JSONDecodeError" in exc_name and self._has_call_with_attr(try_body, "loads") and self._has_print_call(except_body):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: json.loads(...); except json.JSONDecodeError: print(...)` is the canonical CLI-style JSON input parser pattern (per result_migration_review_pass_20260617).",
+   )
+  if exc_set == {"KeyError"} and self._has_call_with_attr(try_body, "loads") and self._has_print_call(except_body):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: json.loads(...); except KeyError: print(...)` is the canonical CLI-style JSON input parser pattern (per result_migration_review_pass_20260617).",
+   )
+
+  # 19. Narrow except + log (sys.stderr.write or logging.*) for defer-not-catch or retry-then-give-up
+  if len(except_body) > 0 and self._has_log_call(except_body) and not exc_set & {"Exception", "BaseException", ""}:
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: ...; except ({', '.join(sorted(exc_set))}): <log>` is the canonical catch+log pattern (defer-not-catch or retry-then-give-up) (per result_migration_review_pass_20260617).",
+   )
+
+  # 20. ImGui scope cleanup guard (narrow except + imgui.end_* call)
+  if exc_set & {"TypeError", "AttributeError", "RuntimeError"} and self._has_imgui_end_call(except_body):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: ...; except ({', '.join(sorted(exc_set))}): imgui.end_*()` is the canonical ImGui scope cleanup guard (per result_migration_review_pass_20260617).",
+   )
+
+  # 21. MCP tool boundary (broad except Exception + return string in str-returning function)
+  enclosing_func = self._current_func_node()
+  if enclosing_func is not None and enclosing_func.returns is not None and ast.unparse(enclosing_func.returns) == "str" and exc_set & {"Exception", "BaseException"} and self._has_string_return(except_body):
+   return (
+    "INTERNAL_COMPLIANT",
+    f"Compliant: `try: ...; except Exception: return <string>` in a `-> str` tool function is the canonical MCP tool boundary pattern (per result_migration_review_pass_20260617).",
+   )
+
+  return None
+
+ def _has_string_return(self, stmts: list[ast.stmt]) -> bool:
+  """True if any statement is a `return <f-string or string constant>`."""
+  for s in stmts:
+   if isinstance(s, ast.Return) and s.value is not None:
+    if isinstance(s.value, ast.Constant) and isinstance(s.value.value, str):
+     return True
+    if isinstance(s.value, ast.JoinedStr):
+     return True
+  return False
+
+ def _has_log_call(self, stmts: list[ast.stmt]) -> bool:
+  """True if any statement is a log call (sys.stderr.write, logging.*, print)."""
+  for s in stmts:
+   for node in ast.walk(s):
+    if isinstance(node, ast.Call):
+     f = node.func
+     if isinstance(f, ast.Attribute) and f.attr in ("write", "error", "warning", "info", "debug", "exception"):
+      return True
+     if isinstance(f, ast.Name) and f.id == "print":
+      return True
+  return False
+
+ def _has_imgui_end_call(self, stmts: list[ast.stmt]) -> bool:
+  """True if any statement is a call to an imgui.end_* function."""
+  for s in stmts:
+   for node in ast.walk(s):
+    if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr.startswith("end_"):
+     return True
+  return False
+
+ def _enclosing_if_is_none_guard(self) -> bool:
+  """True if the current raise is inside an `if <var> is None:` block (validation pattern)."""
+  # The _func_stack holds the function context; we don't track the if-stack.
+  # Walk the AST of the current function and check if the raise is inside
+  # an `if <var> is None:` block.
+  enclosing_func = self._current_func_node()
+  if enclosing_func is None:
+   return False
+  for node in ast.walk(enclosing_func):
+   if node is enclosing_func:
+    continue
+   if isinstance(node, ast.If):
+    test = node.test
+    if isinstance(test, ast.Compare) and isinstance(test.ops[0], ast.Is) and any(isinstance(c, ast.Constant) and c.value is None for c in test.comparators):
+     for child in ast.walk(node):
+      if isinstance(child, ast.Raise) and child is not node:
+       return True
+  return False
+
+ def _function_body_is_just_this_raise(self, node: ast.Raise) -> bool:
+  """True if the function body is just this raise (abstract method pattern)."""
+  enclosing_func = self._current_func_node()
+  if enclosing_func is None:
+   return False
+  body = enclosing_func.body
+  if len(body) != 1:
+   return False
+  return body[0] is node
+
 def _extract_raise_name(self, node: ast.expr) -> str:
  """Extract the exception class name from a raise expression.

@@ -512,6 +706,21 @@ class ExceptionVisitor(ast.NodeVisitor):
     "INTERNAL_PROGRAMMER_RAISE",
     f"Compliant: `{exc_short}` for an impossible state / precondition check. The styleguide reserves `raise` for programmer errors.",
    )
+   # Heuristic added by result_migration_review_pass_20260617:
+   # NotImplementedError as the entire function body = abstract method pattern.
+   if exc_short == "NotImplementedError" and self._function_body_is_just_this_raise(node):
+    return (
+     "INTERNAL_PROGRAMMER_RAISE",
+     f"Compliant: `raise NotImplementedError()` as the entire function body is the canonical abstract-method pattern (per result_migration_review_pass_20260617).",
+    )
+
+  # Heuristic added by result_migration_review_pass_20260617:
+  # `if <var> is None: raise ImportError(...)` = validation raise (precondition check).
+  if exc_short in {"ImportError", "RuntimeError", "ValueError", "KeyError"} and self._enclosing_if_is_none_guard():
+   return (
+    "INTERNAL_PROGRAMMER_RAISE",
+    f"Compliant: `raise {exc_short}` inside `if <var> is None:` is the canonical validation/precondition-check pattern (per result_migration_review_pass_20260617).",
+   )

  return (
   "INTERNAL_RETHROW",
@@ -0,0 +1,291 @@
+"""Tests for the new heuristics added in scripts/audit_exception_handling.py.
+
+Each test creates a fixture file with a specific pattern and runs the audit
+script against it, verifying the pattern is classified correctly.
+
+The new heuristics (added by result_migration_review_pass_20260617):
+1. list.index with ValueError fallback to default index (INTERNAL_COMPLIANT)
+2. dict lookup (KeyError) with default value (INTERNAL_COMPLIANT)
+3. datetime.fromisoformat(s) with ValueError: None (INTERNAL_COMPLIANT)
+4. Path.resolve(strict=True) with OSError/ValueError fallback (INTERNAL_COMPLIANT)
+5. Path.relative_to with ValueError: pass (INTERNAL_COMPLIANT)
+6. asyncio.get_running_loop() with RuntimeError: asyncio.run() (INTERNAL_COMPLIANT)
+7. Narrow except (ImportError, AttributeError) + fallback stub (INTERNAL_COMPLIANT)
+8. raise NotImplementedError() as entire function body (INTERNAL_PROGRAMMER_RAISE)
+9. if None: raise ImportError() validation pattern (INTERNAL_PROGRAMMER_RAISE)
+10. try/except (json.JSONDecodeError, KeyError) around JSON parse + print + return (INTERNAL_COMPLIANT)
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "scripts" / "audit_exception_handling.py"
+
+
+def _run_audit_on_fixture(source: str) -> dict:
+ """Run the audit script on a fixture and return the parsed JSON output."""
+ # Use a temp dir outside tests/artifacts/ to avoid the audit's
+ # default artifacts-exclusion filter.
+ import tempfile
+ tmpdir = Path(tempfile.mkdtemp(prefix="audit_fixture_"))
+ fixture = tmpdir / "audit_heuristic_fixture.py"
+ fixture.write_text(textwrap.dedent(source), encoding="utf-8")
+ try:
+  result = subprocess.run(
+   [sys.executable, str(SCRIPT), "--json", "--src", str(tmpdir), "--verbose"],
+   capture_output=True,
+   text=True,
+   check=False,
+   cwd=str(ROOT),
+  )
+ finally:
+  if fixture.exists():
+   fixture.unlink()
+  if tmpdir.exists():
+   tmpdir.rmdir()
+ if result.returncode not in (0, 1):
+  raise RuntimeError(f"audit failed: {result.stderr}")
+ return json.loads(result.stdout)
+
+
+def _classifications_for_file(data: dict, filename_suffix: str) -> list[dict]:
+ """Return all findings for files whose path ends with `filename_suffix`."""
+ return [
+  f
+  for file_info in data.get("files", [])
+  for f in file_info.get("findings", [])
+  if file_info["filename"].endswith(filename_suffix)
+ ]
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 1: list.index with ValueError fallback to default index
+# ---------------------------------------------------------------------------
+def test_list_index_valueerror_fallback_is_compliant():
+ """try: list.index(x); except ValueError: idx = N is compliant."""
+ src = '''
+ def func(items, target):
+  try:
+   idx = items.index(target)
+  except ValueError:
+   idx = 0
+  return idx
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 1, f"expected 1 except, got {len(excepts)}: {excepts}"
+ assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
+  f"list.index+ValueError fallback should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 2: KeyError lookup with default value
+# ---------------------------------------------------------------------------
+def test_keyerror_lookup_with_default_is_compliant():
+ """try: dict[x]; except KeyError: val = default is compliant."""
+ src = '''
+ def func(d, key):
+  try:
+   val = d[key]
+  except KeyError:
+   val = "default"
+  return val
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 1
+ assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
+  f"KeyError+default should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 3: datetime.fromisoformat(s) with ValueError: None
+# ---------------------------------------------------------------------------
+def test_fromisoformat_valueerror_none_is_compliant():
+ """try: datetime.fromisoformat(s); except ValueError: x = None is compliant."""
+ src = '''
+ import datetime
+ def func(s):
+  try:
+   d = datetime.datetime.fromisoformat(s)
+  except ValueError:
+   d = None
+  return d
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 1
+ assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
+  f"fromisoformat+ValueError:None should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 4: Path.resolve(strict=True) with OSError/ValueError fallback
+# ---------------------------------------------------------------------------
+def test_path_resolve_strict_fallback_is_compliant():
+ """try: Path(p).resolve(strict=True); except (OSError, ValueError): Path(p).resolve() is compliant."""
+ src = '''
+ from pathlib import Path
+ def func(p):
+  try:
+   rp = Path(p).resolve(strict=True)
+  except (OSError, ValueError):
+   rp = Path(p).resolve()
+  return rp
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 1
+ assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
+  f"Path.resolve(strict=True)+fallback should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 5: Path.relative_to with ValueError: pass
+# ---------------------------------------------------------------------------
+def test_path_relative_to_valueerror_is_compliant():
+ """try: rp.relative_to(base); except ValueError: pass is compliant (canonical subpath check)."""
+ src = '''
+ from pathlib import Path
+ def is_subpath(rp, base):
+  try:
+   rp.relative_to(base)
+   return True
+  except ValueError:
+   return False
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 1
+ assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
+  f"Path.relative_to+ValueError should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 6: asyncio.get_running_loop() with RuntimeError: asyncio.run()
+# ---------------------------------------------------------------------------
+def test_asyncio_get_running_loop_fallback_is_compliant():
+ """try: get_running_loop(); except RuntimeError: asyncio.run() is compliant."""
+ src = '''
+ import asyncio
+ def bridge(coro):
+  try:
+   loop = asyncio.get_running_loop()
+   return loop
+  except RuntimeError:
+   return asyncio.run(coro)
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 1
+ assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
+  f"get_running_loop+RuntimeError fallback should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 7: narrow except (ImportError, AttributeError) + fallback stub
+# ---------------------------------------------------------------------------
+def test_import_attr_fallback_stub_is_compliant():
+ """narrow except (ImportError, AttributeError) with fallback attribute assignment is compliant."""
+ src = '''
+ class Stub:
+  available = False
+ def get_thing():
+  try:
+   import real_module
+   return real_module
+  except (ImportError, ModuleNotFoundError, AttributeError):
+   return Stub()
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 1
+ assert excepts[0]["category"] == "INTERNAL_COMPLIANT", (
+  f"narrow except + fallback stub should be INTERNAL_COMPLIANT, got {excepts[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 8: raise NotImplementedError() as entire function body
+# ---------------------------------------------------------------------------
+def test_raise_notimplemented_entire_body_is_programmer_error():
+ """raise NotImplementedError() as the entire function body is INTERNAL_PROGRAMMER_RAISE."""
+ src = '''
+ class Base:
+  def abstract_method(self):
+   raise NotImplementedError()
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ raises = [f for f in findings if f["kind"] == "RAISE"]
+ assert len(raises) == 1
+ assert raises[0]["category"] == "INTERNAL_PROGRAMMER_RAISE", (
+  f"raise NotImplementedError() in entire body should be INTERNAL_PROGRAMMER_RAISE, got {raises[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 9: if None: raise ImportError() validation
+# ---------------------------------------------------------------------------
+def test_validation_raise_is_programmer_error():
+ """if <var> is None: raise ImportError() is INTERNAL_PROGRAMMER_RAISE."""
+ src = '''
+ def func(dep):
+  if dep is None:
+   raise ImportError("dependency missing")
+  return dep
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ raises = [f for f in findings if f["kind"] == "RAISE"]
+ assert len(raises) == 1
+ assert raises[0]["category"] == "INTERNAL_PROGRAMMER_RAISE", (
+  f"validation raise should be INTERNAL_PROGRAMMER_RAISE, got {raises[0]['category']}"
+ )
+
+
+# ---------------------------------------------------------------------------
+# Heuristic 10: try/except (json.JSONDecodeError, KeyError) + print + return
+# ---------------------------------------------------------------------------
+def test_json_parse_with_print_is_compliant():
+ """try/except (json.JSONDecodeError, KeyError) around JSON parse with print is compliant."""
+ src = '''
+ import json
+ def parse(s):
+  try:
+   data = json.loads(s)
+   if not isinstance(data, list):
+    print("not a list")
+    return
+   return data
+  except json.JSONDecodeError as e:
+   print(f"json error: {e}")
+  except KeyError as e:
+   print(f"missing key: {e}")
+ '''
+ data = _run_audit_on_fixture(src)
+ findings = _classifications_for_file(data, "audit_heuristic_fixture.py")
+ excepts = [f for f in findings if f["kind"] == "EXCEPT"]
+ assert len(excepts) == 2
+ for e in excepts:
+  assert e["category"] == "INTERNAL_COMPLIANT", (
+   f"json parse with print should be INTERNAL_COMPLIANT, got {e['category']}"
+  )