learningcircuit--local-deep-research/tests/web/warning_checks/test_orchestrator.py

"""Tests for the calculate_warnings orchestrator (__init__.py).

These test the integration logic: session reading, DB session lifecycle,
setting dispatch to sub-checks, and error handling.
"""

from unittest.mock import Mock, patch


def _make_settings_manager(overrides=None):
    """Build a mock SettingsManager with sensible defaults."""
    defaults = {
        "llm.provider": "ollama",
        "llm.local_context_window_size": 4096,
        "llm.model": "llama3",
        "app.warnings.dismiss_high_context": False,
        "app.warnings.dismiss_model_mismatch": False,
        "app.warnings.dismiss_context_reduced": False,
        "app.warnings.dismiss_legacy_config": False,
        "app.warnings.dismiss_no_backups": False,
        "app.warnings.dismiss_backup_disabled": False,
        "backup.enabled": True,
    }
    if overrides:
        defaults.update(overrides)

    mgr = Mock()
    mgr.get_setting.side_effect = lambda key, default=None: defaults.get(
        key, default
    )
    return mgr


def _patch_orchestrator(settings_manager, db_session=None):
    """Return a nested context manager that patches session, get_user_db_session,
    and get_settings_manager for the warnings orchestrator.
    """
    if db_session is None:
        db_session = Mock()

    class _Ctx:
        def __enter__(self_ctx):
            self_ctx.p1 = patch(
                "local_deep_research.web.warning_checks.session",
                {"username": "test"},
            )
            self_ctx.p2 = patch(
                "local_deep_research.web.warning_checks.get_user_db_session"
            )
            self_ctx.p3 = patch(
                "local_deep_research.web.warning_checks.get_settings_manager",
                return_value=settings_manager,
            )
            self_ctx.p1.start()
            mock_ctx = self_ctx.p2.start()
            mock_ctx.return_value.__enter__ = Mock(return_value=db_session)
            mock_ctx.return_value.__exit__ = Mock(return_value=False)
            self_ctx.p3.start()
            return self_ctx

        def __exit__(self_ctx, *args):
            self_ctx.p3.stop()
            self_ctx.p2.stop()
            self_ctx.p1.stop()

    return _Ctx()


