dennisthiessenandClaude Opus 5
e1607ddbff
Three follow-ups from review of the A6 commits. The cache-summary re-finalize raised a second durable system event on a failed run: _runtime_finish emits for `error`/`rate_limited`, and the dedup key includes the message, so "SEC unavailable" and "SEC unavailable · cache 511 · 2 score inputs changed" landed as two unacknowledged Admin events. Adds `emit_event` so a re-finalize that only rewords an outcome stays silent, with a regression test asserting exactly one event. The rollback section still claimed disabling the SEC job freezes the cache — the opposite of what the same page says two lines earlier, and of what the code now does. Rewritten: there is no Admin cache-off switch, restoring `fundamental_data` alone is temporary because the next run rebuilds it from the same snapshots and code, and a real freeze means stopping the service. Remaining "shadow" wording: the two import jobs have never been shadow since activation, so `_run_shadow_import` -> `_run_source_import`, its section heading, the deployment doc's job label, and the plan doc's "production switch remains" handoff paragraph are all brought up to date. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
435 lines
15 KiB
Python
435 lines
15 KiB
Python
"""Unit tests for app.scheduler module."""
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.scheduler import (
|
|
_DAILY_PIPELINE_STEPS,
|
|
_NEAR_CLOSE_PIPELINE_STEPS,
|
|
_consume_backtest_options,
|
|
_consume_backtest_target_model,
|
|
_parse_frequency,
|
|
_resume_tickers,
|
|
_last_successful,
|
|
_run_source_import,
|
|
run_sec_fundamentals_import,
|
|
configure_scheduler,
|
|
get_job_runtime_snapshot,
|
|
queue_backtest_options,
|
|
queue_backtest_target_model,
|
|
scheduler,
|
|
)
|
|
from app.services.data_import import STATUS_DEFERRED
|
|
|
|
|
|
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")
|
|
|
|
|
|
def test_only_near_close_fetch_skips_redundant_sr_refresh():
|
|
assert dict(_DAILY_PIPELINE_STEPS)["data_collector"] == "collect_ohlcv"
|
|
assert (
|
|
dict(_NEAR_CLOSE_PIPELINE_STEPS)["data_collector"]
|
|
== "collect_ohlcv_for_scan"
|
|
)
|
|
|
|
|
|
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",
|
|
"dolt_earnings_import",
|
|
"sec_fundamentals_import",
|
|
"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",
|
|
"dolt_earnings_import",
|
|
"sec_fundamentals_import",
|
|
"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 TestSourceImportJobs:
|
|
@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_source_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_source_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_deferred_run_is_visible_without_error_status(self, monkeypatch):
|
|
async def enabled(db, job_name):
|
|
return True
|
|
|
|
async def imported(importer):
|
|
return SimpleNamespace(
|
|
status=STATUS_DEFERRED,
|
|
revision="abcdef1234567890",
|
|
error_details="Company Facts publication lag; retrying",
|
|
)
|
|
|
|
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_source_import("sec_fundamentals_import", object())
|
|
|
|
runtime = get_job_runtime_snapshot("sec_fundamentals_import")
|
|
assert runtime["status"] == STATUS_DEFERRED
|
|
assert runtime["processed"] == 0
|
|
assert runtime["message"] == "Company Facts publication lag; retrying"
|
|
|
|
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_source_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_source_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_local_cache_refresh(self, monkeypatch):
|
|
calls = []
|
|
events = []
|
|
|
|
async def enabled(db, job_name):
|
|
return True
|
|
|
|
async def unavailable(importer):
|
|
raise RuntimeError("SEC unavailable")
|
|
|
|
async def refreshed(db):
|
|
calls.append(db)
|
|
return {
|
|
"refreshed": 511,
|
|
"score_inputs_changed": 2,
|
|
"dimension_scores_staled": 2,
|
|
"composite_scores_staled": 2,
|
|
}
|
|
|
|
async def record(**kwargs):
|
|
events.append(kwargs)
|
|
|
|
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._record_system_event", record)
|
|
monkeypatch.setattr(
|
|
"app.scheduler.fundamental_data_refresh_service.refresh",
|
|
refreshed,
|
|
)
|
|
|
|
await run_sec_fundamentals_import()
|
|
await asyncio.sleep(0) # let the fire-and-forget event task run
|
|
|
|
assert len(calls) == 1
|
|
runtime = get_job_runtime_snapshot("sec_fundamentals_import")
|
|
assert runtime["status"] == "error"
|
|
# the failure stays the headline, but the cache result is still visible
|
|
assert runtime["message"] == (
|
|
"SEC unavailable · cache 511 · 2 score inputs changed"
|
|
)
|
|
# Rewording the outcome must not duplicate the durable event: the dedup
|
|
# key includes the message, so a second finish would show up twice in
|
|
# Admin → System Events.
|
|
assert len(events) == 1, events
|
|
|
|
async def test_sec_success_surfaces_cache_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 {
|
|
"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",
|
|
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_source_locked_sec_run_still_reports_the_cache_refresh(
|
|
self, monkeypatch
|
|
):
|
|
"""A skipped import keeps its skip status but shows the cache advanced."""
|
|
|
|
async def enabled(db, job_name):
|
|
return True
|
|
|
|
async def locked(importer):
|
|
return None # another import owns the source lock
|
|
|
|
async def refreshed(db):
|
|
return {
|
|
"refreshed": 511,
|
|
"score_inputs_changed": 0,
|
|
"dimension_scores_staled": 0,
|
|
"composite_scores_staled": 0,
|
|
}
|
|
|
|
monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory)
|
|
monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
|
|
monkeypatch.setattr("app.scheduler.run_import", locked)
|
|
monkeypatch.setattr(
|
|
"app.scheduler.fundamental_data_refresh_service.refresh",
|
|
refreshed,
|
|
)
|
|
|
|
await run_sec_fundamentals_import()
|
|
|
|
runtime = get_job_runtime_snapshot("sec_fundamentals_import")
|
|
assert runtime["status"] == "skipped"
|
|
assert runtime["message"] == (
|
|
"Another import for this source is already running · "
|
|
"cache 511 · 0 score inputs changed"
|
|
)
|
|
|
|
async def test_disabled_sec_job_still_refreshes_local_cache(self, monkeypatch):
|
|
"""Disabling the job stops the SEC fetch, not the local cache.
|
|
|
|
The cache is derived from stored snapshots, earnings events and closes.
|
|
Prices and earnings move daily even when no filing does, and there is no
|
|
provider fallback since A6 — freezing it would silently stale scoring.
|
|
"""
|
|
calls = []
|
|
|
|
async def disabled(db, job_name):
|
|
return False
|
|
|
|
async def should_not_run(*args, **kwargs):
|
|
raise AssertionError("disabled SEC job hit the network")
|
|
|
|
async def refreshed(db):
|
|
calls.append(db)
|
|
return {
|
|
"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", disabled)
|
|
monkeypatch.setattr("app.scheduler.run_import", should_not_run)
|
|
monkeypatch.setattr(
|
|
"app.scheduler.fundamental_data_refresh_service.refresh",
|
|
refreshed,
|
|
)
|
|
|
|
await run_sec_fundamentals_import()
|
|
|
|
assert len(calls) == 1
|
|
runtime = get_job_runtime_snapshot("sec_fundamentals_import")
|
|
assert runtime["status"] == "completed"
|
|
assert runtime["message"] == (
|
|
"Import disabled · cache 511 · 2 score inputs changed"
|
|
)
|