learningcircuit--local-deep-research/tests/utilities/test_log_utils_coverage.py

"""
Coverage tests for log_utils module.

Targets branches not exercised by test_log_utils.py and
test_log_utils_extended.py:
- InterceptHandler.emit: depth-walking loop stops at non-logging frame
- _get_research_id: None when no app context and no record extra
- _get_research_id: reads from record["extra"]["research_id"]
- _process_log_queue: stop-event exits the loop; None entry is skipped
- database_sink: username extraction from record["extra"]["username"]
- flush_log_queue: handles exception inside loop gracefully
- config_logger: debug=True path warns about unsafe logging
- config_logger: file logging disabled by default (no 4th add call)
"""

_logging = __import__("logging")  # test needs stdlib LogRecord
from datetime import datetime  # noqa: E402
from unittest.mock import Mock, patch  # noqa: E402


MODULE = "local_deep_research.utilities.log_utils"


# ---------------------------------------------------------------------------
# InterceptHandler
# ---------------------------------------------------------------------------


class TestInterceptHandlerDepthWalking:
    """The depth-walking loop in emit should traverse past logging frames."""

    def test_emit_with_valid_level(self):
        from local_deep_research.utilities.log_utils import InterceptHandler

        handler = InterceptHandler()
        record = _logging.LogRecord(
            name="test",
            level=_logging.WARNING,
            pathname="test.py",
            lineno=10,
            msg="warn",
            args=(),
            exc_info=None,
        )

        with patch(f"{MODULE}.logger") as mock_logger:
            mock_level = Mock()
            mock_level.name = "WARNING"
            mock_logger.level.return_value = mock_level
            mock_opt = Mock()
            mock_logger.opt.return_value = mock_opt

            handler.emit(record)

            mock_opt.log.assert_called_once()
            # level argument should be the resolved name
            args = mock_opt.log.call_args[0]
            assert args[0] == "WARNING"

    def test_emit_with_unknown_level_uses_levelno(self):
        from local_deep_research.utilities.log_utils import InterceptHandler

        handler = InterceptHandler()
        record = _logging.LogRecord(
            name="test",
            level=37,
            pathname="test.py",
            lineno=1,
            msg="custom",
            args=(),
            exc_info=None,
        )
        record.levelname = "CUSTOM_LEVEL"

        with patch(f"{MODULE}.logger") as mock_logger:
            mock_logger.level.side_effect = ValueError("unknown")
            mock_opt = Mock()
            mock_logger.opt.return_value = mock_opt

            handler.emit(record)

            # levelno (37) used when name lookup fails
            args = mock_opt.log.call_args[0]
            assert args[0] == 37


# ---------------------------------------------------------------------------
# _get_research_id
# ---------------------------------------------------------------------------


class TestGetResearchId:
    def test_returns_none_when_no_context_and_no_extra(self):
        from local_deep_research.utilities.log_utils import _get_research_id

        with patch(f"{MODULE}.has_app_context", return_value=False):
            result = _get_research_id(record={"extra": {}})
        assert result is None

    def test_returns_id_from_record_extra(self):
        from local_deep_research.utilities.log_utils import _get_research_id

        record = {"extra": {"research_id": "record-123"}}
        result = _get_research_id(record=record)
        assert result == "record-123"

    def test_returns_id_from_flask_g_when_no_extra(self):
        from local_deep_research.utilities.log_utils import _get_research_id

        mock_g = Mock()
        mock_g.get.return_value = "flask-456"

        with patch(f"{MODULE}.has_app_context", return_value=True):
            with patch(f"{MODULE}.g", mock_g):
                result = _get_research_id(record={"extra": {}})

        assert result == "flask-456"

    def test_returns_none_with_no_record(self):
        from local_deep_research.utilities.log_utils import _get_research_id

        with patch(f"{MODULE}.has_app_context", return_value=False):
            result = _get_research_id(record=None)
        assert result is None


# ---------------------------------------------------------------------------
# database_sink – username extraction
# ---------------------------------------------------------------------------


class TestDatabaseSinkUsernameExtraction:
    def _make_message(self, extra=None):
        msg = Mock()
        msg.record = {
            "time": datetime.now(),
            "message": "test",
            "name": "mod",
            "function": "fn",
            "line": 1,
            "level": Mock(name="INFO"),
            "extra": extra or {},
        }
        return msg

    def test_username_from_extra_is_queued(self):
        from local_deep_research.utilities.log_utils import database_sink
        import local_deep_research.utilities.log_utils as mod

        msg = self._make_message(extra={"username": "alice"})

        with patch.object(mod, "has_app_context", return_value=False):
            with patch.object(mod, "_log_queue") as mock_q:
                database_sink(msg)

                queued = mock_q.put_nowait.call_args[0][0]
                assert queued["username"] == "alice"

    def test_no_username_queues_none(self):
        """When research_id is set but username isn't, queue entry has
        username=None — the daemon's get_user_db_session will then look it
        up via session/thread-context fallbacks."""
        from local_deep_research.utilities.log_utils import database_sink
        import local_deep_research.utilities.log_utils as mod

        msg = self._make_message(extra={"research_id": "rid-1"})

        with patch.object(mod, "has_app_context", return_value=False):
            with patch.object(mod, "_log_queue") as mock_q:
                database_sink(msg)

                queued = mock_q.put_nowait.call_args[0][0]
                assert queued["username"] is None

    def test_no_research_context_does_not_queue(self):
        """ResearchLog is research-scoped — logs with neither research_id
        nor username should be skipped, not queued."""
        from local_deep_research.utilities.log_utils import database_sink
        import local_deep_research.utilities.log_utils as mod

        msg = self._make_message(extra={})

        with patch.object(mod, "has_app_context", return_value=False):
            with patch.object(mod, "_log_queue") as mock_q:
                database_sink(msg)

                mock_q.put_nowait.assert_not_called()