class TestCalculateWarningsNoDbSession:
    """Tests for when db_session is None."""

    def test_returns_empty_list_when_db_session_is_none(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        with patch(
            "local_deep_research.web.warning_checks.session",
            {"username": "test"},
        ):
            with patch(
                "local_deep_research.web.warning_checks.get_user_db_session"
            ) as mock_ctx:
                mock_ctx.return_value.__enter__ = Mock(return_value=None)
                mock_ctx.return_value.__exit__ = Mock(return_value=False)

                result = calculate_warnings()
                assert result == []

    def test_returns_empty_list_when_no_username(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        with patch(
            "local_deep_research.web.warning_checks.session",
            {},
        ):
            with patch(
                "local_deep_research.web.warning_checks.get_user_db_session"
            ) as mock_ctx:
                mock_ctx.return_value.__enter__ = Mock(return_value=None)
                mock_ctx.return_value.__exit__ = Mock(return_value=False)

                result = calculate_warnings()
                assert result == []


class TestCalculateWarningsErrorHandling:
    """Tests for exception handling."""

    def test_returns_empty_list_on_exception(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        with patch(
            "local_deep_research.web.warning_checks.session",
            {"username": "test"},
        ):
            with patch(
                "local_deep_research.web.warning_checks.get_user_db_session",
                side_effect=RuntimeError("DB exploded"),
            ):
                result = calculate_warnings()
                assert result == []

    def test_returns_empty_list_on_settings_manager_error(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        with patch(
            "local_deep_research.web.warning_checks.session",
            {"username": "test"},
        ):
            with patch(
                "local_deep_research.web.warning_checks.get_user_db_session"
            ) as mock_ctx:
                mock_ctx.return_value.__enter__ = Mock(return_value=Mock())
                mock_ctx.return_value.__exit__ = Mock(return_value=False)

                with patch(
                    "local_deep_research.web.warning_checks.get_settings_manager",
                    side_effect=ValueError("bad settings"),
                ):
                    result = calculate_warnings()
                    assert result == []


class TestCalculateWarningsProviderNormalization:
    """Provider string is lowercased before checks."""

    def test_uppercase_provider_is_normalized(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager(
            {
                "llm.provider": "OLLAMA",
                "llm.local_context_window_size": 16384,
            }
        )
        with _patch_orchestrator(mgr):
            warnings = calculate_warnings()
            assert any(w["type"] == "high_context" for w in warnings)


class TestCalculateWarningsContextCheckGating:
    """Context checks are gated on is_local AND not dismissed."""

    def test_context_checks_skipped_for_non_local_provider(self):
        """Non-local provider should never trigger context history checks,
        even when context is low and history would normally warn."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mock_db_session = Mock()

        mgr = _make_settings_manager(
            {
                "llm.provider": "openai",
                "llm.local_context_window_size": 2048,
                "app.warnings.dismiss_context_reduced": False,
            }
        )
        with _patch_orchestrator(mgr, db_session=mock_db_session):
            warnings = calculate_warnings()

        # DB should not have been queried for context history at all
        mock_db_session.query.assert_not_called()
        assert not any(w["type"] == "context_below_history" for w in warnings)
        assert not any(
            w["type"] == "context_truncation_history" for w in warnings
        )

    def test_context_checks_skipped_when_dismissed(self):
        """Local provider with dismiss_context_reduced=True should skip DB queries."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mock_db_session = Mock()

        mgr = _make_settings_manager(
            {
                "llm.provider": "ollama",
                "llm.local_context_window_size": 2048,
                "app.warnings.dismiss_context_reduced": True,
            }
        )
        with _patch_orchestrator(mgr, db_session=mock_db_session):
            warnings = calculate_warnings()

        mock_db_session.query.assert_not_called()
        assert not any(w["type"] == "context_below_history" for w in warnings)


class TestCalculateWarningsMultipleWarnings:
    """Multiple warnings can fire simultaneously."""

    def test_high_context_and_model_mismatch_together(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager(
            {
                "llm.provider": "ollama",
                "llm.local_context_window_size": 16384,
                "llm.model": "llama3.1:70b",
            }
        )
        with _patch_orchestrator(mgr):
            warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert "high_context" in types
        assert "model_mismatch" in types

    def test_high_context_and_context_history_together(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        mock_db_session = Mock()

        # DB queries for context history
        context_query = Mock()
        context_query.filter.return_value.order_by.return_value.limit.return_value.all.return_value = (
            [(32768,)] * 10
        )

        truncation_query = Mock()
        truncation_query.filter.return_value.filter.return_value.scalar.return_value = 5

        mock_db_session.query.side_effect = [context_query, truncation_query]

        mgr = _make_settings_manager(
            {
                "llm.provider": "ollama",
                "llm.local_context_window_size": 16384,
            }
        )
        with _patch_orchestrator(mgr, db_session=mock_db_session):
            warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert "high_context" in types
        assert "context_below_history" in types
        assert "context_truncation_history" in types

    def test_all_hardware_and_context_warnings_simultaneously(self):
        from local_deep_research.web.warning_checks import calculate_warnings

        mock_db_session = Mock()

        context_query = Mock()
        context_query.filter.return_value.order_by.return_value.limit.return_value.all.return_value = (
            [(32768,)] * 10
        )

        truncation_query = Mock()
        truncation_query.filter.return_value.filter.return_value.scalar.return_value = 2

        mock_db_session.query.side_effect = [context_query, truncation_query]

        mgr = _make_settings_manager(
            {
                "llm.provider": "ollama",
                "llm.local_context_window_size": 16384,
                "llm.model": "llama3.1:70b",
            }
        )
        with _patch_orchestrator(mgr, db_session=mock_db_session):
            with patch(
                "local_deep_research.web.server_config.get_server_config_path",
                return_value=Mock(exists=Mock(return_value=False)),
            ):
                warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert types == {
            "high_context",
            "model_mismatch",
            "context_below_history",
            "context_truncation_history",
            "no_backups",
            # The egress policy defaults to scope="adaptive" (unacknowledged):
            # the public-egress banner fires (adaptive can resolve to a
            # public-allowing scope) alongside the informational banner that
            # states what adaptive resolves to.
            "public_egress_enabled",
            "egress_effective_scope",
        }


class TestCalculateWarningsSingleSession:
    """Verify all settings are read from a single DB session."""

    def test_get_user_db_session_called_once(self):
        """The orchestrator should open exactly one DB session."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager()

        with patch(
            "local_deep_research.web.warning_checks.session",
            {"username": "test"},
        ):
            with patch(
                "local_deep_research.web.warning_checks.get_user_db_session"
            ) as mock_ctx:
                mock_ctx.return_value.__enter__ = Mock(return_value=Mock())
                mock_ctx.return_value.__exit__ = Mock(return_value=False)

                with patch(
                    "local_deep_research.web.warning_checks.get_settings_manager",
                    return_value=mgr,
                ):
                    calculate_warnings()

        # Should be called exactly once (the whole point of the refactor)
        assert mock_ctx.call_count == 1

    def test_get_user_db_session_called_with_username(self):
        """Verify username from session is passed to get_user_db_session."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager()

        with patch(
            "local_deep_research.web.warning_checks.session",
            {"username": "alice"},
        ):
            with patch(
                "local_deep_research.web.warning_checks.get_user_db_session"
            ) as mock_get_db:
                mock_get_db.return_value.__enter__ = Mock(return_value=Mock())
                mock_get_db.return_value.__exit__ = Mock(return_value=False)

                with patch(
                    "local_deep_research.web.warning_checks.get_settings_manager",
                    return_value=mgr,
                ):
                    calculate_warnings()

        mock_get_db.assert_called_once_with("alice")

    def test_get_settings_manager_called_with_session_and_username(self):
        """Verify both db_session and username are passed to get_settings_manager."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager()
        mock_db = Mock()

        with patch(
            "local_deep_research.web.warning_checks.session",
            {"username": "bob"},
        ):
            with patch(
                "local_deep_research.web.warning_checks.get_user_db_session"
            ) as mock_get_db:
                mock_get_db.return_value.__enter__ = Mock(return_value=mock_db)
                mock_get_db.return_value.__exit__ = Mock(return_value=False)

                with patch(
                    "local_deep_research.web.warning_checks.get_settings_manager",
                    return_value=mgr,
                ) as mock_get_sm:
                    calculate_warnings()

        mock_get_sm.assert_called_once_with(mock_db, "bob")

    def test_all_seven_settings_read(self):
        """All 7 required settings are read from the manager."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager()

        with _patch_orchestrator(mgr):
            calculate_warnings()

        called_keys = [call.args[0] for call in mgr.get_setting.call_args_list]
        assert "llm.provider" in called_keys
        assert "llm.local_context_window_size" in called_keys
        assert "llm.model" in called_keys
        assert "app.warnings.dismiss_high_context" in called_keys
        assert "app.warnings.dismiss_model_mismatch" in called_keys
        assert "app.warnings.dismiss_context_reduced" in called_keys
        assert "app.warnings.dismiss_legacy_config" in called_keys


class TestCalculateWarningsLegacyServerConfig:
    """Orchestrator coverage for the legacy_server_config check."""

    def test_legacy_server_config_warning_when_file_exists(self):
        """If server_config.json exists on disk with non-default values, warning appears."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager()
        mock_path = Mock()
        mock_path.exists.return_value = True
        # Provide JSON with a non-default value so the warning fires
        mock_path.read_text.return_value = '{"port": 9999}'
        with _patch_orchestrator(mgr):
            with patch(
                "local_deep_research.web.server_config.get_server_config_path",
                return_value=mock_path,
            ):
                warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert "legacy_server_config" in types

    def test_no_legacy_server_config_warning_when_file_absent(self):
        """If server_config.json does not exist, no warning."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager()
        with _patch_orchestrator(mgr):
            with patch(
                "local_deep_research.web.server_config.get_server_config_path",
                return_value=Mock(exists=Mock(return_value=False)),
            ):
                warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert "legacy_server_config" not in types

    def test_no_legacy_server_config_warning_when_dismissed(self):
        """If dismissed, no warning even when file exists."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager(
            {"app.warnings.dismiss_legacy_config": True}
        )
        with _patch_orchestrator(mgr):
            with patch(
                "local_deep_research.web.server_config.get_server_config_path",
                return_value=Mock(exists=Mock(return_value=True)),
            ):
                warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert "legacy_server_config" not in types


class TestCalculateWarningsFailureIsolation:
    """Verify _safe_check isolates individual check failures."""

    def test_first_check_crash_does_not_kill_second(self):
        """If check_high_context raises, check_model_mismatch still runs."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = _make_settings_manager(
            {
                "llm.provider": "ollama",
                "llm.local_context_window_size": 16384,
                "llm.model": "llama3.1:70b",
            }
        )
        with _patch_orchestrator(mgr):
            with patch(
                "local_deep_research.web.warning_checks.check_high_context",
                side_effect=RuntimeError("boom"),
            ):
                warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert "high_context" not in types
        assert "model_mismatch" in types

    def test_context_check_crash_does_not_kill_other_context_check(self):
        """If check_context_below_history raises, check_context_truncation_history still runs."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mock_db_session = Mock()

        truncation_query = Mock()
        truncation_query.filter.return_value.filter.return_value.scalar.return_value = 3

        mock_db_session.query.return_value = truncation_query

        mgr = _make_settings_manager(
            {
                "llm.provider": "ollama",
                "llm.local_context_window_size": 4096,
            }
        )
        with _patch_orchestrator(mgr, db_session=mock_db_session):
            with patch(
                "local_deep_research.web.warning_checks.check_context_below_history",
                side_effect=RuntimeError("db query failed"),
            ):
                warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        assert "context_below_history" not in types
        assert "context_truncation_history" in types


class TestCalculateWarningsNoneProvider:
    """Document behavior when provider setting is None."""

    def test_none_provider_does_not_crash(self):
        """If get_setting returns None for provider, outer try/except catches the AttributeError."""
        from local_deep_research.web.warning_checks import calculate_warnings

        mgr = Mock()
        mgr.get_setting.side_effect = lambda key, default=None: {
            "llm.provider": None,
            "llm.local_context_window_size": 4096,
            "llm.model": "llama3",
            "app.warnings.dismiss_high_context": False,
            "app.warnings.dismiss_model_mismatch": False,
            "app.warnings.dismiss_context_reduced": False,
        }.get(key, default)

        with _patch_orchestrator(mgr):
            result = calculate_warnings()

        # Should not crash — outer except catches AttributeError from None.lower()
        assert isinstance(result, list)


class TestCalculateWarningsBackupGlobHardening:
    """The backup glob in calculate_warnings() must skip symlinks, so a
    planted symlink cannot inflate the backup count/size used for warnings.
    """

    def test_symlink_is_not_counted_as_a_backup(self, tmp_path):
        from local_deep_research.web.warning_checks import calculate_warnings

        backup_dir = tmp_path / "backups"
        backup_dir.mkdir()

        # The ONLY entry matching the glob is a symlink to an existing file
        # outside the backup dir. With the hardening it is skipped, leaving
        # zero real backups; without it, stat() would follow the link and
        # count it.
        outside = tmp_path / "outside_secret.db"
        outside.write_bytes(b"x" * 4096)
        (backup_dir / "ldr_backup_29991231_235959.db").symlink_to(outside)

        settings_manager = _make_settings_manager()  # backup.enabled=True
        with _patch_orchestrator(settings_manager):
            with patch(
                "local_deep_research.config.paths.get_user_backup_directory",
                return_value=backup_dir,
            ):
                warnings = calculate_warnings()

        types = {w["type"] for w in warnings}
        # Symlink excluded -> backup_count == 0 -> the "no backups" warning
        # fires and no "backup_info" (which would carry the link target size).
        assert "no_backups" in types
        assert "backup_info" not in types