""" Tests for TTL-cached document scheduler settings retrieval in BackgroundJobScheduler. This module tests PR #1411 functionality: - DocumentSchedulerSettings frozen dataclass - TTL caching via _get_document_scheduler_settings() - Cache invalidation methods - Thread-safety of cache operations - Integration with unregister_user and _reload_config """ import threading from dataclasses import FrozenInstanceError from datetime import datetime, UTC from unittest.mock import Mock, MagicMock, patch import pytest @pytest.fixture def mock_background_scheduler(): """Mock BackgroundScheduler for all tests.""" with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_cls: mock_instance = MagicMock() mock_cls.return_value = mock_instance yield mock_instance @pytest.fixture def scheduler_with_cache(mock_background_scheduler): """Create a fresh scheduler instance with TTL cache.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None instance = BackgroundJobScheduler() return instance class TestDocumentSchedulerSettings: """Tests for the DocumentSchedulerSettings frozen dataclass.""" def test_default_values(self): """Dataclass has correct default values.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings() assert settings.enabled is True assert settings.interval_seconds == 1800 assert settings.download_pdfs is False assert settings.extract_text is True assert settings.generate_rag is False assert settings.last_run == "" def test_custom_values(self): """Dataclass accepts custom values.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings( enabled=False, interval_seconds=3600, download_pdfs=True, extract_text=False, generate_rag=True, last_run="2025-01-15T10:30:00+00:00", ) assert settings.enabled is False assert settings.interval_seconds == 3600 assert settings.download_pdfs is True assert settings.extract_text is False assert settings.generate_rag is True assert settings.last_run == "2025-01-15T10:30:00+00:00" def test_frozen_immutability(self): """Verify frozen dataclass raises FrozenInstanceError on mutation.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings() with pytest.raises(FrozenInstanceError): settings.enabled = False with pytest.raises(FrozenInstanceError): settings.interval_seconds = 900 def test_defaults_classmethod(self): """defaults() classmethod returns instance with default values.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) defaults = DocumentSchedulerSettings.defaults() assert defaults.enabled is True assert defaults.interval_seconds == 1800 assert defaults.download_pdfs is False assert defaults.extract_text is True assert defaults.generate_rag is False assert defaults.last_run == "" def test_partial_values_with_defaults(self): """Dataclass fills unspecified fields with defaults.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings( enabled=False, download_pdfs=True, ) assert settings.enabled is False assert settings.interval_seconds == 1800 # default assert settings.download_pdfs is True assert settings.extract_text is True # default assert settings.generate_rag is False # default assert settings.last_run == "" # default class TestGetDocumentSchedulerSettingsCaching: """Tests for _get_document_scheduler_settings caching behavior.""" def test_cache_miss_fetches_from_db(self, scheduler_with_cache): """First call fetches settings from database.""" scheduler = scheduler_with_cache scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "testpass") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: mock_session = MagicMock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_db.return_value = mock_session with patch( "local_deep_research.settings.manager.SettingsManager" ) as mock_sm_cls: mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda key, default: { "document_scheduler.enabled": True, "document_scheduler.interval_seconds": 900, "document_scheduler.download_pdfs": True, "document_scheduler.extract_text": False, "document_scheduler.generate_rag": True, "document_scheduler.last_run": "2025-01-15T12:00:00", }.get(key, default) mock_sm_cls.return_value = mock_sm settings = scheduler._get_document_scheduler_settings( "testuser" ) # Verify settings were fetched from DB assert settings.enabled is True assert settings.interval_seconds == 900 assert settings.download_pdfs is True assert settings.extract_text is False assert settings.generate_rag is True assert settings.last_run == "2025-01-15T12:00:00" # Verify DB was called mock_db.assert_called_once_with("testuser", "testpass") def test_cache_hit_returns_cached(self, scheduler_with_cache): """Second call returns cached settings without DB access.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "testpass") # Pre-populate cache cached_settings = DocumentSchedulerSettings( enabled=False, interval_seconds=1200, download_pdfs=True, extract_text=True, generate_rag=True, last_run="cached-value", ) scheduler._settings_cache["testuser"] = cached_settings with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: settings = scheduler._get_document_scheduler_settings("testuser") # Should return cached settings assert settings is cached_settings assert settings.last_run == "cached-value" # DB should NOT be called mock_db.assert_not_called() def test_force_refresh_bypasses_cache(self, scheduler_with_cache): """force_refresh=True bypasses cache and fetches from DB.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "testpass") # Pre-populate cache cached_settings = DocumentSchedulerSettings( enabled=True, interval_seconds=1800, last_run="old-cached-value", ) scheduler._settings_cache["testuser"] = cached_settings with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: mock_session = MagicMock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_db.return_value = mock_session with patch( "local_deep_research.settings.manager.SettingsManager" ) as mock_sm_cls: mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda key, default: { "document_scheduler.last_run": "fresh-db-value", }.get(key, default) mock_sm_cls.return_value = mock_sm settings = scheduler._get_document_scheduler_settings( "testuser", force_refresh=True ) # Should return fresh settings from DB assert settings.last_run == "fresh-db-value" # DB should be called despite cache being populated mock_db.assert_called_once() def test_returns_defaults_when_no_session(self, scheduler_with_cache): """Returns defaults when user has no session.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache # No user session set up settings = scheduler._get_document_scheduler_settings("nonexistent") # Should return defaults expected_defaults = DocumentSchedulerSettings.defaults() assert settings.enabled == expected_defaults.enabled assert settings.interval_seconds == expected_defaults.interval_seconds assert settings.download_pdfs == expected_defaults.download_pdfs def test_returns_defaults_on_db_exception(self, scheduler_with_cache): """Returns defaults when database access fails.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "testpass") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: mock_db.side_effect = Exception("Database connection failed") settings = scheduler._get_document_scheduler_settings("testuser") # Should return defaults, not raise expected_defaults = DocumentSchedulerSettings.defaults() assert settings.enabled == expected_defaults.enabled assert ( settings.interval_seconds == expected_defaults.interval_seconds ) def test_cache_stores_settings_after_fetch(self, scheduler_with_cache): """Verify settings are stored in cache after database fetch.""" scheduler = scheduler_with_cache scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "testpass") # Cache should be empty initially assert "testuser" not in scheduler._settings_cache with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db: mock_session = MagicMock() mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) mock_db.return_value = mock_session with patch( "local_deep_research.settings.manager.SettingsManager" ) as mock_sm_cls: mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda key, default: { "document_scheduler.interval_seconds": 999, }.get(key, default) mock_sm_cls.return_value = mock_sm # Fetch settings settings = scheduler._get_document_scheduler_settings( "testuser" ) # Verify settings were stored in cache assert "testuser" in scheduler._settings_cache cached = scheduler._settings_cache["testuser"] assert cached.interval_seconds == 999 assert cached is settings # Same object class TestCacheInvalidation: """Tests for cache invalidation methods.""" def test_invalidate_user_removes_entry(self, scheduler_with_cache): """invalidate_user_settings_cache removes specific user entry.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler._settings_cache["user1"] = DocumentSchedulerSettings() scheduler._settings_cache["user2"] = DocumentSchedulerSettings() result = scheduler.invalidate_user_settings_cache("user1") assert result is True assert "user1" not in scheduler._settings_cache assert "user2" in scheduler._settings_cache def test_invalidate_user_returns_false_if_not_found( self, scheduler_with_cache ): """invalidate_user_settings_cache returns False for non-existent user.""" scheduler = scheduler_with_cache result = scheduler.invalidate_user_settings_cache("nonexistent") assert result is False def test_invalidate_all_clears_cache(self, scheduler_with_cache): """invalidate_all_settings_cache clears entire cache.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler._settings_cache["user1"] = DocumentSchedulerSettings() scheduler._settings_cache["user2"] = DocumentSchedulerSettings() scheduler._settings_cache["user3"] = DocumentSchedulerSettings() scheduler.invalidate_all_settings_cache() assert len(scheduler._settings_cache) == 0 def test_invalidate_all_returns_count(self, scheduler_with_cache): """invalidate_all_settings_cache returns number of cleared entries.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler._settings_cache["user1"] = DocumentSchedulerSettings() scheduler._settings_cache["user2"] = DocumentSchedulerSettings() scheduler._settings_cache["user3"] = DocumentSchedulerSettings() count = scheduler.invalidate_all_settings_cache() assert count == 3 class TestCacheIntegration: """Tests for cache integration with other BackgroundJobScheduler methods.""" def test_unregister_user_invalidates_cache( self, scheduler_with_cache, mock_background_scheduler ): """unregister_user calls invalidate_user_settings_cache.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "testpass") scheduler._settings_cache["testuser"] = DocumentSchedulerSettings() scheduler.unregister_user("testuser") # Cache should be invalidated assert "testuser" not in scheduler._settings_cache # User session should also be removed assert "testuser" not in scheduler.user_sessions def test_reload_config_invalidates_all( self, scheduler_with_cache, mock_background_scheduler ): """_reload_config calls invalidate_all_settings_cache.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache scheduler._settings_cache["user1"] = DocumentSchedulerSettings() scheduler._settings_cache["user2"] = DocumentSchedulerSettings() # Set up settings_manager mock mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda key, default: default scheduler.settings_manager = mock_sm scheduler._reload_config() # All cache entries should be cleared assert len(scheduler._settings_cache) == 0 def test_cache_lock_exists(self, scheduler_with_cache): """Scheduler has _settings_cache_lock for thread safety.""" scheduler = scheduler_with_cache assert hasattr(scheduler, "_settings_cache_lock") assert isinstance( scheduler._settings_cache_lock, type(threading.Lock()) ) class TestCacheThreadSafety: """Tests for thread-safety of cache operations.""" def test_concurrent_reads_are_safe(self, scheduler_with_cache): """Multiple threads can read from cache safely.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache # Pre-populate cache for i in range(10): scheduler.user_sessions[f"user{i}"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store(f"user{i}", f"pass{i}") scheduler._settings_cache[f"user{i}"] = DocumentSchedulerSettings( interval_seconds=i * 100 ) results = [] errors = [] def read_settings(username): try: for _ in range(100): settings = scheduler._get_document_scheduler_settings( username ) results.append((username, settings.interval_seconds)) except Exception as e: errors.append(e) threads = [] for i in range(10): t = threading.Thread(target=read_settings, args=(f"user{i}",)) threads.append(t) t.start() for t in threads: t.join() # No errors should occur assert len(errors) == 0 # All reads should return consistent values assert len(results) == 1000 def test_concurrent_reads_and_invalidation_are_safe( self, scheduler_with_cache ): """Concurrent reads and invalidations don't cause race conditions.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache # Pre-populate sessions and cache for i in range(5): scheduler.user_sessions[f"user{i}"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store(f"user{i}", f"pass{i}") scheduler._settings_cache[f"user{i}"] = DocumentSchedulerSettings() errors = [] def read_settings(): try: for _ in range(50): for i in range(5): # May return cached or defaults if invalidated scheduler._get_document_scheduler_settings(f"user{i}") except Exception as e: errors.append(e) def invalidate_settings(): try: for _ in range(50): for i in range(5): scheduler.invalidate_user_settings_cache(f"user{i}") # Re-populate cache entry scheduler._settings_cache[f"user{i}"] = ( DocumentSchedulerSettings() ) except Exception as e: errors.append(e) threads = [] for _ in range(3): threads.append(threading.Thread(target=read_settings)) threads.append(threading.Thread(target=invalidate_settings)) for t in threads: t.start() for t in threads: t.join() # No errors should occur (no race conditions) assert len(errors) == 0 def test_concurrent_invalidate_all_is_safe(self, scheduler_with_cache): """Concurrent invalidate_all calls don't cause issues.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) scheduler = scheduler_with_cache errors = [] counts = [] def populate_and_clear(): try: for _ in range(50): # Populate for i in range(10): scheduler._settings_cache[f"user{i}"] = ( DocumentSchedulerSettings() ) # Clear count = scheduler.invalidate_all_settings_cache() counts.append(count) except Exception as e: errors.append(e) threads = [] for _ in range(4): threads.append(threading.Thread(target=populate_and_clear)) for t in threads: t.start() for t in threads: t.join() # No errors should occur assert len(errors) == 0 # Each invalidate_all should return a count (may be 0 if another cleared first) assert all(isinstance(c, int) and c >= 0 for c in counts)