learningcircuit--local-deep-research/tests/database/test_no_raw_sql.py

#!/usr/bin/env python3
# allow: no-sut-import — guardian; scans the codebase for raw SQL outside allowed locations
"""Test to verify no raw SQL is used in the codebase (except in allowed locations)."""

import sys
from pathlib import Path

sys.path.insert(
    0,
    str(Path(__file__).parent.parent.parent.resolve()),
)

import re


def check_file_for_raw_sql(filepath):
    """Check a single file for raw SQL usage."""
    with open(filepath, "r", encoding="utf-8") as f:
        content = f.read()

    # Skip if it's a test file, migration file, or database-specific files
    filepath_str = str(filepath).lower()
    if any(
        skip in filepath_str
        for skip in [
            "test",
            "migration",
            "encrypted_db.py",
            "sqlcipher_utils.py",
            "thread_local_session.py",
            "queue/processor.py",
            "database/initialize.py",  # Schema migrations using DDL
            "alembic_runner.py",  # Migration runner: drops orphan _alembic_tmp_* tables (#3817), toggles foreign_keys (#3990)
            "auth_db.py",  # SQLAlchemy DDL (CreateTable/CreateIndex), not raw SQL
            "backup_service.py",  # SQLCipher ATTACH/DETACH/export operations
            "journal_quality/db.py",  # Read-only SQLite DB build + PRAGMA user_version
        ]
    ):
        return []

    violations = []
    lines = content.split("\n")

    # Patterns that indicate raw SQL
    sql_patterns = [
        r"cursor\.execute\s*\(",
        r"conn\.execute\s*\(",
        r'session\.execute\s*\(\s*["\']',
        r'["\']SELECT\s+.*FROM\s+',
        r'["\']INSERT\s+INTO\s+',
        r'["\']UPDATE\s+.*SET\s+',
        r'["\']DELETE\s+FROM\s+',
        r'["\']CREATE\s+TABLE\s+',
        r'["\']DROP\s+TABLE\s+',
        r'["\']ALTER\s+TABLE\s+',
    ]

    # Allowed patterns (SQLCipher PRAGMAs and simple connection tests)
    allowed_patterns = [
        r"PRAGMA\s+(cipher_|quick_check|rekey)",  # SQLCipher pragmas
        r"SELECT\s+1(?:\s|$|;)",  # Simple connection test
        r'text\s*\(\s*["\'](?:SELECT\s+1|PRAGMA)',  # SQLAlchemy text() with allowed queries
    ]

    for line_num, line in enumerate(lines, 1):
        for pattern in sql_patterns:
            if re.search(pattern, line, re.IGNORECASE):
                # Check if it's in a comment or docstring
                stripped = line.strip()
                if (
                    stripped.startswith("#")
                    or stripped.startswith('"""')
                    or stripped.startswith("'''")
                ):
                    continue

                # Check if it's an allowed pattern
                is_allowed = False
                for allowed in allowed_patterns:
                    if re.search(allowed, line, re.IGNORECASE):
                        is_allowed = True
                        break

                if not is_allowed:
                    violations.append((line_num, line.strip()))

    return violations


def test_no_raw_sql_in_src():
    """Test that no raw SQL is used in src directory (except allowed patterns).

    Allowed exceptions:
    - encrypted_db.py file (SQLCipher-specific operations)
    - PRAGMA commands for SQLCipher (cipher_*, quick_check, rekey)
    - Simple connection tests (SELECT 1)
    """
    src_path = Path(__file__).parent.parent.parent / "src"
    violations = {}

    # Walk through all Python files
    for filepath in src_path.rglob("*.py"):
        file_violations = check_file_for_raw_sql(filepath)
        if file_violations:
            violations[str(filepath)] = file_violations

    # Report violations
    if violations:
        print("\n❌ Found raw SQL in the following files:")
        for filepath, file_violations in violations.items():
            print(f"\n{filepath}:")
            for line_num, line in file_violations:
                print(f"  Line {line_num}: {line[:80]}...")

        # This should fail the test
        assert False, f"Found raw SQL in {len(violations)} files"
    else:
        print(
            "✓ No raw SQL found in src directory (excluding allowed patterns)"
        )


def test_orm_imports_used():
    """Files that touch the DB should reach for the ORM, not raw SQL.

    Walk the src tree, collect every file that performs a DB operation
    (``get_db_session``, references to ``ResearchHistory`` or
    ``ResearchResource``), and require that the vast majority of them
    also import SQLAlchemy or use ORM-style query helpers.

    The ratio guard is intentionally loose (≥80%) because a handful of
    helper modules legitimately reference these names without
    constructing queries themselves (e.g. type-only imports, fixtures,
    docstring examples).
    """
    src_path = Path(__file__).parent.parent.parent / "src"
    files_with_db_operations = []
    files_with_orm_imports = []

    for filepath in src_path.rglob("*.py"):
        with open(filepath, "r", encoding="utf-8") as f:
            content = f.read()

        # Check if file has database operations
        if any(
            pattern in content
            for pattern in [
                "get_db_session",
                "ResearchHistory",
                "ResearchResource",
            ]
        ):
            files_with_db_operations.append(filepath)

            # Check if it has ORM imports
            if any(
                pattern in content
                for pattern in [
                    "from sqlalchemy",
                    "import.*session",
                    ".query(",
                    ".filter",
                ]
            ):
                files_with_orm_imports.append(filepath)

    # Sanity: at least some files matched the DB-operation heuristic, or
    # the test has silently stopped finding anything (e.g., a refactor
    # renamed the symbols above and the test would otherwise pass with 0
    # files inspected).
    assert files_with_db_operations, (
        "No files with DB operations detected — patterns may be stale"
    )

    orm_ratio = len(files_with_orm_imports) / len(files_with_db_operations)
    assert orm_ratio >= 0.8, (
        f"Only {len(files_with_orm_imports)}/{len(files_with_db_operations)} "
        f"files with DB operations use ORM imports (ratio {orm_ratio:.2f} < 0.80)"
    )


def test_models_imported_from_correct_location():
    """Test that models are imported from the consolidated location."""
    src_path = Path(__file__).parent.parent.parent / "src"
    incorrect_imports = {}

    # Old import patterns that should not be used
    old_patterns = [
        r"from.*web\.models\.database import.*(?:Research|ResearchHistory)",
        r"from.*web\.database\.models import",
        r"from.*benchmarks\.models\.benchmark_models import",
    ]

    for filepath in src_path.rglob("*.py"):
        with open(filepath, "r", encoding="utf-8") as f:
            content = f.read()

        violations = []
        lines = content.split("\n")

        for line_num, line in enumerate(lines, 1):
            for pattern in old_patterns:
                if re.search(pattern, line):
                    violations.append((line_num, line.strip()))

        if violations:
            incorrect_imports[str(filepath)] = violations

    if incorrect_imports:
        print("\n❌ Found imports from old model locations:")
        for filepath, violations in incorrect_imports.items():
            print(f"\n{filepath}:")
            for line_num, line in violations:
                print(f"  Line {line_num}: {line}")

        # This should fail the test
        assert False, (
            f"Found {len(incorrect_imports)} files with incorrect model imports"
        )
    else:
        print("✓ All models imported from correct location (database.models)")


if __name__ == "__main__":
    test_no_raw_sql_in_src()
    test_orm_imports_used()
    test_models_imported_from_correct_location()
    print("\n✅ All SQL/ORM compliance tests passed!")