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

"""
Tests for the check-settings-manager-thread-safety pre-commit hook.

Ensures the hook correctly detects get_settings_manager() calls without
db_session= in background-thread functions (identified by @thread_cleanup
decorator or _background_/_auto_/_worker naming), while allowing legitimate
usage in Flask request context and when db_session= is passed explicitly.
"""

import ast
import sys
from importlib import import_module
from pathlib import Path


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

hook_module = import_module("check-settings-manager-thread-safety")  # noqa: E402
SettingsManagerThreadSafetyChecker = (
    hook_module.SettingsManagerThreadSafetyChecker
)


def _check_code(code: str, filename: str = "src/module.py") -> list:
    """Parse code and return the checker's issues list (line_no, message)."""
    tree = ast.parse(code)
    checker = SettingsManagerThreadSafetyChecker(filename)
    checker.visit(tree)
    return checker.issues


class TestThreadFunctionDetection:
    """Detection tests — functions the hook should classify as background threads."""

    def test_detects_thread_cleanup_decorator(self):
        code = """
@thread_cleanup
def some_func():
    get_settings_manager(username="alice")
"""
        issues = _check_code(code)
        assert len(issues) == 1
        assert issues[0][0] == 4

    def test_detects_thread_cleanup_decorator_call_form(self):
        """@thread_cleanup() with parentheses should also trigger."""
        code = """
@thread_cleanup()
def some_func():
    get_settings_manager(username="alice")
"""
        issues = _check_code(code)
        assert len(issues) == 1

    def test_detects_module_qualified_thread_cleanup_decorator(self):
        """@module.thread_cleanup should be detected via ast.Attribute."""
        code = """
@db.thread_cleanup
def some_func():
    get_settings_manager(username="alice")
"""
        issues = _check_code(code)
        assert len(issues) == 1

    def test_detects_background_prefix(self):
        code = """
def _background_index_worker(username):
    get_settings_manager(username=username)
"""
        issues = _check_code(code)
        assert len(issues) == 1

    def test_detects_auto_prefix(self):
        code = """
def _auto_index_documents_worker(username):
    get_settings_manager(username=username)
"""
        issues = _check_code(code)
        assert len(issues) == 1

    def test_detects_worker_suffix(self):
        code = """
def my_worker(username):
    get_settings_manager(username=username)
"""
        issues = _check_code(code)
        assert len(issues) == 1

    def test_detects_async_thread_function(self):
        """async def _background_* should be visited via visit_AsyncFunctionDef."""
        code = """
async def _background_evaluation(queue):
    get_settings_manager(username="alice")
"""
        issues = _check_code(code)
        assert len(issues) == 1


class TestCallPatternDetection:
    """Tests for how the hook recognises get_settings_manager() call shapes."""

    def test_flags_bare_call_without_db_session(self):
        code = """
@thread_cleanup
def worker():
    get_settings_manager()
"""
        assert len(_check_code(code)) == 1

    def test_flags_username_only_call(self):
        code = """
@thread_cleanup
def worker():
    get_settings_manager(username="alice")
"""
        assert len(_check_code(code)) == 1

    def test_flags_attribute_style_call(self):
        """module.get_settings_manager() should be caught via ast.Attribute."""
        code = """
@thread_cleanup
def worker():
    db_utils.get_settings_manager(username="alice")
"""
        assert len(_check_code(code)) == 1


class TestAllowances:
    """Cases that should NOT be flagged."""

    def test_allows_db_session_kwarg(self):
        code = """
@thread_cleanup
def worker(db_session):
    get_settings_manager(db_session=db_session, username="alice")
"""
        assert _check_code(code) == []

    def test_allows_db_session_only(self):
        code = """
@thread_cleanup
def worker(db_session):
    get_settings_manager(db_session=db_session)
"""
        assert _check_code(code) == []

    def test_allows_positional_db_session(self):
        """db_session is the first positional param - a positional arg counts."""
        code = """
@thread_cleanup
def worker(db_session, username):
    get_settings_manager(db_session, username)
"""
        assert _check_code(code) == []

    def test_allows_positional_db_session_only(self):
        code = """
@thread_cleanup
def worker(db_session):
    get_settings_manager(db_session)
"""
        assert _check_code(code) == []

    def test_allows_non_thread_function(self):
        """Flask route / non-thread code may call get_settings_manager() freely."""
        code = """
def route_handler():
    get_settings_manager(username="alice")
"""
        assert _check_code(code) == []

    def test_allows_unrelated_call(self):
        """Calls to other functions should never be flagged."""
        code = """
@thread_cleanup
def worker():
    get_other_thing(username="alice")
"""
        assert _check_code(code) == []

    def test_allows_thread_function_without_call(self):
        """A thread function that never calls get_settings_manager is fine."""
        code = """
@thread_cleanup
def worker():
    do_something_else()
"""
        assert _check_code(code) == []


