dennisthiessenandClaude Opus 5
083c9dbf7c
Groundwork for the Admin -> Jobs cleanup. Three sources of truth collapse into
app/job_catalog.py, which imports nothing from app so both the scheduler and
admin_service can import it at module level (admin_service otherwise has to
import the scheduler inside functions to dodge a cycle).
PIPELINE_MEMBERS is now DERIVED from the four pipeline step lists instead of
being a literal set in admin_service duplicating four lists in scheduler.py with
nothing asserting they agreed. A test pins that the derivation reproduces the
previous hand-maintained 9 names exactly, so this is behaviour-preserving.
Deletes the private _JOB_NAMES list, which held 16 of the 19 jobs:
benchmark_collector, outcome_evaluator and shadow_book had no runtime row, and
so no "last run" line in the panel, until their first run in a given process.
_job_runtime is now seeded from the catalog, and a test pins the invariant.
Next-run is decided by category rather than by reading a timestamp. A pipeline
step has no schedule of its own, so it reports its parent's ("next via Morning
Pipeline in 3h") instead of nothing; a manual job says manual_only rather than
rendering a date. This also fixes a real bug: triggering a paused job set
next_run_time=now, APScheduler re-armed the 520-week backstop behind it, and the
panel displayed "next run in ~87600h". Two independent guards -- the category
rule, plus _visible_next_run dropping anything past a year -- and an APScheduler
listener that re-pauses steps and manual jobs once their run finishes. The
listener is registered at module level because configure_scheduler is called
more than once and add_listener does not deduplicate.
Migrates backtest and ticker_universe_sync from interval to cron (Sun 03:00 ET
and 01:00 ET). configure_scheduler calls remove_all_jobs() on every startup, so
an interval countdown restarts each deploy -- a 168h backtest needed a week of
uninterrupted uptime to fire even once. The codebase already documented this
pitfall as the reason cron was adopted; these two were never migrated. Both are
now editable in Admin -> Schedule.
Also: list_jobs went from one settings query per job (19) to one for all of
them, and data_backfill is hidden from the listing while staying registered and
API-triggerable.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
111 lines
4.8 KiB
Python
111 lines
4.8 KiB
Python
"""Admin → Jobs listing: categories, ordering, and next-run coherence.
|
|
|
|
The panel used to render 19 jobs as one alphabetical list in which a pipeline
|
|
step, a cron job and a manual job were indistinguishable, and a triggered job
|
|
could advertise a next run ten years out.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from app import job_catalog
|
|
from app.scheduler import configure_scheduler, scheduler
|
|
from app.services.admin_service import _visible_next_run, list_jobs
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _configured_scheduler():
|
|
scheduler.remove_all_jobs()
|
|
configure_scheduler()
|
|
yield
|
|
scheduler.remove_all_jobs()
|
|
|
|
|
|
def _by_name(jobs: list[dict]) -> dict[str, dict]:
|
|
return {job["name"]: job for job in jobs}
|
|
|
|
|
|
class TestVisibleNextRun:
|
|
def test_parked_backstop_is_not_a_schedule(self):
|
|
"""Paused jobs carry a 520-week interval; triggering one re-arms it."""
|
|
backstop = datetime.now(timezone.utc) + timedelta(weeks=520)
|
|
assert _visible_next_run(backstop) is None
|
|
|
|
def test_a_real_upcoming_run_passes_through(self):
|
|
soon = datetime.now(timezone.utc) + timedelta(hours=6)
|
|
assert _visible_next_run(soon) == soon
|
|
|
|
def test_none_stays_none(self):
|
|
assert _visible_next_run(None) is None
|
|
|
|
|
|
class TestListJobs:
|
|
async def test_hidden_jobs_are_not_listed_but_stay_valid(self, db_session):
|
|
jobs = _by_name(await list_jobs(db_session))
|
|
assert "data_backfill" not in jobs
|
|
# Still triggerable through the API, and still registered.
|
|
assert "data_backfill" in job_catalog.VALID_JOB_NAMES
|
|
assert scheduler.get_job("data_backfill") is not None
|
|
|
|
async def test_every_visible_job_has_a_category(self, db_session):
|
|
jobs = await list_jobs(db_session)
|
|
assert {j["name"] for j in jobs} == set(
|
|
job_catalog.VALID_JOB_NAMES - job_catalog.HIDDEN_JOBS
|
|
)
|
|
assert all(j["category"] in job_catalog.CATEGORY_ORDER for j in jobs)
|
|
|
|
async def test_jobs_arrive_grouped_by_category(self, db_session):
|
|
"""The frontend renders sections in payload order, so ordering is the
|
|
API's job — not something each client re-derives."""
|
|
categories = [j["category"] for j in await list_jobs(db_session)]
|
|
ranks = [job_catalog.CATEGORY_ORDER.index(c) for c in categories]
|
|
assert ranks == sorted(ranks)
|
|
|
|
async def test_pipeline_steps_defer_their_schedule_to_the_parent(self, db_session):
|
|
jobs = _by_name(await list_jobs(db_session))
|
|
step = jobs["rr_scanner"]
|
|
assert step["category"] == job_catalog.CATEGORY_STEP
|
|
assert step["next_run_at"] is None
|
|
assert step["next_run_source"] == "via_pipeline"
|
|
assert step["pipelines"] == ["near_close_pipeline"]
|
|
|
|
async def test_step_reports_the_soonest_enabled_parent(self, db_session):
|
|
due = datetime.now(timezone.utc) + timedelta(hours=3)
|
|
scheduler.modify_job("daily_pipeline", next_run_time=due)
|
|
|
|
collector = _by_name(await list_jobs(db_session))["data_collector"]
|
|
assert collector["via_next_run_job"] == "daily_pipeline"
|
|
assert collector["via_next_run_at"] == due.isoformat()
|
|
# Runs in all four pipelines — the reason steps are not nested under one.
|
|
assert set(collector["pipelines"]) == set(job_catalog.PIPELINE_JOBS)
|
|
|
|
async def test_manual_jobs_say_so_instead_of_showing_a_date(self, db_session):
|
|
study = _by_name(await list_jobs(db_session))["event_study"]
|
|
assert study["category"] == job_catalog.CATEGORY_MANUAL
|
|
assert study["next_run_source"] == "manual_only"
|
|
assert study["next_run_at"] is None
|
|
|
|
async def test_a_triggered_manual_job_still_shows_no_next_run(self, db_session):
|
|
"""Regression: triggering re-armed the 520-week backstop, which the panel
|
|
rendered as a real 'next run in ~87600h'."""
|
|
scheduler.modify_job("event_study", next_run_time=datetime.now(timezone.utc))
|
|
scheduler.modify_job("event_study", next_run_time=None)
|
|
|
|
study = _by_name(await list_jobs(db_session))["event_study"]
|
|
assert study["next_run_at"] is None
|
|
|
|
async def test_pipelines_report_their_own_schedule_and_steps(self, db_session):
|
|
pipeline = _by_name(await list_jobs(db_session))["daily_pipeline"]
|
|
assert pipeline["category"] == job_catalog.CATEGORY_PIPELINE
|
|
assert pipeline["next_run_source"] == "own_schedule"
|
|
assert pipeline["steps"] == [
|
|
step for step, _ in job_catalog.PIPELINE_STEPS["daily_pipeline"]
|
|
]
|
|
|
|
async def test_standalone_jobs_keep_their_own_schedule(self, db_session):
|
|
backtest = _by_name(await list_jobs(db_session))["backtest"]
|
|
assert backtest["category"] == job_catalog.CATEGORY_SCHEDULED
|
|
assert backtest["next_run_source"] == "own_schedule"
|
|
assert backtest["pipelines"] == []
|