"""Tests for metrics query_utils module.""" from sqlalchemy import Column, DateTime, String from sqlalchemy.orm import declarative_base from datetime import datetime, timedelta, UTC from local_deep_research.metrics.query_utils import ( get_period_cutoff, get_period_days, get_research_mode_condition, get_time_filter_condition, ) # Create a model for SQLAlchemy column testing Base = declarative_base() class QueryTestModel(Base): """Model for SQLAlchemy column comparisons.""" __tablename__ = "query_test_table" id = Column(String, primary_key=True) timestamp = Column(DateTime) mode = Column(String) class TestGetPeriodDays: """Tests for get_period_days — single period vocabulary (both UIs).""" def test_all_returns_none(self): assert get_period_days("all") is None def test_dashboard_vocabulary(self): # Main dashboard / context-overflow buttons send these. assert get_period_days("7d") == 7 assert get_period_days("30d") == 30 assert get_period_days("3m") == 90 assert get_period_days("1y") == 365 def test_link_analytics_vocabulary(self): # The standalone link-analytics page sends these aliases; they must # resolve to the same day counts as 3m/1y. assert get_period_days("90d") == 90 assert get_period_days("365d") == 365 def test_unknown_uses_default(self): assert get_period_days("nonsense") == 30 assert get_period_days("nonsense", default=7) == 7 class TestGetPeriodCutoff: """Tests for get_period_cutoff function.""" def test_returns_none_for_all(self): assert get_period_cutoff("all") is None def test_known_periods_map_to_days(self): now = datetime.now(UTC) for period, days in [ ("7d", 7), ("30d", 30), ("3m", 90), ("90d", 90), ("1y", 365), ("365d", 365), ]: cutoff = get_period_cutoff(period) expected = now - timedelta(days=days) assert abs((cutoff - expected).total_seconds()) < 5, period def test_unknown_period_defaults_to_30_days(self): cutoff = get_period_cutoff("month") # old vocabulary, no longer valid expected = datetime.now(UTC) - timedelta(days=30) assert abs((cutoff - expected).total_seconds()) < 5 class TestGetTimeFilterCondition: """Tests for get_time_filter_condition function.""" def test_returns_none_for_all_period(self): """Should return None when period is 'all'.""" result = get_time_filter_condition("all", QueryTestModel.timestamp) assert result is None def test_returns_condition_for_7_days(self): """Should return a BinaryExpression for 7 days period.""" result = get_time_filter_condition("7d", QueryTestModel.timestamp) assert result is not None # Should be a comparison expression assert hasattr(result, "left") or hasattr(result, "compare") def test_returns_condition_for_30_days(self): """Should return a condition for 30 days period.""" result = get_time_filter_condition("30d", QueryTestModel.timestamp) assert result is not None def test_returns_condition_for_3_months(self): """Should return a condition for 90 days (3 months).""" result = get_time_filter_condition("3m", QueryTestModel.timestamp) assert result is not None def test_returns_condition_for_1_year(self): """Should return a condition for 365 days (1 year).""" result = get_time_filter_condition("1y", QueryTestModel.timestamp) assert result is not None def test_defaults_to_30_days_for_unknown_period(self): """Should default to 30 days for unknown period strings.""" result = get_time_filter_condition("unknown", QueryTestModel.timestamp) # Should return condition (defaults to 30d) assert result is not None def test_defaults_to_30_days_for_empty_period(self): """Should default to 30 days for empty period string.""" result = get_time_filter_condition("", QueryTestModel.timestamp) assert result is not None def test_all_valid_periods(self): """All standard periods should return conditions.""" periods = ["7d", "30d", "3m", "1y"] for period in periods: result = get_time_filter_condition(period, QueryTestModel.timestamp) assert result is not None, ( f"Period {period} should return a condition" ) class TestGetResearchModeCondition: """Tests for get_research_mode_condition function.""" def test_returns_none_for_all_mode(self): """Should return None when research_mode is 'all'.""" result = get_research_mode_condition("all", QueryTestModel.mode) assert result is None def test_returns_condition_for_quick(self): """Should return condition for 'quick' mode.""" result = get_research_mode_condition("quick", QueryTestModel.mode) assert result is not None def test_returns_condition_for_detailed(self): """Should return condition for 'detailed' mode.""" result = get_research_mode_condition("detailed", QueryTestModel.mode) assert result is not None def test_returns_none_for_unknown_mode(self): """Should return None for unknown mode strings.""" result = get_research_mode_condition("unknown", QueryTestModel.mode) assert result is None def test_returns_none_for_empty_mode(self): """Should return None for empty mode string.""" result = get_research_mode_condition("", QueryTestModel.mode) assert result is None def test_quick_and_detailed_are_valid_modes(self): """Only 'quick' and 'detailed' should return conditions.""" # Valid modes assert ( get_research_mode_condition("quick", QueryTestModel.mode) is not None ) assert ( get_research_mode_condition("detailed", QueryTestModel.mode) is not None ) # Invalid modes assert get_research_mode_condition("fast", QueryTestModel.mode) is None assert get_research_mode_condition("slow", QueryTestModel.mode) is None