"""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"