learningcircuit--local-deep-research/tests/security/test_deprecated_db_hook.py

"""
Tests for the check-deprecated-db pre-commit hook.

Ensures the hook correctly detects usage of deprecated database connection
methods that bypass per-user encrypted databases. This prevents user data
from being written to shared, unencrypted storage.
"""

import sys
from importlib import import_module
from pathlib import Path


# Add the pre-commit hooks directory to path
HOOKS_DIR = Path(__file__).parent.parent.parent / ".pre-commit-hooks"
sys.path.insert(0, str(HOOKS_DIR))

hook_module = import_module("check-deprecated-db")
check_file_fn = hook_module.check_file

# Build the deprecated DB name dynamically so this test file itself
# is not flagged by the check-ldr-db pre-commit hook.
_DEPRECATED_DB = "ldr" + ".db"


def _write_and_check(tmp_path, code: str, filename: str = "src/module.py"):
    """Write code to a temp file and run the checker."""
    p = tmp_path / filename
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(code, encoding="utf-8")
    return check_file_fn(str(p))


class TestDetectsDeprecatedDbConnection:
    """Ensures get_db_connection() usage is flagged."""

    def test_detects_get_db_connection_call(self, tmp_path):
        code = "conn = get_db_connection()\ncursor = conn.cursor()\n"
        issues = _write_and_check(tmp_path, code)
        assert len(issues) >= 1
        assert any("get_db_connection" in i for i in issues)

    def test_detects_get_db_connection_import(self, tmp_path):
        code = "from web.models.database import get_db_connection\n"
        issues = _write_and_check(tmp_path, code)
        assert len(issues) >= 1

    def test_detects_get_db_connection_with_args(self, tmp_path):
        code = 'conn = get_db_connection("some_path")\n'
        issues = _write_and_check(tmp_path, code)
        assert len(issues) >= 1


class TestDetectsRawSessionAccess:
    """Ensures db_manager.get_session() is flagged (leaks QueuePool FDs)."""

    def test_detects_raw_get_session(self, tmp_path):
        code = 'session = db_manager.get_session("user")\n'
        issues = _write_and_check(tmp_path, code)
        assert len(issues) >= 1
        assert any("get_session" in i for i in issues)

    def test_allows_noqa_comment(self, tmp_path):
        code = 'session = db_manager.get_session("user")  # noqa: raw-session\n'
        issues = _write_and_check(tmp_path, code)
        assert len(issues) == 0

    def test_allows_commented_code(self, tmp_path):
        code = '# session = db_manager.get_session("user")\n'
        issues = _write_and_check(tmp_path, code)
        assert len(issues) == 0


class TestDetectsSharedDbAccess:
    """Ensures direct SQLite connection to shared DB is flagged."""

    def test_detects_sqlite_connect_to_shared_db(self, tmp_path):
        code = f'conn = sqlite3.connect("{_DEPRECATED_DB}")\n'
        issues = _write_and_check(tmp_path, code)
        assert len(issues) >= 1


class TestAllowsSafePatterns:
    """Ensures correct patterns are NOT flagged."""

    def test_allows_get_user_db_session(self, tmp_path):
        code = """
with get_user_db_session(username) as session:
    results = session.query(Journal).all()
"""
        issues = _write_and_check(tmp_path, code)
        assert len(issues) == 0

    def test_allows_normal_code(self, tmp_path):
        code = """
def some_function():
    data = process_results()
    return data
"""
        issues = _write_and_check(tmp_path, code)
        assert len(issues) == 0