learningcircuit--local-deep-research/tests/hooks/test_custom_checks_raw_sql.py

"""
Tests for the raw-SQL check inside the custom-checks pre-commit hook.

Covers regressions for previously-missed patterns:
- f-string SQL (e.g. f"SELECT ... FROM ...") — highest-risk form (injection)
- trailing `# SQL` comments that previously bypassed the check
- auth_db.py allowlist inconsistency between execute and statement patterns
- loose "test" substring filename matching (e.g. attestation_service.py)
"""

import subprocess
import sys
import tempfile
from pathlib import Path


HOOK_SCRIPT = (
    Path(__file__).parent.parent.parent
    / ".pre-commit-hooks"
    / "custom-checks.py"
)


def _run_hook(content: str, filename: str) -> subprocess.CompletedProcess:
    """Write content to a temp file with a specific basename (may include subdirs) and run the hook."""
    with tempfile.TemporaryDirectory() as d:
        path = Path(d) / filename
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content)
        return subprocess.run(
            [sys.executable, str(HOOK_SCRIPT), str(path)],
            capture_output=True,
            text=True,
        )


class TestFlagsRawSql:
    """Patterns that must be flagged by the hook."""

    def test_plain_cursor_execute(self):
        result = _run_hook(
            'cursor.execute("SELECT * FROM users")\n',
            "service.py",
        )
        assert result.returncode == 1
        assert "Raw SQL" in result.stdout

    def test_fstring_cursor_execute(self):
        """f-string SQL is an injection risk — must not be silently allowed."""
        result = _run_hook(
            'cursor.execute(f"DELETE FROM users WHERE id = {user_id}")\n',
            "service.py",
        )
        assert result.returncode == 1
        assert "Raw SQL" in result.stdout

    def test_fstring_conn_execute(self):
        result = _run_hook(
            'conn.execute(f"INSERT INTO logs VALUES ({val})")\n',
            "service.py",
        )
        assert result.returncode == 1

    def test_fstring_session_execute(self):
        result = _run_hook(
            "session.execute(f\"UPDATE users SET name = '{n}'\")\n",
            "service.py",
        )
        assert result.returncode == 1

    def test_trailing_sql_comment_does_not_bypass(self):
        """Previously, `# SQL` anywhere on the line disabled the whole check."""
        result = _run_hook(
            'cursor.execute("DELETE FROM users")  # SQL cleanup\n',
            "service.py",
        )
        assert result.returncode == 1

    def test_attestation_filename_not_exempted(self):
        """'attestation' contains the substring 'test' but is not a test file."""
        result = _run_hook(
            'cursor.execute("SELECT * FROM certificates")\n',
            "attestation_service.py",
        )
        assert result.returncode == 1

    def test_uppercase_f_prefix(self):
        """Uppercase F-string prefix must also be caught (re.IGNORECASE)."""
        result = _run_hook(
            'query = F"SELECT * FROM users WHERE id = {uid}"\n',
            "service.py",
        )
        assert result.returncode == 1
        assert "Raw SQL" in result.stdout

    def test_rf_prefix(self):
        """rf"..." raw-f-string prefix variant must be caught."""
        result = _run_hook(
            'query = rf"DELETE FROM users WHERE id = {uid}"\n',
            "service.py",
        )
        assert result.returncode == 1

    def test_fr_prefix(self):
        """fr"..." f-raw-string prefix variant must be caught."""
        result = _run_hook(
            "query = fr\"UPDATE users SET name = '{n}'\"\n",
            "service.py",
        )
        assert result.returncode == 1


class TestAllowsLegitimateUsage:
    """Patterns that must continue to pass."""

    def test_orm_query(self):
        result = _run_hook(
            "session.query(User).filter(User.id == 1).all()\n",
            "service.py",
        )
        assert result.returncode == 0

    def test_text_escape_hatch(self):
        """SQLAlchemy text() is the sanctioned way to run raw SQL."""
        result = _run_hook(
            'session.execute(text("SELECT 1"))\n',
            "service.py",
        )
        assert result.returncode == 0

    def test_test_file_exempted(self):
        result = _run_hook(
            'cursor.execute("DELETE FROM users")\n',
            "test_example.py",
        )
        assert result.returncode == 0

    def test_migration_file_exempted(self):
        result = _run_hook(
            'cursor.execute("CREATE TABLE foo (id INT)")\n',
            "migration_001.py",
        )
        assert result.returncode == 0

    def test_auth_db_exempted_for_statement_pattern(self):
        """Regression: auth_db.py was missing from one of the two allowlists."""
        result = _run_hook(
            'cursor.execute("SELECT * FROM users")\n',
            "auth_db.py",
        )
        assert result.returncode == 0

    def test_auth_db_exempted_for_fstring_pattern(self):
        result = _run_hook(
            'conn.execute(f"INSERT INTO logs VALUES ({val})")\n',
            "auth_db.py",
        )
        assert result.returncode == 0


