""" Tests for news/subscription_manager/scheduler.py Tests cover: - BackgroundJobScheduler singleton pattern - Configuration loading - User session management - Scheduler lifecycle - update_user_info method - unregister_user method - _schedule_user_subscriptions method - _schedule_document_processing method - _get_document_scheduler_settings method - invalidate_user_settings_cache method - invalidate_all_settings_cache method - _check_subscription method - trigger_document_processing method - get_document_scheduler_status method """ import pytest from unittest.mock import Mock, patch, MagicMock import threading from datetime import datetime, timedelta, UTC from apscheduler.jobstores.base import JobLookupError class TestNewsSchedulerSingleton: """Tests for BackgroundJobScheduler singleton pattern.""" def test_news_scheduler_is_singleton(self): """BackgroundJobScheduler follows singleton pattern.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) # Reset singleton for test BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() scheduler1 = BackgroundJobScheduler() scheduler2 = BackgroundJobScheduler() assert scheduler1 is scheduler2 def test_scheduler_has_required_attributes(self): """BackgroundJobScheduler has required attributes after init.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) # Reset singleton for test BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() scheduler = BackgroundJobScheduler() assert hasattr(scheduler, "user_sessions") assert hasattr(scheduler, "lock") assert hasattr(scheduler, "scheduler") assert hasattr(scheduler, "config") assert hasattr(scheduler, "is_running") class TestSchedulerConfiguration: """Tests for scheduler configuration.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_default_config_values(self, scheduler): """Default configuration has expected values.""" config = scheduler.config assert config["enabled"] is True assert config["retention_hours"] == 48 assert config["cleanup_interval_hours"] == 1 assert config["max_jitter_seconds"] == 300 assert config["max_concurrent_jobs"] == 10 assert config["subscription_batch_size"] == 5 assert config["activity_check_interval_minutes"] == 5 def test_initialize_with_settings(self, scheduler): """Scheduler can be initialized with settings manager.""" mock_settings = Mock() mock_settings.get.return_value = None # Should not raise scheduler.initialize_with_settings(mock_settings) assert scheduler.settings_manager is mock_settings class TestSchedulerLifecycle: """Tests for scheduler start/stop lifecycle.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler_instance = MagicMock() mock_scheduler.return_value = mock_scheduler_instance instance = BackgroundJobScheduler() yield instance def test_scheduler_initial_state_not_running(self, scheduler): """Scheduler is not running initially.""" assert scheduler.is_running is False def test_user_sessions_initially_empty(self, scheduler): """User sessions dict is initially empty.""" assert scheduler.user_sessions == {} class TestUserSessionManagement: """Tests for user session tracking.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_lock_is_thread_lock(self, scheduler): """Scheduler has threading lock for thread safety.""" assert isinstance(scheduler.lock, type(threading.Lock())) class TestSchedulerAvailability: """Tests for scheduler availability flag.""" def test_scheduler_is_available(self): """Scheduler availability flag is True.""" from local_deep_research.scheduler.background import ( SCHEDULER_AVAILABLE, ) assert SCHEDULER_AVAILABLE is True class TestSchedulerStart: """Tests for scheduler start method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler_instance = MagicMock() mock_scheduler.return_value = mock_scheduler_instance instance = BackgroundJobScheduler() instance.set_app(MagicMock()) yield instance def test_start_sets_is_running(self, scheduler): """Starting scheduler sets is_running to True.""" scheduler.start() assert scheduler.is_running is True scheduler.scheduler.start.assert_called_once() def test_start_when_disabled(self, scheduler): """Scheduler doesn't start when disabled.""" scheduler.config["enabled"] = False scheduler.start() assert scheduler.is_running is False scheduler.scheduler.start.assert_not_called() def test_start_when_already_running(self, scheduler): """Scheduler warns when already running.""" scheduler.is_running = True scheduler.start() # Should not call start again scheduler.scheduler.start.assert_not_called() def test_start_adds_cleanup_job(self, scheduler): """Starting scheduler adds cleanup job.""" scheduler.start() # Check that add_job was called at least once assert scheduler.scheduler.add_job.called class TestSchedulerStop: """Tests for scheduler stop method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler_instance = MagicMock() mock_scheduler.return_value = mock_scheduler_instance instance = BackgroundJobScheduler() yield instance def test_stop_sets_is_running_false(self, scheduler): """Stopping scheduler sets is_running to False.""" scheduler.is_running = True scheduler.stop() assert scheduler.is_running is False def test_stop_when_not_running(self, scheduler): """Stopping scheduler when not running is safe.""" scheduler.is_running = False # Should not raise scheduler.stop() assert scheduler.is_running is False class TestGetSetting: """Tests for _get_setting method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_get_setting_with_settings_manager(self, scheduler): """_get_setting uses settings manager when available.""" mock_settings = Mock() mock_settings.get_setting.return_value = 100 scheduler.settings_manager = mock_settings result = scheduler._get_setting("some.key", 50) assert result == 100 mock_settings.get_setting.assert_called_once_with( "some.key", default=50 ) def test_get_setting_without_settings_manager(self, scheduler): """_get_setting returns default without settings manager.""" # No settings manager result = scheduler._get_setting("some.key", 50) assert result == 50 class TestSchedulerStatus: """Tests for scheduler status methods.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_get_status_when_not_running(self, scheduler): """Get status when scheduler is not running.""" scheduler.is_running = False if hasattr(scheduler, "get_status"): status = scheduler.get_status() assert ( status.get("running") is False or status.get("is_running") is False ) def test_get_status_when_running(self, scheduler): """Get status when scheduler is running.""" scheduler.is_running = True if hasattr(scheduler, "get_status"): status = scheduler.get_status() assert ( status.get("running") is True or status.get("is_running") is True ) class TestSchedulerRegisterUser: """Tests for user registration methods.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_register_user_adds_to_sessions(self, scheduler): """Registering user adds them to sessions dict.""" if hasattr(scheduler, "register_user_activity"): scheduler.register_user_activity("testuser", "password123") assert "testuser" in scheduler.user_sessions def test_register_user_updates_activity(self, scheduler): """Registering existing user updates last_activity.""" if hasattr(scheduler, "register_user_activity"): scheduler.register_user_activity("testuser", "password123") first_activity = scheduler.user_sessions["testuser"].get( "last_activity" ) # Register again import time time.sleep(0.1) scheduler.register_user_activity("testuser", "password123") second_activity = scheduler.user_sessions["testuser"].get( "last_activity" ) # Activity should be updated if first_activity and second_activity: assert second_activity >= first_activity class TestSchedulerUnregisterUser: """Tests for user unregistration.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_unregister_removes_user(self, scheduler): """Unregistering user removes them from sessions.""" # Set up proper session structure scheduler.user_sessions["testuser"] = { "scheduled_jobs": [], "last_activity": None, } scheduler._credential_store.store("testuser", "test") if hasattr(scheduler, "unregister_user"): scheduler.unregister_user("testuser") assert "testuser" not in scheduler.user_sessions def test_unregister_nonexistent_user(self, scheduler): """Unregistering non-existent user is safe.""" if hasattr(scheduler, "unregister_user"): # Should not raise scheduler.unregister_user("nonexistent") class TestScheduleUserSubscriptions: """Tests for _schedule_user_subscriptions method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_schedule_user_subscriptions_uses_jitter(self, scheduler): """_schedule_user_subscriptions applies random jitter.""" # Verify the scheduler has max_jitter_seconds config assert "max_jitter_seconds" in scheduler.config assert scheduler.config["max_jitter_seconds"] == 300 def test_schedule_user_subscriptions_respects_batch_size(self, scheduler): """_schedule_user_subscriptions respects subscription_batch_size.""" assert "subscription_batch_size" in scheduler.config assert scheduler.config["subscription_batch_size"] == 5 def test_schedule_user_subscriptions_jitter_calculation(self, scheduler): """Jitter is calculated based on max_jitter_seconds.""" import random random.seed(42) # Make deterministic for test max_jitter = scheduler.config["max_jitter_seconds"] # Generate some jitter values jitters = [random.randint(0, max_jitter) for _ in range(10)] # All values should be within range assert all(0 <= j <= max_jitter for j in jitters) def test_schedule_user_subscriptions_schedules_jobs(self, scheduler): """_schedule_user_subscriptions adds jobs to the scheduler.""" if hasattr(scheduler, "_schedule_user_subscriptions"): # Method exists assert callable(scheduler._schedule_user_subscriptions) class TestProcessUserDocuments: """Tests for _process_user_documents method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_process_user_documents_batch_processing(self, scheduler): """_process_user_documents processes in batches.""" # Verify batch size config exists assert "subscription_batch_size" in scheduler.config def test_process_user_documents_max_concurrent(self, scheduler): """_process_user_documents respects max_concurrent_jobs.""" assert "max_concurrent_jobs" in scheduler.config assert scheduler.config["max_concurrent_jobs"] == 10 class TestStoreResearchResult: """Tests for _store_research_result method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_store_research_result_serialization(self, scheduler): """Research results are properly serialized.""" # The scheduler should have retention_hours configured assert "retention_hours" in scheduler.config assert scheduler.config["retention_hours"] == 48 class TestCleanupOldResults: """Tests for cleanup functionality.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_cleanup_interval_configured(self, scheduler): """Cleanup interval is properly configured.""" assert "cleanup_interval_hours" in scheduler.config assert scheduler.config["cleanup_interval_hours"] == 1 def test_retention_hours_configured(self, scheduler): """Retention hours is properly configured.""" assert "retention_hours" in scheduler.config assert scheduler.config["retention_hours"] == 48 class TestActivityTracking: """Tests for user activity tracking.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_activity_check_interval_configured(self, scheduler): """Activity check interval is properly configured.""" assert "activity_check_interval_minutes" in scheduler.config assert scheduler.config["activity_check_interval_minutes"] == 5 def test_inactive_user_detection(self, scheduler): """Inactive users can be detected.""" from datetime import datetime, timedelta, UTC if hasattr(scheduler, "user_sessions"): # Set up a user session with old activity old_activity = datetime.now(UTC) - timedelta(hours=1) scheduler.user_sessions["old_user"] = { "scheduled_jobs": [], "last_activity": old_activity, } scheduler._credential_store.store("old_user", "test") # The user session should be in the dict assert "old_user" in scheduler.user_sessions class TestSchedulerExceptionHandling: """Tests for scheduler exception handling.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_scheduler_handles_job_exceptions(self, scheduler): """Scheduler handles exceptions in job execution.""" # The scheduler should have proper error handling assert scheduler.scheduler is not None def test_scheduler_recovers_from_errors(self, scheduler): """Scheduler can recover from errors.""" scheduler.is_running = True # Stopping should work even after errors scheduler.stop() assert scheduler.is_running is False # ============================================================================= # Phase 2 Tests: Comprehensive testing of critical scheduler methods # ============================================================================= class TestUpdateUserInfo: """Tests for update_user_info method - CRITICAL for user session management.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_creates_new_session_for_new_user(self, scheduler): """New user gets session entry created.""" scheduler.is_running = True with patch.object( scheduler, "_schedule_user_subscriptions" ) as mock_schedule: scheduler.update_user_info("newuser", "password123") assert "newuser" in scheduler.user_sessions mock_schedule.assert_called_once_with("newuser") def test_stores_password_in_credential_store(self, scheduler): """Password is stored in credential store, not in session dict.""" scheduler.is_running = True with patch.object(scheduler, "_schedule_user_subscriptions"): scheduler.update_user_info("testuser", "mySecretPassword") assert "password" not in scheduler.user_sessions["testuser"] assert ( scheduler._credential_store.retrieve("testuser") == "mySecretPassword" ) def test_sets_last_activity_time(self, scheduler): """Activity timestamp is set on new user.""" scheduler.is_running = True with patch.object(scheduler, "_schedule_user_subscriptions"): before = datetime.now(UTC) scheduler.update_user_info("testuser", "password") after = datetime.now(UTC) last_activity = scheduler.user_sessions["testuser"]["last_activity"] assert before <= last_activity <= after def test_initializes_empty_scheduled_jobs_set(self, scheduler): """Jobs set is empty initially for new user.""" scheduler.is_running = True with patch.object(scheduler, "_schedule_user_subscriptions"): scheduler.update_user_info("testuser", "password") assert ( scheduler.user_sessions["testuser"]["scheduled_jobs"] == set() ) def test_updates_existing_user_password(self, scheduler): """Password update works for existing user.""" scheduler.is_running = True scheduler.user_sessions["existinguser"] = { "last_activity": datetime.now(UTC) - timedelta(hours=1), "scheduled_jobs": {"job1"}, } scheduler._credential_store.store("existinguser", "oldpassword") with patch.object(scheduler, "_schedule_user_subscriptions"): scheduler.update_user_info("existinguser", "newpassword") assert ( scheduler._credential_store.retrieve("existinguser") == "newpassword" ) # Jobs should be preserved assert ( "job1" in scheduler.user_sessions["existinguser"]["scheduled_jobs"] ) def test_updates_last_activity_for_existing_user(self, scheduler): """Activity timestamp is updated for existing user.""" scheduler.is_running = True old_time = datetime.now(UTC) - timedelta(hours=1) scheduler.user_sessions["existinguser"] = { "last_activity": old_time, "scheduled_jobs": set(), } scheduler._credential_store.store("existinguser", "password") with patch.object(scheduler, "_schedule_user_subscriptions"): scheduler.update_user_info("existinguser", "password") new_activity = scheduler.user_sessions["existinguser"][ "last_activity" ] assert new_activity > old_time def test_does_nothing_when_scheduler_not_running(self, scheduler): """Graceful return when scheduler is not running.""" scheduler.is_running = False scheduler.update_user_info("testuser", "password") assert "testuser" not in scheduler.user_sessions def test_calls_schedule_user_subscriptions(self, scheduler): """Triggers subscription scheduling after user info update.""" scheduler.is_running = True with patch.object( scheduler, "_schedule_user_subscriptions" ) as mock_schedule: scheduler.update_user_info("user1", "pass1") scheduler.update_user_info("user1", "pass2") # Update existing assert mock_schedule.call_count == 2 mock_schedule.assert_called_with("user1") class TestUnregisterUserComprehensive: """Comprehensive tests for unregister_user method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_removes_user_from_sessions(self, scheduler): """Session is deleted when user unregisters.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") scheduler.unregister_user("testuser") assert "testuser" not in scheduler.user_sessions def test_removes_all_scheduled_jobs(self, scheduler): """All jobs are cleaned up when user unregisters.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": {"job1", "job2", "job3"}, "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") scheduler.unregister_user("testuser") # Verify remove_job was called for each job assert scheduler.scheduler.remove_job.call_count == 3 scheduler.scheduler.remove_job.assert_any_call("job1") scheduler.scheduler.remove_job.assert_any_call("job2") scheduler.scheduler.remove_job.assert_any_call("job3") def test_handles_job_lookup_error(self, scheduler): """Graceful handling when job not found during removal.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": {"missing_job"}, "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") scheduler.scheduler.remove_job.side_effect = JobLookupError( "missing_job" ) # Should not raise scheduler.unregister_user("testuser") assert "testuser" not in scheduler.user_sessions def test_invalidates_settings_cache(self, scheduler): """Cache is cleared when user unregisters.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") # Add to cache scheduler._settings_cache["testuser"] = MagicMock() scheduler.unregister_user("testuser") assert "testuser" not in scheduler._settings_cache def test_handles_nonexistent_user(self, scheduler): """No error for unknown user.""" # Should not raise scheduler.unregister_user("nonexistent") assert scheduler.scheduler.remove_job.call_count == 0 def test_thread_safe_removal(self, scheduler): """Lock is used properly during removal.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") # Track if lock was acquired original_lock = scheduler.lock lock_acquired = [] class TrackingLock: def __enter__(self): lock_acquired.append(True) return original_lock.__enter__() def __exit__(self, *args): return original_lock.__exit__(*args) scheduler.lock = TrackingLock() scheduler.unregister_user("testuser") assert len(lock_acquired) == 1 class TestScheduleUserSubscriptionsComprehensive: """Comprehensive tests for _schedule_user_subscriptions - CRITICAL scheduling logic.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_returns_early_if_no_session(self, scheduler): """Graceful return without session info.""" # No session set up with patch.object( scheduler, "_schedule_document_processing" ) as mock_doc: scheduler._schedule_user_subscriptions("nonexistent") # Should not crash, document processing should not be called mock_doc.assert_not_called() def test_queries_active_subscriptions(self, scheduler): """Database query filters for active subscriptions.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") # status is the source of truth for "active" (see # NewsSubscription.active_filter); the query now uses .filter() # with that predicate rather than filter_by(is_active=True). # Compare the actual SQLAlchemy expression so the test pins the # predicate itself, not merely that .filter() was invoked. from local_deep_research.database.models.news import ( NewsSubscription, ) (predicate,) = mock_query.filter.call_args.args assert predicate.compare(NewsSubscription.active_filter()) def test_clears_old_jobs_before_scheduling(self, scheduler): """Old jobs are removed before scheduling new ones.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": {"old_job_1", "old_job_2"}, "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_db.query.return_value.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") # Old jobs should have been removed assert scheduler.scheduler.remove_job.call_count >= 2 def test_calculates_jitter_within_bounds(self, scheduler): """Jitter respects max_jitter_seconds config.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") scheduler.config["max_jitter_seconds"] = 100 mock_db = MagicMock() # Create mock subscription mock_sub = MagicMock() mock_sub.id = 1 mock_sub.refresh_interval_minutes = 30 mock_sub.name = "Test Sub" mock_sub.query_or_topic = "test query" mock_db.query.return_value.filter.return_value.all.return_value = [ mock_sub ] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch( "local_deep_research.scheduler.background.random" ) as mock_random: mock_random.randint.return_value = 50 with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") # Verify randint was called with correct bounds mock_random.randint.assert_called_with(0, 100) def test_uses_interval_trigger_for_hourly(self, scheduler): """<=60 min refresh interval uses interval trigger.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.id = 1 mock_sub.refresh_interval_minutes = 60 # Exactly 60 minutes mock_sub.name = "Hourly Sub" mock_sub.query_or_topic = "hourly query" mock_db.query.return_value.filter.return_value.all.return_value = [ mock_sub ] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") # Should use interval trigger call_args = scheduler.scheduler.add_job.call_args assert call_args.kwargs.get("trigger") == "interval" def test_uses_date_trigger_for_infrequent(self, scheduler): """>60 min refresh interval uses date trigger.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.id = 1 mock_sub.refresh_interval_minutes = 120 # 2 hours mock_sub.name = "Infrequent Sub" mock_sub.query_or_topic = "infrequent query" mock_sub.next_refresh = None # No previous refresh mock_db.query.return_value.filter.return_value.all.return_value = [ mock_sub ] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") call_args = scheduler.scheduler.add_job.call_args assert call_args.kwargs.get("trigger") == "date" def test_schedules_subscription_with_no_next_refresh(self, scheduler): """Subscription with no next_refresh gets scheduled in future.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.id = 1 mock_sub.refresh_interval_minutes = 120 # > 60 min uses date trigger mock_sub.name = "No Refresh Sub" mock_sub.query_or_topic = "query" mock_sub.next_refresh = None # No previous refresh time mock_db.query.return_value.filter.return_value.all.return_value = [ mock_sub ] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") # Should schedule with date trigger call_args = scheduler.scheduler.add_job.call_args assert call_args is not None, "add_job was not called" assert call_args.kwargs.get("trigger") == "date" # run_date should be in the future (refresh_minutes + jitter) run_date = call_args.kwargs.get("run_date") assert run_date is not None time_diff = (run_date - datetime.now(UTC)).total_seconds() # Should be approximately 120 minutes (7200 seconds) + jitter assert ( 7100 <= time_diff <= 7200 + scheduler.config["max_jitter_seconds"] + 1 ) def test_adds_job_to_scheduler(self, scheduler): """Job is registered with scheduler.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.id = 42 mock_sub.refresh_interval_minutes = 30 mock_sub.name = "Test Sub" mock_sub.query_or_topic = "test query" mock_db.query.return_value.filter.return_value.all.return_value = [ mock_sub ] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") # Verify add_job was called with correct ID scheduler.scheduler.add_job.assert_called() call_kwargs = scheduler.scheduler.add_job.call_args.kwargs assert call_kwargs["id"] == "testuser_42" def test_tracks_job_in_session_jobs_set(self, scheduler): """Job ID is added to user's scheduled_jobs set.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.id = 99 mock_sub.refresh_interval_minutes = 30 mock_sub.name = "Tracked Sub" mock_sub.query_or_topic = "tracked query" mock_db.query.return_value.filter.return_value.all.return_value = [ mock_sub ] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") assert ( "testuser_99" in scheduler.user_sessions["testuser"]["scheduled_jobs"] ) def test_handles_database_error(self, scheduler): """Exception during database access is handled.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.side_effect = Exception( "Database connection failed" ) # Should not raise with patch.object(scheduler, "_schedule_document_processing"): scheduler._schedule_user_subscriptions("testuser") class TestScheduleDocumentProcessing: """Tests for _schedule_document_processing method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_returns_early_if_no_session(self, scheduler): """Graceful return without session info.""" # No session scheduler._schedule_document_processing("nonexistent") # Should not add any jobs scheduler.scheduler.add_job.assert_not_called() def test_skips_if_disabled_in_settings(self, scheduler): """Respects enabled=False setting.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) disabled_settings = DocumentSchedulerSettings(enabled=False) with patch.object( scheduler, "_get_document_scheduler_settings", return_value=disabled_settings, ): scheduler._schedule_document_processing("testuser") # Should not add any jobs scheduler.scheduler.add_job.assert_not_called() def test_removes_existing_document_job(self, scheduler): """Old document job is cleaned up.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": {"testuser_document_processing"}, "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings( enabled=True, interval_seconds=1800 ) scheduler.scheduler.get_job.return_value = MagicMock() with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): scheduler._schedule_document_processing("testuser") # The document-processing job is torn down before being re-added. # (_schedule_document_processing also tears down the opt-in # library-sweep job, so use assert_any_call rather than asserting the # last call.) scheduler.scheduler.remove_job.assert_any_call( "testuser_document_processing" ) def test_creates_job_with_correct_interval(self, scheduler): """Interval from settings is used.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings( enabled=True, interval_seconds=3600 ) scheduler.scheduler.get_job.return_value = MagicMock() scheduler.scheduler.remove_job.side_effect = JobLookupError("not found") with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): scheduler._schedule_document_processing("testuser") call_kwargs = scheduler.scheduler.add_job.call_args.kwargs assert call_kwargs["seconds"] == 3600 def test_job_has_jitter(self, scheduler): """Jitter is applied to prevent simultaneous processing.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings(enabled=True) scheduler.scheduler.get_job.return_value = MagicMock() with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): scheduler._schedule_document_processing("testuser") call_kwargs = scheduler.scheduler.add_job.call_args.kwargs assert call_kwargs.get("jitter") == 30 def test_job_has_max_instances_1(self, scheduler): """Prevents overlapping document processing.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings(enabled=True) scheduler.scheduler.get_job.return_value = MagicMock() with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): scheduler._schedule_document_processing("testuser") call_kwargs = scheduler.scheduler.add_job.call_args.kwargs assert call_kwargs.get("max_instances") == 1 def test_verifies_job_was_added(self, scheduler): """Job verification check occurs.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings(enabled=True) mock_job = MagicMock() mock_job.next_run_time = datetime.now(UTC) scheduler.scheduler.get_job.return_value = mock_job with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): scheduler._schedule_document_processing("testuser") # Verify get_job was called to verify the job exists scheduler.scheduler.get_job.assert_called_with( "testuser_document_processing" ) def test_handles_job_lookup_error_on_remove(self, scheduler): """Graceful handling when existing job not found.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings(enabled=True) scheduler.scheduler.remove_job.side_effect = JobLookupError("not found") scheduler.scheduler.get_job.return_value = MagicMock() with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): # Should not raise scheduler._schedule_document_processing("testuser") class TestGetDocumentSchedulerSettings: """Tests for _get_document_scheduler_settings method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_returns_cached_settings(self, scheduler): """Cache hit returns cached value.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) cached_settings = DocumentSchedulerSettings( enabled=True, interval_seconds=999, download_pdfs=True ) scheduler._settings_cache["cacheduser"] = cached_settings result = scheduler._get_document_scheduler_settings("cacheduser") assert result is cached_settings assert result.interval_seconds == 999 def test_fetches_from_db_on_cache_miss(self, scheduler): """DB query on cache miss.""" scheduler.user_sessions["dbuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("dbuser", "password") mock_db = MagicMock() mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda key, default: { "document_scheduler.enabled": True, "document_scheduler.interval_seconds": 2400, "document_scheduler.download_pdfs": True, "document_scheduler.extract_text": False, "document_scheduler.generate_rag": True, "document_scheduler.last_run": "2024-01-01T00:00:00", }.get(key, default) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch( "local_deep_research.settings.manager.SettingsManager", return_value=mock_sm, ): result = scheduler._get_document_scheduler_settings("dbuser") assert result.enabled is True assert result.interval_seconds == 2400 assert result.download_pdfs is True assert result.extract_text is False assert result.generate_rag is True def test_caches_fetched_settings(self, scheduler): """Result is cached after DB fetch.""" scheduler.user_sessions["cacheuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("cacheuser", "password") mock_db = MagicMock() mock_sm = MagicMock() mock_sm.get_setting.return_value = True with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch( "local_deep_research.settings.manager.SettingsManager", return_value=mock_sm, ): scheduler._get_document_scheduler_settings("cacheuser") assert "cacheuser" in scheduler._settings_cache def test_returns_defaults_on_no_session(self, scheduler): """Graceful return with defaults without session.""" result = scheduler._get_document_scheduler_settings("nosession") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) defaults = DocumentSchedulerSettings.defaults() assert result.enabled == defaults.enabled assert result.interval_seconds == defaults.interval_seconds def test_returns_defaults_on_db_error(self, scheduler): """Error handling returns defaults.""" scheduler.user_sessions["erroruser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("erroruser", "password") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.side_effect = Exception("DB connection failed") result = scheduler._get_document_scheduler_settings("erroruser") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) defaults = DocumentSchedulerSettings.defaults() assert result.enabled == defaults.enabled def test_force_refresh_bypasses_cache(self, scheduler): """Force refresh fetches from DB even with cached value.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) cached_settings = DocumentSchedulerSettings(interval_seconds=999) scheduler._settings_cache["refreshuser"] = cached_settings scheduler.user_sessions["refreshuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("refreshuser", "password") mock_db = MagicMock() mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda key, default: { "document_scheduler.interval_seconds": 5000, }.get(key, default) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch( "local_deep_research.settings.manager.SettingsManager", return_value=mock_sm, ): result = scheduler._get_document_scheduler_settings( "refreshuser", force_refresh=True ) # Should get fresh value, not cached assert result.interval_seconds == 5000 def test_settings_are_frozen_dataclass(self, scheduler): """Immutability for thread safety verified.""" from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings( enabled=True, interval_seconds=1800 ) # Frozen dataclass should raise FrozenInstanceError on modification with pytest.raises(Exception): # dataclasses.FrozenInstanceError settings.enabled = False class TestInvalidateUserSettingsCache: """Tests for invalidate_user_settings_cache method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_removes_user_from_cache(self, scheduler): """Entry is deleted from cache.""" scheduler._settings_cache["testuser"] = MagicMock() scheduler.invalidate_user_settings_cache("testuser") assert "testuser" not in scheduler._settings_cache def test_returns_true_if_found(self, scheduler): """True returned on successful removal.""" scheduler._settings_cache["existinguser"] = MagicMock() result = scheduler.invalidate_user_settings_cache("existinguser") assert result is True def test_returns_false_if_not_found(self, scheduler): """False returned when user not in cache.""" result = scheduler.invalidate_user_settings_cache("nonexistent") assert result is False def test_thread_safe(self, scheduler): """Uses lock for thread safety.""" scheduler._settings_cache["testuser"] = MagicMock() # Track if lock was acquired original_lock = scheduler._settings_cache_lock lock_acquired = [] class TrackingLock: def __enter__(self): lock_acquired.append(True) return original_lock.__enter__() def __exit__(self, *args): return original_lock.__exit__(*args) scheduler._settings_cache_lock = TrackingLock() scheduler.invalidate_user_settings_cache("testuser") assert len(lock_acquired) == 1 class TestInvalidateAllSettingsCache: """Tests for invalidate_all_settings_cache method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_clears_all_entries(self, scheduler): """Cache is emptied.""" scheduler._settings_cache["user1"] = MagicMock() scheduler._settings_cache["user2"] = MagicMock() scheduler._settings_cache["user3"] = MagicMock() scheduler.invalidate_all_settings_cache() assert len(scheduler._settings_cache) == 0 def test_returns_count_cleared(self, scheduler): """Returns correct count of cleared entries.""" scheduler._settings_cache["user1"] = MagicMock() scheduler._settings_cache["user2"] = MagicMock() result = scheduler.invalidate_all_settings_cache() assert result == 2 def test_thread_safe(self, scheduler): """Uses lock for thread safety.""" scheduler._settings_cache["testuser"] = MagicMock() original_lock = scheduler._settings_cache_lock lock_acquired = [] class TrackingLock: def __enter__(self): lock_acquired.append(True) return original_lock.__enter__() def __exit__(self, *args): return original_lock.__exit__(*args) scheduler._settings_cache_lock = TrackingLock() scheduler.invalidate_all_settings_cache() assert len(lock_acquired) == 1 class TestCheckSubscription: """Tests for _check_subscription method - CRITICAL subscription refresh logic.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_removes_job_if_no_session(self, scheduler): """Cleanup when user gone.""" # No session scheduler._check_subscription("goneuser", 123) scheduler.scheduler.remove_job.assert_called_with("goneuser_123") def test_handles_job_removal_error(self, scheduler): """Graceful handling when job removal fails.""" scheduler.scheduler.remove_job.side_effect = JobLookupError("not found") # Should not raise scheduler._check_subscription("goneuser", 123) def test_skips_inactive_subscription(self, scheduler): """Respects is_active=False.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.status = "paused" mock_db.query.return_value.get.return_value = mock_sub with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object( scheduler, "_trigger_subscription_research_sync" ) as mock_trigger: scheduler._check_subscription("testuser", 1) mock_trigger.assert_not_called() def test_replaces_date_placeholder(self, scheduler): """YYYY-MM-DD is replaced with actual date.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.status = "active" mock_sub.query_or_topic = "News from YYYY-MM-DD" mock_sub.id = 1 mock_sub.name = "Test" mock_sub.refresh_interval_minutes = 60 mock_sub.model_provider = "openai" mock_sub.model = "gpt-4" mock_sub.search_strategy = "news" mock_sub.search_engine = "google" mock_db.query.return_value.get.return_value = mock_sub # Get the job and set up mock scheduler.scheduler.get_job.return_value = None with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch( "local_deep_research.news.core.utils.get_local_date_string", return_value="2024-06-15", ): with patch.object( scheduler, "_trigger_subscription_research_sync" ) as mock_trigger: scheduler._check_subscription("testuser", 1) call_args = mock_trigger.call_args subscription_data = call_args[0][1] assert subscription_data["query"] == "News from 2024-06-15" assert ( subscription_data["original_query"] == "News from YYYY-MM-DD" ) def test_updates_last_refresh_time(self, scheduler): """Timestamp is updated after check.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.status = "active" mock_sub.query_or_topic = "Test query" mock_sub.id = 1 mock_sub.name = "Test" mock_sub.refresh_interval_minutes = 60 mock_sub.last_refresh = None mock_sub.next_refresh = None mock_db.query.return_value.get.return_value = mock_sub scheduler.scheduler.get_job.return_value = None with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_trigger_subscription_research_sync"): scheduler._check_subscription("testuser", 1) # last_refresh should have been updated assert mock_sub.last_refresh is not None mock_db.commit.assert_called() def test_calculates_next_refresh(self, scheduler): """Next run time is calculated.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.status = "active" mock_sub.query_or_topic = "Test query" mock_sub.id = 1 mock_sub.name = "Test" mock_sub.refresh_interval_minutes = 60 mock_db.query.return_value.get.return_value = mock_sub scheduler.scheduler.get_job.return_value = None before = datetime.now(UTC) with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_trigger_subscription_research_sync"): scheduler._check_subscription("testuser", 1) # next_refresh should be approximately 60 minutes from now assert mock_sub.next_refresh is not None time_diff = (mock_sub.next_refresh - before).total_seconds() # Should be approximately 60 minutes (3600 seconds) assert 3590 <= time_diff <= 3610 def test_triggers_research(self, scheduler): """Research API is called.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.status = "active" mock_sub.query_or_topic = "Test query" mock_sub.id = 1 mock_sub.name = "Test Sub" mock_sub.refresh_interval_minutes = 60 mock_sub.model_provider = "openai" mock_sub.model = "gpt-4" mock_sub.search_strategy = "news" mock_sub.search_engine = "google" mock_db.query.return_value.get.return_value = mock_sub scheduler.scheduler.get_job.return_value = None with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object( scheduler, "_trigger_subscription_research_sync" ) as mock_trigger: scheduler._check_subscription("testuser", 1) mock_trigger.assert_called_once() call_args = mock_trigger.call_args[0] assert call_args[0] == "testuser" assert call_args[1]["id"] == 1 def test_reschedules_for_date_trigger(self, scheduler): """Continues scheduling with date trigger.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.status = "active" mock_sub.query_or_topic = "Test query" mock_sub.id = 1 mock_sub.name = "Test" mock_sub.refresh_interval_minutes = 120 # > 60, so uses date trigger mock_db.query.return_value.get.return_value = mock_sub # Mock existing job with DateTrigger mock_job = MagicMock() mock_job.trigger.__class__.__name__ = "DateTrigger" scheduler.scheduler.get_job.return_value = mock_job with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) with patch.object(scheduler, "_trigger_subscription_research_sync"): scheduler._check_subscription("testuser", 1) # Should reschedule with add_job scheduler.scheduler.add_job.assert_called() call_kwargs = scheduler.scheduler.add_job.call_args.kwargs assert call_kwargs["trigger"] == "date" assert call_kwargs["id"] == "testuser_1" def test_handles_database_error(self, scheduler): """Exception is handled gracefully.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.side_effect = Exception("DB error") # Should not raise scheduler._check_subscription("testuser", 1) class TestTriggerDocumentProcessing: """Tests for trigger_document_processing method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_returns_false_if_no_session(self, scheduler): """Graceful return without session.""" scheduler.is_running = True result = scheduler.trigger_document_processing("nonexistent") assert result is False def test_returns_false_if_not_running(self, scheduler): """Graceful return when scheduler stopped.""" scheduler.is_running = False scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") result = scheduler.trigger_document_processing("testuser") assert result is False def test_schedules_immediate_job(self, scheduler): """Job is scheduled for 1 second from now.""" scheduler.is_running = True scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_job = MagicMock() mock_job.next_run_time = datetime.now(UTC) scheduler.scheduler.get_job.return_value = mock_job before = datetime.now(UTC) result = scheduler.trigger_document_processing("testuser") assert result is True call_kwargs = scheduler.scheduler.add_job.call_args.kwargs assert call_kwargs["trigger"] == "date" run_date = call_kwargs["run_date"] # Should be approximately 1 second from now time_diff = (run_date - before).total_seconds() assert 0 <= time_diff <= 2 def test_verifies_job_added(self, scheduler): """Verification check occurs.""" scheduler.is_running = True scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_job = MagicMock() scheduler.scheduler.get_job.return_value = mock_job scheduler.trigger_document_processing("testuser") scheduler.scheduler.get_job.assert_called_with( "testuser_document_processing_manual" ) def test_returns_false_if_job_not_added(self, scheduler): """Returns False when job verification fails.""" scheduler.is_running = True scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") scheduler.scheduler.get_job.return_value = None result = scheduler.trigger_document_processing("testuser") assert result is False def test_handles_exception(self, scheduler): """Error is handled gracefully.""" scheduler.is_running = True scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") scheduler.scheduler.add_job.side_effect = Exception("Scheduler error") result = scheduler.trigger_document_processing("testuser") assert result is False class TestGetDocumentSchedulerStatus: """Tests for get_document_scheduler_status method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_returns_disabled_for_unknown_user(self, scheduler): """Unknown user returns disabled status.""" result = scheduler.get_document_scheduler_status("unknown") assert result["enabled"] is False assert "message" in result def test_includes_all_processing_options(self, scheduler): """All processing flags are present.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings( enabled=True, download_pdfs=True, extract_text=False, generate_rag=True, ) with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): result = scheduler.get_document_scheduler_status("testuser") assert "processing_options" in result assert result["processing_options"]["download_pdfs"] is True assert result["processing_options"]["extract_text"] is False assert result["processing_options"]["generate_rag"] is True def test_shows_has_scheduled_job(self, scheduler): """Job tracking is correct.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": {"testuser_document_processing"}, "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings(enabled=True) with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): result = scheduler.get_document_scheduler_status("testuser") assert result["has_scheduled_job"] is True def test_shows_user_active_status(self, scheduler): """Active flag is correct.""" scheduler.user_sessions["activeuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("activeuser", "password") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings(enabled=True) with patch.object( scheduler, "_get_document_scheduler_settings", return_value=settings ): result = scheduler.get_document_scheduler_status("activeuser") assert result["user_active"] is True def test_handles_exception(self, scheduler): """Error returns safe dict.""" scheduler.user_sessions["erroruser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("erroruser", "password") with patch.object( scheduler, "_get_document_scheduler_settings", side_effect=Exception("Settings error"), ): result = scheduler.get_document_scheduler_status("erroruser") assert result["enabled"] is False assert "message" in result class TestCheckUserOverdueSubscriptions: """Tests for _check_user_overdue_subscriptions method.""" @pytest.fixture def scheduler(self): """Create a fresh scheduler instance.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) BackgroundJobScheduler._instance = None with patch( "local_deep_research.scheduler.background.BackgroundScheduler" ) as mock_scheduler: mock_scheduler.return_value = MagicMock() instance = BackgroundJobScheduler() yield instance def test_returns_early_if_no_session(self, scheduler): """Graceful return without session.""" # No session scheduler._check_user_overdue_subscriptions("nonexistent") # Should not attempt to query database scheduler.scheduler.add_job.assert_not_called() def test_finds_overdue_subscriptions(self, scheduler): """Queries for overdue subscriptions correctly.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value.all.return_value = [] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) scheduler._check_user_overdue_subscriptions("testuser") # Verify query was made mock_db.query.assert_called() def test_schedules_overdue_with_delay(self, scheduler): """Overdue subscriptions are scheduled with random delay.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") mock_db = MagicMock() mock_sub = MagicMock() mock_sub.id = 1 mock_sub.name = "Overdue Sub" mock_sub.query_or_topic = "overdue query" mock_query = MagicMock() mock_db.query.return_value = mock_query mock_query.filter.return_value.all.return_value = [mock_sub] with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.return_value.__enter__ = MagicMock( return_value=mock_db ) mock_db_session.return_value.__exit__ = MagicMock( return_value=False ) scheduler._check_user_overdue_subscriptions("testuser") # Should schedule job scheduler.scheduler.add_job.assert_called() call_kwargs = scheduler.scheduler.add_job.call_args.kwargs assert call_kwargs["trigger"] == "date" def test_handles_database_error(self, scheduler): """Exception is handled gracefully.""" scheduler.user_sessions["testuser"] = { "scheduled_jobs": set(), "last_activity": datetime.now(UTC), } scheduler._credential_store.store("testuser", "password") with patch( "local_deep_research.database.session_context.get_user_db_session" ) as mock_db_session: mock_db_session.side_effect = Exception("DB error") # Should not raise scheduler._check_user_overdue_subscriptions("testuser") class TestSchedulerEgressBackstop: """The document scheduler runs on an APScheduler worker thread with no egress context; _arm_egress_backstop must set one from the user's saved settings so DownloadService fetches get the audit-hook secondary net (R2-8). @thread_cleanup clears it on exit.""" def test_arm_egress_backstop_sets_context_from_settings(self): from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) from local_deep_research.security.egress.policy import EgressScope sched = BackgroundJobScheduler() fake_sm = MagicMock() fake_sm.get_settings_snapshot.return_value = { "policy.egress_scope": "private_only", "search.tool": "library", } fake_sm.get_setting.side_effect = lambda k, d=None: ( "library" if k == "search.tool" else d ) with patch( "local_deep_research.security.egress.audit_hook.set_active_context" ) as mock_set: sched._arm_egress_backstop(fake_sm, "alice") assert mock_set.call_count == 1 ctx = mock_set.call_args[0][0] assert ctx.scope == EgressScope.PRIVATE_ONLY def test_arm_egress_backstop_never_raises_on_bad_settings(self): from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) sched = BackgroundJobScheduler() fake_sm = MagicMock() fake_sm.get_settings_snapshot.side_effect = RuntimeError("no db") # Must swallow the error (best-effort backstop). sched._arm_egress_backstop(fake_sm, "alice")