diff --git a/scripts/audit_exception_handling.py b/scripts/audit_exception_handling.py index 8d28b09b..c064b168 100644 --- a/scripts/audit_exception_handling.py +++ b/scripts/audit_exception_handling.py @@ -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: ; 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))}): ` 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))}): ` 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 ` 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 `.""" + 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 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 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 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 is None:` is the canonical validation/precondition-check pattern (per result_migration_review_pass_20260617).", + ) return ( "INTERNAL_RETHROW", diff --git a/tests/test_audit_exception_handling_heuristics.py b/tests/test_audit_exception_handling_heuristics.py new file mode 100644 index 00000000..814ad752 --- /dev/null +++ b/tests/test_audit_exception_handling_heuristics.py @@ -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 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']}" + )