class TestExecuteCallVariants:
    """Every execute-call variant in db_execute_patterns must be flagged.

    Locks in current behavior so a refactor of the pattern list can't silently
    drop coverage for one of the call styles.
    """

    def test_cursor_executemany_flagged(self):
        result = _run_hook(
            'cursor.executemany("INSERT INTO users VALUES (?)", rows)\n',
            "service.py",
        )
        assert result.returncode == 1

    def test_conn_execute_flagged(self):
        result = _run_hook(
            'conn.execute("DELETE FROM users WHERE id = 1")\n',
            "service.py",
        )
        assert result.returncode == 1

    def test_connection_execute_flagged(self):
        result = _run_hook(
            'connection.execute("SELECT * FROM users")\n',
            "service.py",
        )
        assert result.returncode == 1


class TestExemptionBranches:
    """Every branch of _is_raw_sql_exempt must be covered.

    If any branch silently regresses (e.g. DB_UTIL_FILES entry removed,
    _test.py suffix check dropped), one of these will fail.
    """

    def test_tests_directory_path_exempted(self):
        """Files under a /tests/ directory are exempt even without test_ prefix."""
        result = _run_hook(
            'cursor.execute("DELETE FROM users")\n',
            "tests/helpers.py",
        )
        assert result.returncode == 0

    def test_underscore_test_suffix_exempted(self):
        """The `_test.py` suffix (Go-style) is an exempted test convention."""
        result = _run_hook(
            'cursor.execute("DELETE FROM users")\n',
            "model_test.py",
        )
        assert result.returncode == 0

    def test_alembic_filename_exempted(self):
        result = _run_hook(
            'cursor.execute("CREATE TABLE foo (id INT)")\n',
            "alembic_env.py",
        )
        assert result.returncode == 0

    def test_sqlcipher_utils_exempted(self):
        result = _run_hook(
            'cursor.execute("SELECT * FROM users")\n',
            "sqlcipher_utils.py",
        )
        assert result.returncode == 0

    def test_backup_service_exempted(self):
        result = _run_hook(
            'cursor.execute("SELECT * FROM users")\n',
            "backup_service.py",
        )
        assert result.returncode == 0

    def test_encrypted_db_exempted(self):
        result = _run_hook(
            'cursor.execute("SELECT * FROM users")\n',
            "encrypted_db.py",
        )
        assert result.returncode == 0


class TestSkipLogic:
    """Lines that start with comments / docstrings / are empty must be skipped.

    The skip block at the top of the per-line loop is load-bearing: without it,
    any file explaining SQL in comments would be flagged. Lock in its behavior.
    """

    def test_full_line_comment_with_sql_not_flagged(self):
        """A comment line containing SQL keywords must not be flagged."""
        result = _run_hook(
            '# Example: cursor.execute("SELECT * FROM users")\n',
            "service.py",
        )
        assert result.returncode == 0

    def test_docstring_line_with_sql_not_flagged(self):
        """A line that starts with triple-quotes must be skipped."""
        result = _run_hook(
            '"""cursor.execute(\'SELECT * FROM users\')"""\n',
            "service.py",
        )
        assert result.returncode == 0

    def test_single_quote_docstring_line_not_flagged(self):
        result = _run_hook(
            "'''SELECT * FROM users'''\n",
            "service.py",
        )
        assert result.returncode == 0


class TestOrmPatternsNotFlagged:
    """Legitimate ORM builder chains must never trigger raw-SQL detection."""

    def test_filter_by_not_flagged(self):
        result = _run_hook(
            "session.query(User).filter_by(id=1).all()\n",
            "service.py",
        )
        assert result.returncode == 0

    def test_join_order_by_not_flagged(self):
        result = _run_hook(
            "session.query(User).join(Role).order_by(User.id).all()\n",
            "service.py",
        )
        assert result.returncode == 0

    def test_group_by_not_flagged(self):
        result = _run_hook(
            "session.query(User).group_by(User.role).all()\n",
            "service.py",
        )
        assert result.returncode == 0