""" Coverage tests for news/subscription_manager/scheduler.py Targets the ~132 missing statements (77% -> higher coverage). Focuses on: - DocumentSchedulerSettings dataclass paths - Settings cache (hit, miss, invalidation, error paths) - start/stop lifecycle - _schedule_user_subscriptions edge cases (naive datetime, no next_refresh, JobLookupError) - _schedule_document_processing (existing job removal, job verification failure) - _process_user_documents full flow (download_pdfs, extract_text, generate_rag paths) - _check_subscription with DateTrigger reschedule - _trigger_subscription_research_sync with/without search_engine - _store_research_result edge cases (sources formatting, headline fallbacks, make_serializable) - _reload_config with exception path - get_document_scheduler_status exception path - trigger_document_processing exception and job-verify-fail paths - initialize_with_settings (success + exception) - SettingsContext inner class """ import pytest from unittest.mock import Mock, MagicMock, patch from datetime import datetime, timedelta, UTC # Commonly used patch targets (local imports in scheduler.py) DB_SESSION = "local_deep_research.database.session_context.get_user_db_session" SETTINGS_MGR = "local_deep_research.settings.manager.SettingsManager" DOWNLOAD_SVC = "local_deep_research.research_library.services.download_service.DownloadService" IS_DOWNLOADABLE = ( "local_deep_research.research_library.utils.is_downloadable_url" ) QUICK_SUMMARY = "local_deep_research.api.research_functions.quick_summary" SET_CTX = "local_deep_research.config.thread_settings.set_settings_context" HEADLINE_GEN = ( "local_deep_research.news.utils.headline_generator.generate_headline" ) TOPIC_GEN = "local_deep_research.news.utils.topic_generator.generate_topics" REPORT_STORAGE = "local_deep_research.storage.get_report_storage" FORMAT_LINKS = ( "local_deep_research.utilities.search_utilities.format_links_to_markdown" ) CITATION_FMT = ( "local_deep_research.text_optimization.citation_formatter.CitationFormatter" ) CITATION_MODE = ( "local_deep_research.text_optimization.citation_formatter.CitationMode" ) GET_SETTING_SNAP = ( "local_deep_research.config.search_config.get_setting_from_snapshot" ) DATE_STRING = "local_deep_research.news.core.utils.get_local_date_string" # Top-level imports in scheduler.py. # NOTE: RAG indexing moved out of _process_user_documents into the unified # _reconcile_unindexed_documents reconciler (covered in tests/news/ # test_library_sweep.py), so this module no longer patches LibraryRAGService / # get_default_library_id. SCHED_MOD = "local_deep_research.scheduler.background" # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _reset_singleton(): """Reset BackgroundJobScheduler singleton before each test.""" from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) import local_deep_research.scheduler.background as mod BackgroundJobScheduler._instance = None mod._scheduler_instance = None yield BackgroundJobScheduler._instance = None mod._scheduler_instance = None @pytest.fixture def mock_bg(): """Mock BackgroundScheduler globally.""" with patch(f"{SCHED_MOD}.BackgroundScheduler") as cls: inst = MagicMock() cls.return_value = inst yield inst @pytest.fixture def sched(mock_bg): from local_deep_research.scheduler.background import ( BackgroundJobScheduler, ) instance = BackgroundJobScheduler() instance.set_app(MagicMock()) return instance @pytest.fixture def running(sched): sched.is_running = True return sched def _make_session(jobs=None): """Create a session dict (password stored separately in credential store).""" return { "last_activity": datetime.now(UTC), "scheduled_jobs": jobs if jobs is not None else set(), } def _setup_user(sched, username, password="pw", jobs=None): """Set up a user session with credentials in the credential store.""" sched.user_sessions[username] = _make_session(jobs=jobs) sched._credential_store.store(username, password) def _ctx(mock_session): """Make mock_session usable as a context manager.""" mock_session.__enter__ = Mock(return_value=mock_session) mock_session.__exit__ = Mock(return_value=False) return mock_session # --------------------------------------------------------------------------- # DocumentSchedulerSettings # --------------------------------------------------------------------------- class TestDocumentSchedulerSettings: def test_defaults_returns_instance(self): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) d = DocumentSchedulerSettings.defaults() assert d.enabled is True assert d.interval_seconds == 1800 assert d.download_pdfs is False assert d.extract_text is True assert d.generate_rag is False assert d.last_run == "" def test_frozen_immutable(self): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) d = DocumentSchedulerSettings() with pytest.raises(AttributeError): d.enabled = False def test_custom_values(self): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) d = DocumentSchedulerSettings( enabled=False, interval_seconds=900, download_pdfs=True, extract_text=False, generate_rag=True, last_run="2024-01-01", ) assert d.enabled is False assert d.interval_seconds == 900 # --------------------------------------------------------------------------- # initialize_with_settings # --------------------------------------------------------------------------- class TestInitializeWithSettings: def test_initialize_success(self, sched): sm = MagicMock() sm.get_setting.side_effect = lambda k, default: default sched.initialize_with_settings(sm) assert sched.settings_manager is sm def test_initialize_exception_keeps_defaults(self, sched): sm = MagicMock() sm.get_setting.side_effect = RuntimeError("boom") old_config = sched.config.copy() sched.initialize_with_settings(sm) assert sched.config == old_config # --------------------------------------------------------------------------- # _get_setting # --------------------------------------------------------------------------- class TestGetSetting: def test_with_settings_manager(self, sched): sm = MagicMock() sm.get_setting.return_value = 42 sched.settings_manager = sm assert sched._get_setting("some.key", 10) == 42 def test_without_settings_manager(self, sched): assert sched._get_setting("some.key", 10) == 10 def test_with_none_settings_manager(self, sched): sched.settings_manager = None assert sched._get_setting("k", 99) == 99 # --------------------------------------------------------------------------- # Settings cache # --------------------------------------------------------------------------- class TestSettingsCache: def test_cache_hit(self, sched): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings(enabled=False) sched._settings_cache["alice"] = settings result = sched._get_document_scheduler_settings("alice") assert result is settings def test_cache_miss_no_session_returns_defaults(self, sched): result = sched._get_document_scheduler_settings("nobody") assert result.enabled is True def test_cache_miss_fetches_from_db(self, sched): _setup_user(sched, "bob") mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda k, default: { "document_scheduler.enabled": True, "document_scheduler.interval_seconds": 600, "document_scheduler.download_pdfs": True, "document_scheduler.extract_text": False, "document_scheduler.generate_rag": True, "document_scheduler.last_run": "2024-06-01", }.get(k, default) mock_db = _ctx(MagicMock()) with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), ): result = sched._get_document_scheduler_settings("bob") assert result.interval_seconds == 600 assert "bob" in sched._settings_cache def test_cache_miss_db_error_returns_defaults(self, sched): _setup_user(sched, "carol") with patch(DB_SESSION, side_effect=RuntimeError("db down")): result = sched._get_document_scheduler_settings("carol") assert result.enabled is True def test_force_refresh_bypasses_cache(self, sched): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) sched._settings_cache["dave"] = DocumentSchedulerSettings(enabled=False) _setup_user(sched, "dave") mock_sm = MagicMock() mock_sm.get_setting.side_effect = lambda k, default: default mock_db = _ctx(MagicMock()) with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), ): result = sched._get_document_scheduler_settings( "dave", force_refresh=True ) assert result.enabled is True def test_invalidate_user_found(self, sched): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) sched._settings_cache["eve"] = DocumentSchedulerSettings() assert sched.invalidate_user_settings_cache("eve") is True def test_invalidate_user_not_found(self, sched): assert sched.invalidate_user_settings_cache("ghost") is False def test_invalidate_all(self, sched): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) sched._settings_cache["a"] = DocumentSchedulerSettings() sched._settings_cache["b"] = DocumentSchedulerSettings() count = sched.invalidate_all_settings_cache() assert count == 2 # --------------------------------------------------------------------------- # start / stop # --------------------------------------------------------------------------- class TestStartStop: def test_start_disabled(self, sched, mock_bg): sched.config["enabled"] = False sched.start() assert sched.is_running is False def test_start_already_running(self, running, mock_bg): mock_bg.start.reset_mock() running.start() mock_bg.start.assert_not_called() def test_start_success(self, sched, mock_bg): sched.start() assert sched.is_running is True mock_bg.start.assert_called_once() assert mock_bg.add_job.call_count == 3 def test_stop(self, running, mock_bg): _setup_user(running, "x") running.stop() assert running.is_running is False assert len(running.user_sessions) == 0 def test_stop_when_not_running(self, sched, mock_bg): sched.stop() mock_bg.shutdown.assert_not_called() # --------------------------------------------------------------------------- # _schedule_user_subscriptions edge cases # --------------------------------------------------------------------------- class TestScheduleUserSubscriptionsEdgeCases: def test_naive_next_refresh_assumed_utc(self, sched, mock_bg): _setup_user(sched, "u") sub = MagicMock() sub.id = 10 sub.name = "Naive" sub.query_or_topic = "query" sub.refresh_interval_minutes = 120 sub.next_refresh = datetime(2099, 1, 1) # naive, future mock_db = _ctx(MagicMock()) mock_db.query.return_value.filter.return_value.all.return_value = [sub] with ( patch(DB_SESSION, return_value=mock_db), patch.object(sched, "_schedule_document_processing"), ): sched._schedule_user_subscriptions("u") mock_bg.add_job.assert_called() def test_no_next_refresh_uses_calculated(self, sched, mock_bg): _setup_user(sched, "u") sub = MagicMock() sub.id = 20 sub.name = "No Next" sub.query_or_topic = "q" sub.refresh_interval_minutes = 120 sub.next_refresh = None mock_db = _ctx(MagicMock()) mock_db.query.return_value.filter.return_value.all.return_value = [sub] with ( patch(DB_SESSION, return_value=mock_db), patch.object(sched, "_schedule_document_processing"), ): sched._schedule_user_subscriptions("u") mock_bg.add_job.assert_called() def test_job_lookup_error_on_clear(self, sched, mock_bg): from apscheduler.jobstores.base import JobLookupError mock_bg.remove_job.side_effect = JobLookupError("x") _setup_user(sched, "u", jobs={"old_job"}) mock_db = _ctx(MagicMock()) mock_db.query.return_value.filter.return_value.all.return_value = [] with ( patch(DB_SESSION, return_value=mock_db), patch.object(sched, "_schedule_document_processing"), ): sched._schedule_user_subscriptions("u") def test_db_exception_handled(self, sched, mock_bg): _setup_user(sched, "u") with ( patch(DB_SESSION, side_effect=RuntimeError("db fail")), patch.object(sched, "_schedule_document_processing"), ): sched._schedule_user_subscriptions("u") def test_sub_name_none_uses_query(self, sched, mock_bg): _setup_user(sched, "u") sub = MagicMock() sub.id = 30 sub.name = None sub.query_or_topic = "a" * 50 sub.refresh_interval_minutes = 30 sub.next_refresh = None mock_db = _ctx(MagicMock()) mock_db.query.return_value.filter.return_value.all.return_value = [sub] with ( patch(DB_SESSION, return_value=mock_db), patch.object(sched, "_schedule_document_processing"), ): sched._schedule_user_subscriptions("u") call_kwargs = mock_bg.add_job.call_args[1] assert len(call_kwargs["name"]) <= 36 # --------------------------------------------------------------------------- # _schedule_document_processing edge cases # --------------------------------------------------------------------------- class TestScheduleDocumentProcessingEdgeCases: def test_existing_job_removed(self, sched, mock_bg): _setup_user(sched, "u", jobs={"u_document_processing"}) from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) with patch.object( sched, "_get_document_scheduler_settings", return_value=DocumentSchedulerSettings(enabled=True), ): mock_bg.get_job.return_value = MagicMock( next_run_time=datetime.now(UTC) ) sched._schedule_document_processing("u") # _schedule_document_processing also tears down the opt-in library # sweep job, so the document-processing removal may not be the last # remove_job call — assert it happened rather than that it was last. mock_bg.remove_job.assert_any_call("u_document_processing") def test_wires_up_reconciler(self, sched, mock_bg): """_schedule_document_processing must invoke _schedule_reconciler — the ONLY production path that gives the library-sweep job a lifecycle. Guards the integration seam: deleting the self._schedule_reconciler(...) call would silently disable the whole feature while every isolated reconciler test still passes. """ _setup_user(sched, "u") from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) settings = DocumentSchedulerSettings( enabled=True, sweep_library_collections=True ) with ( patch.object( sched, "_get_document_scheduler_settings", return_value=settings, ), patch.object(sched, "_schedule_reconciler") as mock_reconciler, ): sched._schedule_document_processing("u") mock_reconciler.assert_called_once() assert mock_reconciler.call_args[0][0] == "u" def test_existing_job_not_found_ok(self, sched, mock_bg): from apscheduler.jobstores.base import JobLookupError from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) _setup_user(sched, "u") mock_bg.remove_job.side_effect = JobLookupError("x") with patch.object( sched, "_get_document_scheduler_settings", return_value=DocumentSchedulerSettings(enabled=True), ): mock_bg.get_job.return_value = MagicMock( next_run_time=datetime.now(UTC) ) sched._schedule_document_processing("u") def test_job_verification_failure(self, sched, mock_bg): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) _setup_user(sched, "u") with patch.object( sched, "_get_document_scheduler_settings", return_value=DocumentSchedulerSettings(enabled=True), ): mock_bg.get_job.return_value = None sched._schedule_document_processing("u") def test_exception_in_scheduling(self, sched, mock_bg): _setup_user(sched, "u") with patch.object( sched, "_get_document_scheduler_settings", side_effect=RuntimeError("boom"), ): sched._schedule_document_processing("u") # --------------------------------------------------------------------------- # _process_user_documents full flow # --------------------------------------------------------------------------- class TestProcessUserDocumentsFull: def _dss(self, **kw): from local_deep_research.scheduler.background import ( DocumentSchedulerSettings, ) defaults = dict( enabled=True, download_pdfs=False, extract_text=False, generate_rag=False, last_run="", interval_seconds=1800, ) defaults.update(kw) return DocumentSchedulerSettings(**defaults) def _make_research(self, id=1, title="T", completed_at=None): r = MagicMock() r.id = id r.title = title r.completed_at = completed_at or datetime.now(UTC) return r def _setup_db_chain(self, mock_db, research_sessions): chain = mock_db.query.return_value chain.filter.return_value = chain chain.order_by.return_value = chain chain.limit.return_value = chain chain.filter_by.return_value = chain chain.outerjoin.return_value = chain chain.all.return_value = research_sessions return chain def test_no_new_research(self, sched): _setup_user(sched, "u") mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, []) with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(download_pdfs=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), ): sched._process_user_documents("u") def test_download_pdfs(self, sched): _setup_user(sched, "u") research = self._make_research() mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, [research]) mock_ds = MagicMock() _ctx(mock_ds) mock_ds.queue_research_downloads.return_value = 3 with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(download_pdfs=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, return_value=mock_ds), ): sched._process_user_documents("u") mock_ds.queue_research_downloads.assert_called_once() def test_download_pdfs_exception(self, sched): _setup_user(sched, "u") research = self._make_research() mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, [research]) with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(download_pdfs=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, side_effect=RuntimeError("fail")), ): sched._process_user_documents("u") def test_extract_text_success_and_failure(self, sched): _setup_user(sched, "u") research = self._make_research() r_ok = MagicMock(id=10, url="https://arxiv.org/pdf/1.pdf") r_fail = MagicMock(id=11, url="https://arxiv.org/pdf/2.pdf") mock_db = _ctx(MagicMock()) call_count = [0] chain = mock_db.query.return_value chain.filter.return_value = chain chain.order_by.return_value = chain chain.limit.return_value = chain chain.filter_by.return_value = chain def all_fn(): call_count[0] += 1 if call_count[0] == 1: return [research] return [r_ok, r_fail] chain.all.side_effect = all_fn mock_ds = MagicMock() _ctx(mock_ds) mock_ds.download_as_text.side_effect = [ (True, None), (False, "timeout"), ] with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(extract_text=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, return_value=mock_ds), patch(IS_DOWNLOADABLE, return_value=True), ): sched._process_user_documents("u") def test_extract_text_resource_exception(self, sched): _setup_user(sched, "u") research = self._make_research() resource = MagicMock(id=10, url="https://arxiv.org/pdf/1.pdf") mock_db = _ctx(MagicMock()) call_count = [0] chain = mock_db.query.return_value chain.filter.return_value = chain chain.order_by.return_value = chain chain.limit.return_value = chain chain.filter_by.return_value = chain def all_fn(): call_count[0] += 1 return [research] if call_count[0] == 1 else [resource] chain.all.side_effect = all_fn mock_ds = MagicMock() _ctx(mock_ds) mock_ds.download_as_text.side_effect = RuntimeError("boom") with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(extract_text=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, return_value=mock_ds), patch(IS_DOWNLOADABLE, return_value=True), ): sched._process_user_documents("u") def test_extract_text_outer_exception(self, sched): _setup_user(sched, "u") research = self._make_research() mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, [research]) with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(extract_text=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, side_effect=RuntimeError("import fail")), ): sched._process_user_documents("u") def test_generate_rag_alone_is_noop_in_process_user_documents(self, sched): """RAG indexing of research downloads was RETIRED from _process_user_documents into the reconciler. With generate_rag=True but download/extract OFF, this pass short-circuits and never opens a DB session — there is no inline RAG block left to run. """ _setup_user(sched, "u") mock_db = _ctx(MagicMock()) with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(generate_rag=True), ), patch(DB_SESSION, return_value=mock_db) as mock_session, patch(SETTINGS_MGR, return_value=MagicMock()), ): sched._process_user_documents("u") # generate_rag alone no longer enables the download/extract pass. mock_session.assert_not_called() def test_completed_at_string_parsing(self, sched): _setup_user(sched, "u") research = self._make_research(completed_at="2024-06-15T10:00:00Z") mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, [research]) mock_ds = MagicMock() _ctx(mock_ds) mock_ds.queue_research_downloads.return_value = 0 with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(download_pdfs=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, return_value=mock_ds), ): sched._process_user_documents("u") def test_completed_at_invalid_string(self, sched): _setup_user(sched, "u") research = self._make_research(completed_at="not-a-date") research.title = "A" * 60 # long title for truncation mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, [research]) mock_ds = MagicMock() _ctx(mock_ds) mock_ds.queue_research_downloads.return_value = 0 with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(download_pdfs=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, return_value=mock_ds), ): sched._process_user_documents("u") def test_last_run_set_filters_query(self, sched): _setup_user(sched, "u") mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, []) with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss( download_pdfs=True, last_run="2024-01-01T00:00:00" ), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), ): sched._process_user_documents("u") def test_outer_exception(self, sched): _setup_user(sched, "u") with patch.object( sched, "_get_document_scheduler_settings", side_effect=RuntimeError("total fail"), ): sched._process_user_documents("u") def test_completed_at_none(self, sched): _setup_user(sched, "u") research = self._make_research(completed_at=None) research.title = None research.completed_at = None mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, [research]) mock_ds = MagicMock() _ctx(mock_ds) mock_ds.queue_research_downloads.return_value = 0 with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(download_pdfs=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, return_value=mock_ds), ): sched._process_user_documents("u") def test_research_processing_exception(self, sched): _setup_user(sched, "u") research = self._make_research() mock_db = _ctx(MagicMock()) self._setup_db_chain(mock_db, [research]) with ( patch.object( sched, "_get_document_scheduler_settings", return_value=self._dss(download_pdfs=True), ), patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=MagicMock()), patch(DOWNLOAD_SVC, side_effect=RuntimeError("fail")), ): sched._process_user_documents("u") # --------------------------------------------------------------------------- # _check_subscription with DateTrigger reschedule # --------------------------------------------------------------------------- class TestCheckSubscriptionReschedule: def test_reschedule_date_trigger(self, sched, mock_bg): _setup_user(sched, "u") sub = MagicMock() sub.id = 5 sub.status = "active" sub.query_or_topic = "no placeholder" sub.refresh_interval_minutes = 60 sub.name = "Test" sub.model_provider = "test" sub.model = "test" sub.search_strategy = "news" sub.search_engine = "searxng" mock_db = _ctx(MagicMock()) mock_db.query.return_value.get.return_value = sub mock_job = MagicMock() mock_job.trigger.__class__.__name__ = "DateTrigger" mock_bg.get_job.return_value = mock_job with ( patch(DB_SESSION, return_value=mock_db), patch.object(sched, "_trigger_subscription_research_sync"), ): sched._check_subscription("u", 5) assert mock_bg.add_job.call_count >= 1 def test_no_session_job_lookup_error(self, sched, mock_bg): from apscheduler.jobstores.base import JobLookupError mock_bg.remove_job.side_effect = JobLookupError("x") sched._check_subscription("nobody", 1) def test_exception_in_check(self, sched, mock_bg): _setup_user(sched, "u") with patch(DB_SESSION, side_effect=RuntimeError("fail")): sched._check_subscription("u", 1) def test_sub_not_active(self, sched, mock_bg): _setup_user(sched, "u") sub = MagicMock() sub.status = "paused" mock_db = _ctx(MagicMock()) mock_db.query.return_value.get.return_value = sub with patch(DB_SESSION, return_value=mock_db): sched._check_subscription("u", 1) def test_interval_trigger_no_reschedule(self, sched, mock_bg): """Non-DateTrigger job should not be rescheduled.""" _setup_user(sched, "u") sub = MagicMock() sub.id = 5 sub.status = "active" sub.query_or_topic = "q" sub.refresh_interval_minutes = 60 sub.name = "T" sub.model_provider = "t" sub.model = "t" sub.search_strategy = "n" sub.search_engine = "a" mock_db = _ctx(MagicMock()) mock_db.query.return_value.get.return_value = sub mock_job = MagicMock() mock_job.trigger.__class__.__name__ = "IntervalTrigger" mock_bg.get_job.return_value = mock_job with ( patch(DB_SESSION, return_value=mock_db), patch.object(sched, "_trigger_subscription_research_sync"), ): sched._check_subscription("u", 5) # --------------------------------------------------------------------------- # _trigger_subscription_research_sync # --------------------------------------------------------------------------- class TestTriggerResearchSync: def _base_sub(self, search_engine="searxng"): return { "id": 1, "name": "T", "query": "q", "original_query": "q", "model_provider": "openai", "model": "gpt-4", "search_strategy": "news", "search_engine": search_engine, } def test_no_search_engine(self, sched): _setup_user(sched, "u") mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = { "search.tool": {"value": "google", "ui_element": "select"} } with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(QUICK_SUMMARY, return_value={"report": "r"}) as qs, patch(SET_CTX), patch.object(sched, "_store_research_result"), ): sched._trigger_subscription_research_sync( "u", self._base_sub(search_engine=None) ) qs.assert_called_once() def test_with_search_engine(self, sched): _setup_user(sched, "u") mock_db = _ctx(MagicMock()) mock_sm = MagicMock() snapshot = {} mock_sm.get_settings_snapshot.return_value = snapshot with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(QUICK_SUMMARY, return_value={"report": "r"}), patch(SET_CTX), patch.object(sched, "_store_research_result"), ): sched._trigger_subscription_research_sync( "u", self._base_sub(search_engine="bing") ) assert snapshot["search.tool"]["value"] == "bing" def test_exception_handled(self, sched): _setup_user(sched, "u") with patch(DB_SESSION, side_effect=RuntimeError("fail")): sched._trigger_subscription_research_sync("u", self._base_sub()) # --------------------------------------------------------------------------- # _store_research_result edge cases # --------------------------------------------------------------------------- class TestStoreResearchResult: def _call_store(self, sched, result, sub=None, headline="H", topics=None): sub = sub or {"name": "Sub", "query": "q"} mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = {} mock_cf = MagicMock() mock_cf.return_value.format_document.return_value = "formatted" with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(HEADLINE_GEN, return_value=headline), patch(TOPIC_GEN, return_value=topics or []), patch(REPORT_STORAGE), patch(FORMAT_LINKS, return_value="- [link](url)"), patch(CITATION_FMT, mock_cf), patch(GET_SETTING_SNAP, return_value="domain_id_hyperlinks"), ): sched._store_research_result("u", "pw", "rid", 1, result, sub) return mock_db def test_with_sources(self, sched): self._call_store( sched, {"report": "R", "query": "q", "sources": [{"url": "http://x"}]}, ) def test_no_report_uses_summary(self, sched): self._call_store(sched, {"summary": "S", "query": "q", "sources": []}) def test_no_report_no_summary_uses_json(self, sched): mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = {} with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(HEADLINE_GEN, return_value=None), patch(TOPIC_GEN, return_value=[]), patch(REPORT_STORAGE), ): sched._store_research_result( "u", "pw", "rid", 1, {"query": "q", "sources": []}, {"name": "", "query": "q"}, ) def test_no_headline_with_name(self, sched): mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = {} mock_cf = MagicMock() mock_cf.return_value.format_document.return_value = "f" with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(HEADLINE_GEN, return_value=None), patch(TOPIC_GEN, return_value=[]), patch(REPORT_STORAGE), patch(CITATION_FMT, mock_cf), patch(GET_SETTING_SNAP, return_value="number_hyperlinks"), ): sched._store_research_result( "u", "pw", "rid", 1, {"report": "R", "query": "q", "sources": []}, {"name": "My Sub", "query": "q"}, ) def test_no_headline_no_name(self, sched): mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = {} mock_cf = MagicMock() mock_cf.return_value.format_document.return_value = "f" with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(HEADLINE_GEN, return_value=None), patch(TOPIC_GEN, return_value=[]), patch(REPORT_STORAGE), patch(CITATION_FMT, mock_cf), patch(GET_SETTING_SNAP, return_value="no_hyperlinks"), ): sched._store_research_result( "u", "pw", "rid", 1, {"report": "R", "query": "q" * 100, "sources": []}, {"name": "", "query": "q" * 100}, ) def test_make_serializable_with_dict_method(self, sched): class Obj: def dict(self): return {"key": "val"} result = Obj() mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = {} mock_cf = MagicMock() mock_cf.return_value.format_document.return_value = "f" with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(HEADLINE_GEN, return_value="H"), patch(TOPIC_GEN, return_value=[]), patch(REPORT_STORAGE), patch(CITATION_FMT, mock_cf), patch(GET_SETTING_SNAP, return_value="domain_hyperlinks"), ): sched._store_research_result( "u", "pw", "rid", 1, result, {"name": "S", "query": "q"} ) def test_store_exception(self, sched): with patch(DB_SESSION, side_effect=RuntimeError("fail")): sched._store_research_result( "u", "p", "r", 1, {}, {"name": "S", "query": "q"} ) def test_citation_format_domain_id_always(self, sched): """Test domain_id_always_hyperlinks citation mode.""" mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = {} mock_cf = MagicMock() mock_cf.return_value.format_document.return_value = "f" with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(HEADLINE_GEN, return_value="H"), patch(TOPIC_GEN, return_value=["t1"]), patch(REPORT_STORAGE), patch(FORMAT_LINKS, return_value=""), patch(CITATION_FMT, mock_cf), patch(GET_SETTING_SNAP, return_value="domain_id_always_hyperlinks"), ): sched._store_research_result( "u", "pw", "rid", 1, {"report": "R", "query": "q", "sources": [{"url": "x"}]}, {"name": "S", "query": "q"}, ) # --------------------------------------------------------------------------- # get_document_scheduler_status exception # --------------------------------------------------------------------------- class TestDocSchedulerStatusException: def test_exception_returns_error(self, sched): _setup_user(sched, "u") with patch.object( sched, "_get_document_scheduler_settings", side_effect=ValueError("boom"), ): status = sched.get_document_scheduler_status("u") assert status["enabled"] is False assert "ValueError" in status["message"] # --------------------------------------------------------------------------- # trigger_document_processing edge cases # --------------------------------------------------------------------------- class TestTriggerDocProcessingEdgeCases: def test_job_verification_fails(self, running, mock_bg): _setup_user(running, "u") mock_bg.get_job.return_value = None assert running.trigger_document_processing("u") is False def test_exception_returns_false(self, running, mock_bg): _setup_user(running, "u") mock_bg.add_job.side_effect = RuntimeError("fail") assert running.trigger_document_processing("u") is False # --------------------------------------------------------------------------- # _reload_config # --------------------------------------------------------------------------- class TestReloadConfigEdgeCases: def test_exception_in_reload(self, sched, mock_bg): sm = MagicMock() sm.get_setting.side_effect = RuntimeError("fail") sched.settings_manager = sm sched._reload_config() def test_no_retention_change(self, sched, mock_bg): sm = MagicMock() sm.get_setting.side_effect = lambda k, default: default sched.settings_manager = sm sched._reload_config() # --------------------------------------------------------------------------- # _check_user_overdue_subscriptions # --------------------------------------------------------------------------- class TestOverdueSubscriptionsEdgeCases: def test_exception_handled(self, sched, mock_bg): _setup_user(sched, "u") with patch(DB_SESSION, side_effect=RuntimeError("fail")): sched._check_user_overdue_subscriptions("u") def test_no_overdue(self, sched, mock_bg): _setup_user(sched, "u") mock_db = _ctx(MagicMock()) mock_db.query.return_value.filter.return_value.all.return_value = [] with patch(DB_SESSION, return_value=mock_db): sched._check_user_overdue_subscriptions("u") mock_bg.add_job.assert_not_called() # --------------------------------------------------------------------------- # Cleanup with mixed active/inactive # --------------------------------------------------------------------------- class TestCleanupMixed: def test_mixed_users(self, sched, mock_bg): _setup_user(sched, "old", jobs={"j1"}) sched.user_sessions["old"]["last_activity"] = datetime.now( UTC ) - timedelta(hours=100) _setup_user(sched, "new") cleaned = sched._cleanup_inactive_users() assert cleaned == 1 assert "old" not in sched.user_sessions assert "new" in sched.user_sessions def test_cleanup_job_lookup_error(self, sched, mock_bg): from apscheduler.jobstores.base import JobLookupError _setup_user(sched, "old", jobs={"j1"}) sched.user_sessions["old"]["last_activity"] = datetime.now( UTC ) - timedelta(hours=100) mock_bg.remove_job.side_effect = JobLookupError("j1") assert sched._cleanup_inactive_users() == 1 # --------------------------------------------------------------------------- # SnapshotSettingsContext usage # --------------------------------------------------------------------------- class TestSnapshotSettingsContext: def test_settings_context_with_value_dicts(self, sched): _setup_user(sched, "u") sub = { "id": 1, "name": "T", "query": "q", "original_query": "q", "model_provider": "openai", "model": "gpt-4", "search_strategy": "news", "search_engine": "searxng", } mock_db = _ctx(MagicMock()) mock_sm = MagicMock() mock_sm.get_settings_snapshot.return_value = { "key1": {"value": "v1", "ui_element": "text"}, "key2": "plain_value", } captured = [] def capture(ctx): captured.append(ctx) with ( patch(DB_SESSION, return_value=mock_db), patch(SETTINGS_MGR, return_value=mock_sm), patch(QUICK_SUMMARY, return_value={"report": "r"}), patch(SET_CTX, side_effect=capture), patch.object(sched, "_store_research_result"), ): sched._trigger_subscription_research_sync("u", sub) assert len(captured) == 1 ctx = captured[0] assert ctx.get_setting("key1") == "v1" assert ctx.get_setting("key2") == "plain_value" assert ctx.get_setting("missing", "default") == "default"