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

"""Tests for DB-backed context warning checks.

Only mock needed: SQLAlchemy session (no Flask dependency).
"""

from unittest.mock import Mock

from local_deep_research.web.warning_checks.context import (
    check_context_below_history,
    check_context_truncation_history,
)


def _mock_db_with_context_records(records, truncation_count=0):
    """Build a mock db_session that returns *records* for the context query
    and *truncation_count* for the truncation count query.
    """
    db_session = Mock()

    context_query = Mock()
    context_query.filter.return_value.order_by.return_value.limit.return_value.all.return_value = records

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

    db_session.query.side_effect = [context_query, truncation_query]
    return db_session


class TestCheckContextBelowHistory:
    """Tests for check_context_below_history."""

    def test_warns_when_below_historical_min(self):
        db = _mock_db_with_context_records([(8192,)] * 10)
        result = check_context_below_history(db, local_context=2048)
        assert result is not None
        assert result["type"] == "context_below_history"
        assert "2,048" in result["message"]

    def test_no_warning_when_at_or_above_historical(self):
        db = _mock_db_with_context_records([(4096,)] * 10)
        result = check_context_below_history(db, local_context=4096)
        assert result is None

    def test_no_warning_fewer_than_5_records(self):
        db = _mock_db_with_context_records([(8192,)] * 3)
        result = check_context_below_history(db, local_context=2048)
        assert result is None

    def test_no_warning_empty_records(self):
        db = _mock_db_with_context_records([])
        result = check_context_below_history(db, local_context=2048)
        assert result is None

    def test_exactly_5_records_is_enough(self):
        """Boundary: exactly 5 records should be sufficient."""
        db = _mock_db_with_context_records([(8192,)] * 5)
        result = check_context_below_history(db, local_context=2048)
        assert result is not None
        assert result["type"] == "context_below_history"

    def test_exactly_4_records_not_enough(self):
        """Boundary: 4 records is below the threshold of 5."""
        db = _mock_db_with_context_records([(8192,)] * 4)
        result = check_context_below_history(db, local_context=2048)
        assert result is None

    def test_none_values_filtered_out(self):
        """Records with None context_limit are filtered from percentile calc."""
        # 6 records total but 4 are None → only 2 valid → below threshold
        records = [(None,)] * 4 + [(8192,)] * 2
        db = _mock_db_with_context_records(records)
        result = check_context_below_history(db, local_context=2048)
        # Only 2 non-None records < 5 threshold, but the query returned 6 rows
        # so len(recent_contexts) >= 5 passes, but after filtering context_values
        # has only 2 entries. Still, percentile calc should work on 2 values.
        # With sorted [8192, 8192], idx=0, min_safe=8192, 2048 < 8192 → warn
        assert result is not None

    def test_all_none_values_after_filtering(self):
        """If all records have None/0 context_limit after filtering, no warning."""
        records = [(None,)] * 5 + [(0,)] * 5
        db = _mock_db_with_context_records(records)
        result = check_context_below_history(db, local_context=2048)
        assert result is None

    def test_mixed_context_values_percentile(self):
        """With varied values, percentile should pick the 1st percentile."""
        # 100 records: sorted values [1000, 2000, 2000, ..., 2000, 8000, 8000, ...]
        records = [(1000,)] * 1 + [(2000,)] * 49 + [(8000,)] * 50
        db = _mock_db_with_context_records(records)
        # idx = max(0, int(100 * 0.01)) = 1 → sorted[1] = 2000
        result = check_context_below_history(db, local_context=1500)
        assert result is not None
        assert "1,500" in result["message"]

    def test_mixed_context_values_no_warning_above_percentile(self):
        """No warning when above the 1st percentile threshold."""
        records = [(1000,)] * 1 + [(2000,)] * 49 + [(8000,)] * 50
        db = _mock_db_with_context_records(records)
        result = check_context_below_history(db, local_context=2000)
        assert result is None

    def test_min_safe_context_in_message(self):
        db = _mock_db_with_context_records([(16384,)] * 10)
        result = check_context_below_history(db, local_context=4096)
        assert "16,384" in result["message"]

    def test_dismiss_key(self):
        db = _mock_db_with_context_records([(8192,)] * 10)
        result = check_context_below_history(db, local_context=2048)
        assert result["dismissKey"] == "app.warnings.dismiss_context_reduced"


class TestCheckContextTruncationHistory:
    """Tests for check_context_truncation_history."""

    def test_warns_when_truncation_exists(self):
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = 3
        result = check_context_truncation_history(db, local_context=4096)
        assert result is not None
        assert result["type"] == "context_truncation_history"
        assert "3 time(s)" in result["message"]

    def test_no_warning_zero_truncations(self):
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = 0
        result = check_context_truncation_history(db, local_context=4096)
        assert result is None

    def test_no_warning_none_truncations(self):
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = None
        result = check_context_truncation_history(db, local_context=4096)
        assert result is None

    def test_single_truncation_message(self):
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = 1
        result = check_context_truncation_history(db, local_context=4096)
        assert "1 time(s)" in result["message"]

    def test_large_truncation_count(self):
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = 42
        result = check_context_truncation_history(db, local_context=4096)
        assert "42 time(s)" in result["message"]

    def test_warning_dict_has_all_required_keys(self):
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = 1
        result = check_context_truncation_history(db, local_context=4096)
        required = {
            "type",
            "icon",
            "title",
            "message",
            "dismissKey",
        }
        assert required.issubset(set(result.keys()))

    def test_warning_dict_has_action_link(self):
        """Truncation warnings carry actionUrl/actionLabel pointing to the diagnostic page."""
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = 1
        result = check_context_truncation_history(db, local_context=4096)
        assert result["actionUrl"] == "/metrics/context-overflow"
        assert result["actionLabel"]

    def test_dismiss_key(self):
        db = Mock()
        db.query.return_value.filter.return_value.filter.return_value.scalar.return_value = 1
        result = check_context_truncation_history(db, local_context=4096)
        assert result["dismissKey"] == "app.warnings.dismiss_context_reduced"