class TestMultipleAndPositional:
    """Multi-issue reporting and line-number correctness."""

    def test_reports_multiple_violations(self):
        code = """
@thread_cleanup
def worker_one():
    get_settings_manager(username="a")
    get_settings_manager(username="b")
"""
        issues = _check_code(code)
        assert len(issues) == 2

    def test_reports_correct_line_numbers(self):
        code = """
@thread_cleanup
def worker():
    x = 1
    get_settings_manager(username="a")
"""
        issues = _check_code(code)
        assert len(issues) == 1
        assert issues[0][0] == 5

    def test_mixed_safe_and_unsafe_in_same_function(self):
        code = """
@thread_cleanup
def worker(db_session):
    get_settings_manager(db_session=db_session)  # safe
    get_settings_manager(username="x")           # unsafe
"""
        issues = _check_code(code)
        assert len(issues) == 1
        assert issues[0][0] == 5


class TestCheckFileIntegration:
    """check_file() integration tests using tmp_path."""

    def test_clean_file_passes(self, tmp_path):
        clean = tmp_path / "clean.py"
        clean.write_text("""
@thread_cleanup
def worker(db_session):
    get_settings_manager(db_session=db_session, username="x")
""")
        assert hook_module.check_file(clean) == []

    def test_violation_file_fails(self, tmp_path):
        bad = tmp_path / "bad.py"
        bad.write_text("""
@thread_cleanup
def worker():
    get_settings_manager(username="x")
""")
        issues = hook_module.check_file(bad)
        assert len(issues) == 1
        filepath, lineno, _msg = issues[0]
        assert filepath == str(bad)
        assert lineno == 4

    def test_syntax_error_is_tolerated(self, tmp_path, capsys):
        """Files with syntax errors should not raise; they print to stderr."""
        broken = tmp_path / "broken.py"
        broken.write_text("def bad(:\n    pass")
        issues = hook_module.check_file(broken)
        assert issues == []
        captured = capsys.readouterr()
        assert "Syntax error" in captured.err

    def test_non_python_file_via_main_is_skipped(self, tmp_path, monkeypatch):
        """main() should skip non-.py files without inspecting them."""
        non_py = tmp_path / "README.md"
        non_py.write_text("not python @thread_cleanup def worker():")
        monkeypatch.setattr(sys, "argv", ["hook", str(non_py)])
        assert hook_module.main() == 0

    def test_main_returns_zero_on_clean(self, tmp_path, monkeypatch):
        clean = tmp_path / "clean.py"
        clean.write_text("""
def route():
    get_settings_manager(username="x")
""")
        monkeypatch.setattr(sys, "argv", ["hook", str(clean)])
        assert hook_module.main() == 0

    def test_main_returns_one_on_violation(self, tmp_path, monkeypatch, capsys):
        bad = tmp_path / "bad.py"
        bad.write_text("""
@thread_cleanup
def worker():
    get_settings_manager(username="x")
""")
        monkeypatch.setattr(sys, "argv", ["hook", str(bad)])
        assert hook_module.main() == 1
        captured = capsys.readouterr()
        assert "get_settings_manager()" in captured.out
        assert "get_user_db_session" in captured.out

    def test_main_with_no_files_returns_zero(self, monkeypatch):
        monkeypatch.setattr(sys, "argv", ["hook"])
        assert hook_module.main() == 0


class TestNestedFunctions:
    """Tests for the nested-function walker that prevents double-reporting."""

    def test_nested_thread_in_thread_reports_once(self):
        """Unsafe call inside a nested thread fn must be reported exactly once,
        not twice (once from the outer walk + once from the inner visit)."""
        code = """
@thread_cleanup
def outer_worker(db_session):
    get_settings_manager(db_session)  # safe

    @thread_cleanup
    def inner_worker():
        get_settings_manager(username="x")  # unsafe, reported once
"""
        issues = _check_code(code)
        assert len(issues) == 1

    def test_nested_non_thread_function_not_checked(self):
        """An unsafe call inside a nested non-thread fn should not be flagged
        just because the outer function is a thread fn."""
        code = """
@thread_cleanup
def outer_worker():
    def inner_helper():
        get_settings_manager(username="x")  # NOT flagged: inner is not a thread fn
"""
        assert _check_code(code) == []

    def test_lambda_inside_thread_not_checked(self):
        """A lambda body inside a thread fn is not walked (lambdas have no name
        to classify)."""
        code = """
@thread_cleanup
def outer_worker():
    cb = lambda: get_settings_manager(username="x")
"""
        assert _check_code(code) == []


class TestThreadFunctionClassifier:
    """Unit tests for SettingsManagerThreadSafetyChecker._is_thread_function."""

    def _classify(self, code: str) -> bool:
        tree = ast.parse(code)
        func = tree.body[0]
        checker = SettingsManagerThreadSafetyChecker("x.py")
        return checker._is_thread_function(func)

    def test_plain_function_is_not_thread(self):
        assert self._classify("def foo(): pass") is False

    def test_background_prefix_is_thread(self):
        assert self._classify("def _background_x(): pass") is True

    def test_auto_prefix_is_thread(self):
        assert self._classify("def _auto_x(): pass") is True

    def test_worker_suffix_is_thread(self):
        assert self._classify("def foo_worker(): pass") is True

    def test_decorator_is_thread(self):
        code = "@thread_cleanup\ndef foo(): pass"
        assert self._classify(code) is True

    def test_unrelated_decorator_is_not_thread(self):
        code = "@staticmethod\ndef foo(): pass"
        assert self._classify(code) is False