90 lines
2.8 KiB
Python
90 lines
2.8 KiB
Python
"""
|
|
Pruebas del programador de mantenimiento diario: gating "máximo 1/día", catch-up tras
|
|
reinicios y respeto de run_at_hour. Se mockea ConfigRepository con un dict en memoria.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from app.engine import maintenance_scheduler as msched
|
|
from app.engine.maintenance_scheduler import DailyMaintenanceScheduler
|
|
|
|
|
|
@pytest.fixture
|
|
def state_store(monkeypatch):
|
|
store: dict = {}
|
|
monkeypatch.setattr(
|
|
msched.ConfigRepository, "get", staticmethod(lambda key, default=None: store.get(key, default))
|
|
)
|
|
monkeypatch.setattr(
|
|
msched.ConfigRepository, "set", staticmethod(lambda key, value: store.__setitem__(key, value))
|
|
)
|
|
monkeypatch.setattr(msched.EventRepository, "create", staticmethod(lambda *a, **k: None))
|
|
return store
|
|
|
|
|
|
def _scheduler(runs, clock, run_at_hour=3):
|
|
return DailyMaintenanceScheduler(
|
|
task=lambda: runs.append(1),
|
|
run_at_hour=run_at_hour,
|
|
clock=clock,
|
|
)
|
|
|
|
|
|
def test_corre_una_vez_por_dia(state_store):
|
|
runs: list[int] = []
|
|
now = {"dt": datetime(2026, 7, 24, 5, 0, 0)}
|
|
sched = _scheduler(runs, lambda: now["dt"])
|
|
|
|
sched._run_if_due() # primer día: corre
|
|
sched._run_if_due() # mismo día: NO corre
|
|
assert len(runs) == 1
|
|
|
|
|
|
def test_catch_up_al_cambiar_de_dia(state_store):
|
|
runs: list[int] = []
|
|
now = {"dt": datetime(2026, 7, 24, 5, 0, 0)}
|
|
sched = _scheduler(runs, lambda: now["dt"])
|
|
|
|
sched._run_if_due()
|
|
assert len(runs) == 1
|
|
|
|
now["dt"] = datetime(2026, 7, 25, 5, 0, 0) # día nuevo
|
|
sched._run_if_due()
|
|
assert len(runs) == 2
|
|
|
|
|
|
def test_respeta_run_at_hour(state_store):
|
|
runs: list[int] = []
|
|
now = {"dt": datetime(2026, 7, 24, 1, 0, 0)} # antes de las 3
|
|
sched = _scheduler(runs, lambda: now["dt"], run_at_hour=3)
|
|
|
|
sched._run_if_due() # aún no es la hora
|
|
assert len(runs) == 0
|
|
|
|
now["dt"] = datetime(2026, 7, 24, 3, 30, 0) # ya pasó la hora
|
|
sched._run_if_due()
|
|
assert len(runs) == 1
|
|
|
|
|
|
def test_run_at_hour_none_corre_al_primer_wake(state_store):
|
|
runs: list[int] = []
|
|
now = {"dt": datetime(2026, 7, 24, 0, 5, 0)}
|
|
sched = _scheduler(runs, lambda: now["dt"], run_at_hour=None)
|
|
sched._run_if_due()
|
|
assert len(runs) == 1
|
|
|
|
|
|
def test_claim_al_inicio_persiste_fecha_aunque_falle(state_store):
|
|
"""Si la tarea falla, el turno del día igual se consume (claim al inicio)."""
|
|
now = {"dt": datetime(2026, 7, 24, 5, 0, 0)}
|
|
|
|
def boom():
|
|
raise RuntimeError("fallo de limpieza")
|
|
|
|
sched = DailyMaintenanceScheduler(task=boom, run_at_hour=3, clock=lambda: now["dt"])
|
|
sched._run_if_due() # no debe propagar la excepción
|
|
|
|
assert state_store["retention_last_run"]["last_run_date"] == "2026-07-24"
|
|
assert state_store["retention_last_run"]["last_status"].startswith("error")
|