"""Unit tests for app.scheduler module.""" from types import SimpleNamespace import pytest from app.scheduler import ( _consume_backtest_options, _consume_backtest_target_model, _parse_frequency, _resume_tickers, _last_successful, _run_shadow_import, collect_fundamentals, run_fundamentals_parity_report, run_sec_fundamentals_import, configure_scheduler, get_job_runtime_snapshot, queue_backtest_options, queue_backtest_target_model, scheduler, ) def test_manual_backtest_target_model_is_one_shot(): assert queue_backtest_target_model("structural_sr") == "structural_sr" assert _consume_backtest_target_model() == "structural_sr" assert _consume_backtest_target_model() == "production_gtl" def test_manual_backtest_target_model_rejects_removed_research_arms(): with pytest.raises(ValueError, match="Unknown backtest target model"): queue_backtest_target_model("production_control") def test_manual_backtest_options_are_one_shot_and_default_back_to_weekly(): assert queue_backtest_options("structural_sr", "daily") == ( "structural_sr", "daily", ) assert _consume_backtest_options() == ("structural_sr", "daily") assert _consume_backtest_options() == ("production_gtl", "weekly") class TestParseFrequency: def test_hourly(self): assert _parse_frequency("hourly") == {"hours": 1} def test_daily(self): assert _parse_frequency("daily") == {"hours": 24} def test_case_insensitive(self): assert _parse_frequency("Hourly") == {"hours": 1} assert _parse_frequency("DAILY") == {"hours": 24} def test_weekly_maps_to_one_week(self): assert _parse_frequency("weekly") == {"weeks": 1} def test_unknown_defaults_to_daily(self): assert _parse_frequency("monthly") == {"hours": 24} assert _parse_frequency("") == {"hours": 24} class TestResumeTickers: def test_no_previous_returns_full_list(self): symbols = ["AAPL", "GOOG", "MSFT"] _last_successful["test_job"] = None result = _resume_tickers(symbols, "test_job") assert result == ["AAPL", "GOOG", "MSFT"] def test_resume_after_first(self): symbols = ["AAPL", "GOOG", "MSFT"] _last_successful["test_job"] = "AAPL" result = _resume_tickers(symbols, "test_job") # Should start from GOOG, then wrap around assert result == ["GOOG", "MSFT", "AAPL"] def test_resume_after_middle(self): symbols = ["AAPL", "GOOG", "MSFT", "TSLA"] _last_successful["test_job"] = "GOOG" result = _resume_tickers(symbols, "test_job") assert result == ["MSFT", "TSLA", "AAPL", "GOOG"] def test_resume_after_last(self): symbols = ["AAPL", "GOOG", "MSFT"] _last_successful["test_job"] = "MSFT" result = _resume_tickers(symbols, "test_job") # All already processed, wraps to full list assert result == ["AAPL", "GOOG", "MSFT"] def test_unknown_last_returns_full_list(self): symbols = ["AAPL", "GOOG", "MSFT"] _last_successful["test_job"] = "NVDA" result = _resume_tickers(symbols, "test_job") assert result == ["AAPL", "GOOG", "MSFT"] def test_empty_list(self): _last_successful["test_job"] = "AAPL" result = _resume_tickers([], "test_job") assert result == [] class TestConfigureScheduler: def test_configure_adds_all_jobs(self): # Remove any existing jobs first scheduler.remove_all_jobs() configure_scheduler() jobs = scheduler.get_jobs() job_ids = {j.id for j in jobs} assert job_ids == { "data_collector", "data_backfill", "benchmark_collector", "sentiment_collector", "fundamental_collector", "dolt_earnings_import", "sec_fundamentals_import", "fundamentals_parity_report", "rr_scanner", "shadow_book", "ticker_universe_sync", "outcome_evaluator", "alerts", "market_regime", "regime_monitor", "event_study", "backtest", "daily_pipeline", "near_close_pipeline", "after_close_pipeline", "intraday_pipeline", } def test_configure_is_idempotent(self): scheduler.remove_all_jobs() configure_scheduler() configure_scheduler() # Should replace, not duplicate job_ids = [j.id for j in scheduler.get_jobs()] # Each ID should appear exactly once assert sorted(job_ids) == sorted([ "after_close_pipeline", "alerts", "backtest", "benchmark_collector", "daily_pipeline", "intraday_pipeline", "data_collector", "data_backfill", "fundamental_collector", "dolt_earnings_import", "sec_fundamentals_import", "fundamentals_parity_report", "market_regime", "near_close_pipeline", "regime_monitor", "event_study", "outcome_evaluator", "rr_scanner", "sentiment_collector", "shadow_book", "ticker_universe_sync", ]) class _SessionContext: async def __aenter__(self): return object() async def __aexit__(self, *exc): return None class TestFundamentalCollector: @staticmethod def _session_factory(): return _SessionContext() async def test_skips_legacy_provider_when_cutover_is_active(self, monkeypatch): async def enabled(db, job_name): return True async def cutover_enabled(db): return True async def unexpected_ticker_lookup(db): raise AssertionError("legacy ticker lookup must not run after cutover") monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr( "app.scheduler.fundamental_data_refresh_service.is_enabled", cutover_enabled, ) monkeypatch.setattr( "app.scheduler._get_fundamental_priority_tickers", unexpected_ticker_lookup, ) await collect_fundamentals() runtime = get_job_runtime_snapshot("fundamental_collector") assert runtime["status"] == "skipped" assert runtime["processed"] == 0 assert runtime["total"] == 0 assert runtime["message"] == "SEC + Dolt fundamentals cutover is active" class TestShadowImportJobs: @staticmethod def _session_factory(): return _SessionContext() async def test_promoted_run_surfaces_completion(self, monkeypatch): async def enabled(db, job_name): return True async def imported(importer): return SimpleNamespace( status="promoted", revision="abcdef1234567890", error_details=None ) monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler.run_import", imported) await _run_shadow_import("dolt_earnings_import", object()) runtime = get_job_runtime_snapshot("dolt_earnings_import") assert runtime["status"] == "completed" assert runtime["processed"] == 1 assert runtime["message"] == "promoted · abcdef123456" async def test_failed_run_surfaces_error(self, monkeypatch): async def enabled(db, job_name): return True async def imported(importer): return SimpleNamespace( status="failed", revision=None, error_details="validation failed" ) monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler.run_import", imported) await _run_shadow_import("sec_fundamentals_import", object()) runtime = get_job_runtime_snapshot("sec_fundamentals_import") assert runtime["status"] == "error" assert runtime["processed"] == 0 assert runtime["message"] == "validation failed" async def test_source_lock_surfaces_skipped(self, monkeypatch): async def enabled(db, job_name): return True async def imported(importer): return None monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler.run_import", imported) await _run_shadow_import("dolt_earnings_import", object()) runtime = get_job_runtime_snapshot("dolt_earnings_import") assert runtime["status"] == "skipped" assert "already running" in runtime["message"] async def test_disabled_job_never_runs_importer(self, monkeypatch): async def disabled(db, job_name): return False async def should_not_run(importer): raise AssertionError("disabled job ran importer") monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", disabled) monkeypatch.setattr("app.scheduler.run_import", should_not_run) await _run_shadow_import("sec_fundamentals_import", object()) runtime = get_job_runtime_snapshot("sec_fundamentals_import") assert runtime["status"] == "skipped" assert runtime["message"] == "Disabled" async def test_sec_failure_still_runs_activated_local_refresh(self, monkeypatch): calls = [] async def enabled(db, job_name): return True async def unavailable(importer): raise RuntimeError("SEC unavailable") async def refreshed(db): calls.append(db) return { "enabled": True, "refreshed": 511, "score_inputs_changed": 2, "dimension_scores_staled": 2, "composite_scores_staled": 2, } monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler.run_import", unavailable) monkeypatch.setattr( "app.scheduler.fundamental_data_refresh_service.refresh_if_enabled", refreshed, ) await run_sec_fundamentals_import() assert len(calls) == 1 runtime = get_job_runtime_snapshot("sec_fundamentals_import") assert runtime["status"] == "error" assert runtime["message"] == "SEC unavailable" async def test_sec_success_surfaces_activated_refresh_summary(self, monkeypatch): async def enabled(db, job_name): return True async def imported(importer): return SimpleNamespace( status="no_op", revision="abcdef1234567890", error_details=None ) async def refreshed(db): return { "enabled": True, "refreshed": 511, "score_inputs_changed": 2, "dimension_scores_staled": 2, "composite_scores_staled": 2, } monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr("app.scheduler.run_import", imported) monkeypatch.setattr( "app.scheduler.fundamental_data_refresh_service.refresh_if_enabled", refreshed, ) await run_sec_fundamentals_import() runtime = get_job_runtime_snapshot("sec_fundamentals_import") assert runtime["status"] == "completed" assert runtime["message"] == ( "no_op · abcdef123456 · cache 511 · 2 score inputs changed" ) async def test_disabled_sec_job_does_not_run_local_refresh(self, monkeypatch): async def disabled(db, job_name): return False async def should_not_run(*args, **kwargs): raise AssertionError("disabled SEC job ran work") monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", disabled) monkeypatch.setattr("app.scheduler.run_import", should_not_run) monkeypatch.setattr( "app.scheduler.fundamental_data_refresh_service.refresh_if_enabled", should_not_run, ) await run_sec_fundamentals_import() runtime = get_job_runtime_snapshot("sec_fundamentals_import") assert runtime["status"] == "skipped" assert runtime["message"] == "Disabled" async def test_fundamentals_parity_job_surfaces_report_summary(monkeypatch): async def enabled(db, job_name): return True async def generated(db, report_dir): return ( { "generated_at": "2026-07-23T10:30:00+00:00", "summary": { "universe_count": 511, "fundamental_score_material_changes": 12, }, }, {"json": "report.json", "csv": "report.csv"}, ) monkeypatch.setattr("app.scheduler.async_session_factory", TestShadowImportJobs._session_factory) monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) monkeypatch.setattr( "app.scheduler.fundamentals_parity_service.generate_and_store", generated ) await run_fundamentals_parity_report() runtime = get_job_runtime_snapshot("fundamentals_parity_report") assert runtime["status"] == "completed" assert runtime["message"] == "511 tickers · 12 material score changes"