/.pre-commit-hooks/check-silent-cleanup.py

#!/usr/bin/env python3
"""
Pre-commit hook to flag unsafe cleanup patterns.

Check 1 — Silent close() methods:
  Detects `def close(self)` methods that contain bare `except Exception: pass`
  or `except: pass` — cleanup failures should be logged via safe_close() from
  utilities.resource_utils, or with explicit logger calls.

Check 2 — Bare .close() in finally/except blocks:
  Detects `obj.close()` calls inside finally/except blocks that are not wrapped
  in safe_close(). If .close() raises, it masks the original exception.
"""

import ast
import sys

# Files/patterns that are excluded from this check
ALLOWED_PATTERNS = {
    "tests/",
    "test_",
    "_test.py",
    # LLM wrapper close() methods delegate to _close_base_llm which has
    # selective httpx-client introspection logic — not a simple .close() call.
    "config/llm_config.py",
    "rate_limiting/llm/wrapper.py",
}


class SilentCleanupChecker(ast.NodeVisitor):
    """AST visitor to detect silent exceptions in close() methods."""

    def __init__(self, filename: str):
        self.filename = filename
        self.errors = []

    def visit_FunctionDef(self, node):
        # Only inspect `def close(self, ...):` methods
        if node.name != "close":
            self.generic_visit(node)
            return

        # Walk the close() body looking for try/except
        for child in ast.walk(node):
            if not isinstance(child, ast.ExceptHandler):
                continue

            # Check if the handler is "except Exception:" or bare "except:"
            if child.type is not None and not (
                isinstance(child.type, ast.Name)
                and child.type.id == "Exception"
            ):
                continue

            # Check if handler body is *only* `pass` (no logging, no raise)
            if self._is_silent_handler(child):
                self.errors.append(
                    (
                        child.lineno,
                        "Silent exception in close() method — use "
                        "safe_close() from utilities.resource_utils "
                        "or add explicit logging so cleanup failures "
                        "are visible.",
                    )
                )

        self.generic_visit(node)

    @staticmethod
    def _is_silent_handler(handler: ast.ExceptHandler) -> bool:
        """Return True if handler body is only `pass` with no logging/raise."""
        for stmt in handler.body:
            # pass statement — continue checking
            if isinstance(stmt, ast.Pass):
                continue

            # raise — not silent
            if isinstance(stmt, ast.Raise):
                return False

            # Any expression that looks like logger.something(...)
            if isinstance(stmt, ast.Expr) and isinstance(stmt.value, ast.Call):
                func = stmt.value.func
                if isinstance(func, ast.Attribute) and isinstance(
                    func.value, ast.Name
                ):
                    if func.value.id == "logger":
                        return False

            # Any other statement means it's not just `pass`
            return False

        # If we got here, every statement was `pass`
        return True

    def _is_file_allowed(self) -> bool:
        for pattern in ALLOWED_PATTERNS:
            if pattern in self.filename:
                return True
        return False


class BareCloseInFinallyChecker(ast.NodeVisitor):
    """AST visitor to detect bare .close() calls in finally/except blocks.

    A bare `obj.close()` in a finally or except block can mask the original
    exception if .close() itself raises. Use safe_close() instead.
    """

    def __init__(self, filename: str):
        self.filename = filename
        self.errors = []

    def visit_Try(self, node):
        # Check finally body
        for stmt in node.finalbody:
            self._check_block(stmt)
        # Check except handlers
        for handler in node.handlers:
            for stmt in handler.body:
                self._check_block(stmt)
        self.generic_visit(node)

    def _check_block(self, node):
        """Check a single statement for bare .close() calls.

        Recurses into if/elif/else bodies but stops at nested try blocks
        (which would provide their own exception protection).
        """
        # Direct .close() call as a statement
        if self._is_bare_close(node):
            self.errors.append(
                (
                    node.lineno,
                    "Bare .close() in finally/except block — if "
                    ".close() raises, it masks the original exception. "
                    "Use safe_close() from utilities.resource_utils.",
                )
            )
            return

        # Recurse into if/elif/else bodies (common pattern: if x is not None: x.close())
        if isinstance(node, ast.If):
            for stmt in node.body:
                self._check_block(stmt)
            for stmt in node.orelse:
                self._check_block(stmt)

    # Names that are not resource handles — .close() on these is safe
    _SAFE_CLOSE_NAMES = {"plt", "figure", "fig", "ax"}

    @classmethod
    def _is_bare_close(cls, node) -> bool:
        """Return True if node is an expression statement calling .close()."""
        if not isinstance(node, ast.Expr):
            return False
        if not isinstance(node.value, ast.Call):
            return False
        call = node.value
        if not (
            isinstance(call.func, ast.Attribute)
            and call.func.attr == "close"
            and not call.args
            and not call.keywords
        ):
            return False
        # Skip known-safe names (e.g. plt.close(), cursor.close())
        if isinstance(call.func.value, ast.Name):
            if call.func.value.id in cls._SAFE_CLOSE_NAMES:
                return False
        return True


def check_file(filename: str) -> bool:
    """Check a single Python file for silent cleanup patterns."""
    if not filename.endswith(".py"):
        return True

    # Skip allowed files
    for pattern in ALLOWED_PATTERNS:
        if pattern in filename:
            return True

    try:
        with open(filename, "r", encoding="utf-8") as f:
            content = f.read()
    except Exception as e:
        print(f"Error reading {filename}: {e}")
        return False

    try:
        tree = ast.parse(content, filename=filename)

        checker = SilentCleanupChecker(filename)
        checker.visit(tree)

        bare_checker = BareCloseInFinallyChecker(filename)
        bare_checker.visit(tree)

        all_errors = checker.errors + bare_checker.errors
        if all_errors:
            print(f"\n{filename}:")
            for line_num, error in sorted(all_errors):
                print(f"  Line {line_num}: {error}")
            return False

    except SyntaxError:
        pass
    except Exception as e:
        print(f"Error parsing {filename}: {e}")
        return False

    return True


def main():
    """Main function to check all provided Python files."""
    if len(sys.argv) < 2:
        print("Usage: check-silent-cleanup.py <file1> <file2> ...")
        sys.exit(1)

    files_to_check = sys.argv[1:]
    has_errors = False

    for filename in files_to_check:
        if not check_file(filename):
            has_errors = True

    if has_errors:
        print("\n" + "=" * 70)
        print("Silent Cleanup: Unsafe close() patterns detected!")
        print("=" * 70)
        print("\nTo fix, replace silent try/except blocks with safe_close():")
        print("  from ...utilities.resource_utils import safe_close")
        print('  safe_close(self.client, "client name")')
        print("\nOr add explicit logging:")
        print('  logger.warning("Failed to close resource", exc_info=True)')
        sys.exit(1)
    else:
        sys.exit(0)


if __name__ == "__main__":
    main()