"""Daemon tests with a fake scheduler (captures add_job) and a fake weekly clock.""" from __future__ import annotations import signal from datetime import datetime, timedelta from zoneinfo import ZoneInfo import pytest from _fakes import NOW, WEEK, FakeClock, FakeDeliverer, FakeStore from apscheduler.events import EVENT_JOB_MISSED, JobExecutionEvent from apscheduler.schedulers import SchedulerNotRunningError from apscheduler.triggers.date import DateTrigger from apscheduler.triggers.interval import IntervalTrigger from sebrief import daemon as daemon_mod from sebrief.config import Settings from sebrief.daemon import DIGEST_JOB_ID, HEARTBEAT_JOB_ID, RECOVERY_JOB_ID, Daemon KST = ZoneInfo("Asia/Seoul") NEXT_RUN = ("2026-W37", datetime(2026, 9, 14, 7, 0, tzinfo=KST)) class FakeScheduler: def __init__(self): self.jobs: dict[str, dict] = {} self.add_calls: list[dict] = [] self.shutdown_calls: list[bool] = [] self.listeners: list[tuple[object, int | None]] = [] def add_listener(self, callback, mask=None): self.listeners.append((callback, mask)) def add_job(self, func, trigger=None, **kwargs): entry = {"func": func, "trigger": trigger, **kwargs} self.add_calls.append(entry) self.jobs[kwargs.get("id", func.__name__)] = entry return entry def shutdown(self, wait=True): self.shutdown_calls.append(wait) class RunRecorder: def __init__(self, fail: bool = False): self.weeks: list[str] = [] self.fail = fail def __call__(self, week: str): self.weeks.append(week) if self.fail: raise RuntimeError("pipeline exploded") def make_settings(tmp_path, **overrides) -> Settings: values = {"SEBRIEF_DATA_DIR": str(tmp_path / "data"), "SEBRIEF_LOG_DIR": str(tmp_path / "logs")} values.update(overrides) return Settings(_env_file=None, **values) def make_daemon(tmp_path, *, clock=None, store=None, run_fn=None, settings=None, deliverer=None): scheduler = FakeScheduler() clock = clock or FakeClock(next_run=NEXT_RUN) store = store or FakeStore() run_fn = run_fn or RunRecorder() deliverer = deliverer or FakeDeliverer("file") d = Daemon( settings or make_settings(tmp_path), scheduler=scheduler, clock=clock, store=store, run_fn=run_fn, now_fn=lambda: NOW, deliverers_fn=lambda s: [deliverer], ) return d, scheduler, clock, store, run_fn, deliverer # -------------------------------------------------------------------------------- scheduling def test_schedule_next_schedules_one_shot_job(tmp_path): d, scheduler, *_ = make_daemon(tmp_path) assert d.schedule_next() == NEXT_RUN job = scheduler.jobs[DIGEST_JOB_ID] assert isinstance(job["trigger"], DateTrigger) assert job["trigger"].run_date == NEXT_RUN[1] assert job["args"] == [NEXT_RUN[0], NEXT_RUN[1]] assert job["replace_existing"] is True # None: a late (sleep/wake) job must still FIRE so it can reschedule itself assert job["misfire_grace_time"] is None def test_schedule_next_survives_clock_error(tmp_path): class BrokenClock(FakeClock): def next_run(self, now): raise RuntimeError("clock broken") d, scheduler, *_ = make_daemon(tmp_path, clock=BrokenClock()) assert d.schedule_next() is None assert DIGEST_JOB_ID not in scheduler.jobs d.clock = FakeClock(next_run=NEXT_RUN) d._heartbeat_job() # heartbeat retries scheduling assert DIGEST_JOB_ID in scheduler.jobs def test_start_schedules_digest_and_heartbeat(tmp_path): d, scheduler, *_ = make_daemon(tmp_path) d.start() assert set(scheduler.jobs) == {DIGEST_JOB_ID, HEARTBEAT_JOB_ID} hb = scheduler.jobs[HEARTBEAT_JOB_ID] assert isinstance(hb["trigger"], IntervalTrigger) assert hb["trigger"].interval.total_seconds() == 3600 assert (d._on_job_missed, EVENT_JOB_MISSED) in scheduler.listeners # ---------------------------------------------------------------------------------- catch-up def test_catchup_runs_when_candidate_and_not_yet_digested(tmp_path): clock = FakeClock(catchup=WEEK, next_run=NEXT_RUN) d, _scheduler, clock, _store, run_fn, _ = make_daemon(tmp_path, clock=clock) assert d.run_catchup() is True assert run_fn.weeks == [WEEK] assert clock.calls[0] == ("catchup_candidate", (NOW, 72)) def test_catchup_skipped_when_digest_exists(tmp_path): store = FakeStore() store.digests[WEEK] = object() d, _s, _c, _st, run_fn, _ = make_daemon(tmp_path, clock=FakeClock(catchup=WEEK), store=store) assert d.run_catchup() is False assert run_fn.weeks == [] def test_catchup_skipped_without_candidate(tmp_path): d, _s, _c, _st, run_fn, _ = make_daemon(tmp_path) assert d.run_catchup() is False assert run_fn.weeks == [] def test_catchup_failure_is_logged_and_start_continues(tmp_path): d, scheduler, _c, _st, run_fn, deliverer = make_daemon( tmp_path, clock=FakeClock(catchup=WEEK, next_run=NEXT_RUN), run_fn=RunRecorder(fail=True) ) d.start() # must not raise assert run_fn.weeks == [WEEK] assert DIGEST_JOB_ID in scheduler.jobs assert deliverer.errors and "catch-up" in deliverer.errors[0][0] def test_catchup_window_comes_from_settings(tmp_path): settings = make_settings(tmp_path, SEBRIEF_CATCHUP_WINDOW_HOURS="24") d, _s, clock, *_ = make_daemon(tmp_path, settings=settings) d.run_catchup() assert clock.calls[0] == ("catchup_candidate", (NOW, 24)) # -------------------------------------------------------------------------------------- job def test_job_runs_pipeline_and_reschedules(tmp_path): d, scheduler, clock, _st, run_fn, deliverer = make_daemon(tmp_path) d.schedule_next() job = scheduler.jobs[DIGEST_JOB_ID] clock._next_run = ("2026-W38", datetime(2026, 9, 21, 7, 0, tzinfo=KST)) job["func"](*job["args"]) assert run_fn.weeks == [NEXT_RUN[0]] assert scheduler.jobs[DIGEST_JOB_ID]["args"] == ["2026-W38", clock._next_run[1]] assert d.next_run == clock._next_run assert deliverer.errors == [] def test_job_exception_notifies_and_still_reschedules(tmp_path): d, scheduler, _clock, _st, run_fn, deliverer = make_daemon(tmp_path, run_fn=RunRecorder(fail=True)) d.schedule_next() job = scheduler.jobs[DIGEST_JOB_ID] job["func"](*job["args"]) # must not raise assert run_fn.weeks == [NEXT_RUN[0]] assert scheduler.jobs[DIGEST_JOB_ID]["trigger"].run_date == NEXT_RUN[1] subject, detail = deliverer.errors[0] assert "다이제스트 실패" in subject and "pipeline exploded" in detail @pytest.mark.parametrize(("lateness_h", "expect_run"), [(0, True), (48, True), (72, False), (100, False)]) def test_late_job_honours_catchup_window(tmp_path, lateness_h, expect_run): d, scheduler, _c, _st, run_fn, deliverer = make_daemon(tmp_path) run_at = NOW - timedelta(hours=lateness_h) d._digest_job(WEEK, run_at) assert (run_fn.weeks == [WEEK]) is expect_run assert scheduler.jobs[DIGEST_JOB_ID]["trigger"].run_date == NEXT_RUN[1] # always rescheduled assert deliverer.errors == [] # --------------------------------------------------------------------------- heartbeat / missed def test_heartbeat_recovers_dropped_job_inside_window(tmp_path): now = {"t": NOW} d, scheduler, clock, _st, run_fn, _ = make_daemon(tmp_path) d.now_fn = lambda: now["t"] d.schedule_next() del scheduler.jobs[DIGEST_JOB_ID] # what APScheduler does when it drops a misfired job now["t"] = NEXT_RUN[1] + timedelta(hours=2) clock.catchup = NEXT_RUN[0] d._heartbeat_job() assert run_fn.weeks == [NEXT_RUN[0]] assert DIGEST_JOB_ID in scheduler.jobs def test_heartbeat_leaves_schedule_alone_when_run_is_imminent(tmp_path): now = {"t": NOW} d, scheduler, _c, _st, run_fn, _ = make_daemon(tmp_path) d.now_fn = lambda: now["t"] d.schedule_next() for delta in (timedelta(minutes=-30), timedelta(0), timedelta(minutes=2)): now["t"] = NEXT_RUN[1] + delta d._heartbeat_job() assert len(scheduler.add_calls) == 1 and run_fn.weeks == [] def test_heartbeat_does_not_run_while_digest_in_progress(tmp_path): now = {"t": NOW} d, scheduler, clock, _st, run_fn, _ = make_daemon(tmp_path) d.now_fn = lambda: now["t"] d.schedule_next() now["t"] = NEXT_RUN[1] + timedelta(hours=1) clock.catchup = NEXT_RUN[0] with d._run_lock: # a digest is running on another executor thread d._heartbeat_job() assert run_fn.weeks == [] # no concurrent second run assert DIGEST_JOB_ID in scheduler.jobs # but re-armed def test_on_job_missed_queues_recovery_under_a_different_id(tmp_path): d, scheduler, clock, _st, run_fn, _ = make_daemon(tmp_path) d.start() del scheduler.jobs[DIGEST_JOB_ID] clock.catchup = NEXT_RUN[0] d._on_job_missed(JobExecutionEvent(EVENT_JOB_MISSED, DIGEST_JOB_ID, "default", NEXT_RUN[1])) assert scheduler.add_calls[-1]["id"] == RECOVERY_JOB_ID recovery = scheduler.jobs[RECOVERY_JOB_ID] assert isinstance(recovery["trigger"], DateTrigger) and recovery["misfire_grace_time"] is None recovery["func"]() assert run_fn.weeks == [NEXT_RUN[0]] assert DIGEST_JOB_ID in scheduler.jobs n_calls = len(scheduler.add_calls) d._on_job_missed(JobExecutionEvent(EVENT_JOB_MISSED, HEARTBEAT_JOB_ID, "default", NEXT_RUN[1])) assert len(scheduler.add_calls) == n_calls # other jobs are ignored # --------------------------------------------------------------------------------- shutdown def test_signal_handler_shuts_scheduler_down(tmp_path): d, scheduler, *_ = make_daemon(tmp_path) old_term, old_int = signal.getsignal(signal.SIGTERM), signal.getsignal(signal.SIGINT) try: d.install_signal_handlers() handler = signal.getsignal(signal.SIGTERM) assert callable(handler) and handler is not old_term handler(signal.SIGTERM, None) finally: signal.signal(signal.SIGTERM, old_term) signal.signal(signal.SIGINT, old_int) assert scheduler.shutdown_calls == [False] def test_signal_during_catchup_unwinds_when_scheduler_not_running(tmp_path): class NotStarted(FakeScheduler): def shutdown(self, wait=True): super().shutdown(wait) raise SchedulerNotRunningError d, *_ = make_daemon(tmp_path) d.scheduler = NotStarted() old_term, old_int = signal.getsignal(signal.SIGTERM), signal.getsignal(signal.SIGINT) try: d.install_signal_handlers() handler = signal.getsignal(signal.SIGTERM) with pytest.raises(KeyboardInterrupt): handler(signal.SIGTERM, None) finally: signal.signal(signal.SIGTERM, old_term) signal.signal(signal.SIGINT, old_int) assert d.stop_requested is True def test_shutdown_during_catchup_prevents_scheduling(tmp_path): class NotStarted(FakeScheduler): def shutdown(self, wait=True): super().shutdown(wait) raise SchedulerNotRunningError holder: dict = {} def run_fn(week): holder["d"].shutdown() d, *_ = make_daemon(tmp_path, clock=FakeClock(catchup=WEEK, next_run=NEXT_RUN), run_fn=run_fn) d.scheduler = NotStarted() holder["d"] = d d.start() assert d.stop_requested is True assert d.scheduler.jobs == {} # nothing scheduled; run_daemon skips scheduler.start() # ------------------------------------------------------------------------------- run_daemon def test_run_daemon_wires_real_parts_and_blocks_on_scheduler(monkeypatch, tmp_path): started = {} fake_store = FakeStore() class FakeBlocking(FakeScheduler): def __init__(self, timezone=None): super().__init__() started["timezone"] = timezone def start(self, paused=False): started["started"] = True raise KeyboardInterrupt # simulate Ctrl-C ending the blocking loop monkeypatch.setattr(daemon_mod, "BlockingScheduler", FakeBlocking) monkeypatch.setattr(daemon_mod, "WeeklyClock", lambda wd, t: FakeClock(next_run=NEXT_RUN)) monkeypatch.setattr(daemon_mod, "Store", lambda path: fake_store) monkeypatch.setattr(daemon_mod.signal, "signal", lambda *a, **k: None) daemon_mod.run_daemon(make_settings(tmp_path)) assert started["started"] is True assert str(started["timezone"]) == "Asia/Seoul" assert fake_store.closed is True def test_run_lock_serialises_catchup_and_job(tmp_path): seen = {} d, *_ = make_daemon(tmp_path) def run_fn(week): seen["locked"] = d._run_lock.locked() d.run_fn = run_fn d._digest_job(WEEK, NOW) assert seen["locked"] is True def test_notify_error_failures_are_swallowed(tmp_path): class Exploding(FakeDeliverer): def notify_error(self, subject, detail): raise RuntimeError("channel down") d, *_ = make_daemon(tmp_path, run_fn=RunRecorder(fail=True), deliverer=Exploding("x")) d._digest_job(WEEK) # must not raise d.deliverers_fn = lambda s: (_ for _ in ()).throw(RuntimeError("cannot build")) d._digest_job(WEEK) # must not raise either