# ---------------------------------------------------------------------------
# flush_log_queue – exception inside loop
# ---------------------------------------------------------------------------


class TestFlushLogQueueExceptionHandling:
    def test_exception_in_write_does_not_abort_flush(self):
        """If _write_log_to_database raises, flush should continue."""
        from local_deep_research.utilities.log_utils import flush_log_queue
        import local_deep_research.utilities.log_utils as mod

        entry = {
            "timestamp": datetime.now(),
            "message": "m",
            "module": "m",
            "function": "f",
            "line_no": 1,
            "level": "INFO",
            "research_id": None,
            "username": None,
        }

        with patch.object(mod, "_log_queue") as mock_q:
            mock_q.empty.side_effect = [False, True]
            mock_q.get_nowait.return_value = entry

            with patch.object(
                mod, "_write_log_to_database", side_effect=Exception("db fail")
            ):
                # Should not raise
                flush_log_queue()


# ---------------------------------------------------------------------------
# config_logger – debug path and no-file-logging path
# ---------------------------------------------------------------------------


class TestConfigLoggerDebugAndFileLogging:
    def test_debug_true_adds_warning(self):
        from local_deep_research.utilities.log_utils import config_logger

        with patch(f"{MODULE}.logger") as mock_logger:
            config_logger("app", debug=True)

            mock_logger.warning.assert_called()
            warning_msg = mock_logger.warning.call_args[0][0]
            assert "DEBUG" in warning_msg or "debug" in warning_msg.lower()

    def test_file_logging_disabled_by_default(self):
        """With LDR_ENABLE_FILE_LOGGING unset, no file sink is added."""
        from local_deep_research.utilities.log_utils import config_logger
        import os

        env = {
            k: v
            for k, v in os.environ.items()
            if k != "LDR_ENABLE_FILE_LOGGING"
        }

        with patch.dict("os.environ", env, clear=True):
            with patch(f"{MODULE}.logger") as mock_logger:
                config_logger("app")

                # Only 3 sinks: stderr, database_sink, frontend_progress_sink
                assert mock_logger.add.call_count == 3

    def test_stderr_level_info_when_not_debug(self):
        from local_deep_research.utilities.log_utils import config_logger

        with patch(f"{MODULE}.logger") as mock_logger:
            config_logger("app", debug=False)

            # First add call is stderr
            first_add = mock_logger.add.call_args_list[0]
            assert first_add[1].get("level") == "INFO"

    def test_stderr_level_debug_when_debug_true(self):
        from local_deep_research.utilities.log_utils import config_logger

        with patch(f"{MODULE}.logger") as mock_logger:
            config_logger("app", debug=True)

            first_add = mock_logger.add.call_args_list[0]
            assert first_add[1].get("level") == "DEBUG"


# ---------------------------------------------------------------------------
# Sink-level diagnose policy (#4182): even when the operator opts into
# LDR_LOGURU_DIAGNOSE, the persisted DB sink and the browser-facing
# frontend sink must NEVER render frame locals (which can hold the
# SQLCipher master password). Only the local stderr sink may.
# ---------------------------------------------------------------------------


class TestDiagnoseSinkPolicy:
    def test_diagnose_never_enabled_on_db_or_frontend_sink(self):
        from local_deep_research.utilities.log_utils import (
            config_logger,
            database_sink,
            frontend_progress_sink,
        )
        import os

        env = {
            k: v
            for k, v in os.environ.items()
            if k != "LDR_ENABLE_FILE_LOGGING"
        }
        # Opt into diagnose the way an operator debugging locally would.
        env["LDR_LOGURU_DIAGNOSE"] = "true"

        with patch.dict("os.environ", env, clear=True):
            with patch(f"{MODULE}.logger") as mock_logger:
                config_logger("app", debug=True)

        # Map each add() call to its sink (first positional arg).
        by_sink = {
            call.args[0]: call.kwargs.get("diagnose")
            for call in mock_logger.add.call_args_list
        }

        # stderr opted in -> diagnose may be True; the persisted and
        # shipped sinks must be hard False regardless.
        assert by_sink[database_sink] is False
        assert by_sink[frontend_progress_sink] is False

    def test_stderr_gets_diagnose_when_opted_in(self):
        import sys
        from local_deep_research.utilities.log_utils import config_logger
        import os

        env = {
            k: v
            for k, v in os.environ.items()
            if k != "LDR_ENABLE_FILE_LOGGING"
        }
        env["LDR_LOGURU_DIAGNOSE"] = "true"

        with patch.dict("os.environ", env, clear=True):
            with patch(f"{MODULE}.logger") as mock_logger:
                config_logger("app", debug=True)

        stderr_calls = [
            c
            for c in mock_logger.add.call_args_list
            if c.args and c.args[0] is sys.stderr
        ]
        assert stderr_calls, "stderr sink was not configured"
        assert stderr_calls[0].kwargs.get("diagnose") is True