#!/usr/bin/env python3 """Ejercita el servicio con reloj simulado, sin red ni Telegram. Comprueba el ciclo completo: planificar -> avisar -> fichar, el killswitch, el retraso de 15 min, la pausa y la recuperacion de fichajes perdidos. """ import asyncio, json, os, sys, tempfile from datetime import date, datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app import settings, schedule from app.daemon import ANNOUNCE_LEAD, Service from app.planner import ENTRADA ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) CAL = json.load(open(os.path.join(ROOT, "data", "calendar_cache.json"))) OK = FAIL = 0 def check(cond, label): global OK, FAIL print((" ✅ " if cond else " ❌ ") + label) OK, FAIL = OK + bool(cond), FAIL + (not cond) class FakeExecutor: """Sustituye la red por respuestas deterministas.""" def __init__(self): self.cal = schedule.Calendar(CAL) self.checks = [] self.fail_next = False def ensure_login(self, force=False): pass def calendar(self, force=False): return self.cal def balance(self, day=None): return {"P": 9750, "W": 7810, "X": -50, "XAnt": -726, "total": -776} def today_summary(self, day=None): return {"checks": [], "P": 510, "W": 0, "X": 0, "tipo": "H 8"} def do_check(self, typ): if self.fail_next: self.fail_next = False raise RuntimeError("timeout simulado") self.checks.append(typ) return {"typ": typ, "repeated": False, "check_id": 1000 + len(self.checks), "recorded": "09:00", "real": "08:58"} class FakeService(Service): def __init__(self, cfg, db): super().__init__(cfg, pin="0000", db_path=db) self.executor = FakeExecutor() self.clock = datetime(2026, 9, 28, 3, 0, tzinfo=self.tz) # lunes laborable self.sent = [] def now(self): return self.clock async def notify(self, text, keyboard=None): self.sent.append((text, keyboard)) class M: message_id = len(self.sent) return M() async def advance_to(self, when): """Avanza el reloj disparando tick() en cada evento intermedio.""" while self.clock < when: nxt = self._next_event() self.clock = min(when, max(nxt, self.clock + timedelta(seconds=1))) await self.tick() async def main(): cfg = settings.load() cfg["use_google_calendar"] = False # el test no debe salir a la red db = os.path.join(tempfile.mkdtemp(), "t.db") svc = FakeService(cfg, db) day = date(2026, 9, 28) print("\n--- plan del dia ---") await svc.ensure_plan() plan = svc.store.get_plan(day) check(len(plan) == 4, "se planifican 4 fichajes (lunes, 2 tramos)") for r in plan: print(" idx=%s %-7s %s" % (r["idx"], "ENTRADA" if r["typ"] == ENTRADA else "SALIDA", datetime.fromisoformat(r["planned_at"]).strftime("%H:%M:%S"))) p0 = datetime.fromisoformat(plan[0]["planned_at"]) check(p0.hour == 8 and p0.minute >= 55, "la primera entrada cae antes de las 09:00") print("\n--- aviso con antelacion (killswitch) ---") await svc.advance_to(p0 - ANNOUNCE_LEAD + timedelta(seconds=5)) check(any("ENTRADA" in t and "programada" in t for t, _ in svc.sent), "llega el aviso previo") check(any(k is not None for _, k in svc.sent), "el aviso trae teclado inline") check(svc.store.get_check(day, 0)["announced_at"] is not None, "queda marcado como anunciado") check(not svc.executor.checks, "todavia NO se ha fichado") print("\n--- el fichaje se ejecuta a su hora ---") await svc.advance_to(p0 + timedelta(seconds=2)) check(svc.executor.checks == [ENTRADA], "se ficha la entrada y solo esa") check(svc.store.get_check(day, 0)["state"] == "done", "queda en estado done") check(svc.store.get_check(day, 0)["check_id"] == 1001, "guarda el id devuelto por Timenet") print("\n--- killswitch: cancelar el siguiente ---") svc.store.set_state(day, 1, "cancelled", note="prueba") p2 = datetime.fromisoformat(svc.store.get_check(day, 2)["planned_at"]) await svc.advance_to(p2 + timedelta(seconds=2)) check(svc.executor.checks == [ENTRADA, ENTRADA], "el cancelado NO se ficha, el siguiente si") print("\n--- retraso de 15 min ---") svc2 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t2.db")) await svc2.ensure_plan() orig = datetime.fromisoformat(svc2.store.get_check(day, 0)["planned_at"]) svc2.store.reschedule(day, 0, orig + timedelta(minutes=15)) await svc2.advance_to(orig + timedelta(seconds=30)) check(not svc2.executor.checks, "tras retrasarlo no ficha a la hora vieja") check(svc2.store.get_check(day, 0)["announced_at"] is None, "se vuelve a anunciar") await svc2.advance_to(orig + timedelta(minutes=15, seconds=5)) check(svc2.executor.checks == [ENTRADA], "ficha a la hora nueva") print("\n--- pausa global ---") svc3 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t3.db")) await svc3.ensure_plan() svc3.set_paused(True) last = datetime.fromisoformat(svc3.store.get_plan(day)[-1]["planned_at"]) await svc3.advance_to(last + timedelta(minutes=1)) check(not svc3.executor.checks, "en pausa no ficha nada") print("\n--- recuperacion: el equipo estaba apagado ---") svc4 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t4.db")) await svc4.ensure_plan() svc4.clock = datetime(2026, 9, 28, 16, 0, tzinfo=svc4.tz) # arranca tardisimo await svc4.tick() check(len(svc4.executor.checks) == 3, "ficha los 3 que se habian perdido, en orden") check(any("tarde respecto del plan" in t for t, _ in svc4.sent), "avisa de que van tarde") print("\n--- fallo de red al fichar ---") svc5 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t5.db")) await svc5.ensure_plan() svc5.executor.fail_next = True f0 = datetime.fromisoformat(svc5.store.get_check(day, 0)["planned_at"]) await svc5.advance_to(f0 + timedelta(seconds=2)) check(svc5.store.get_check(day, 0)["state"] == "failed", "el fichaje queda marcado failed") check(any("No se pudo fichar" in t for t, _ in svc5.sent), "avisa del fallo por Telegram") print("\n--- dia no laborable ---") svc6 = FakeService(cfg, os.path.join(tempfile.mkdtemp(), "t6.db")) svc6.clock = datetime(2026, 9, 24, 3, 0, tzinfo=svc6.tz) # Festiu Local await svc6.ensure_plan() check(not svc6.store.get_plan(date(2026, 9, 24)), "no planifica en festivo local") print("\n%d OK, %d fallos" % (OK, FAIL)) sys.exit(1 if FAIL else 0) if __name__ == "__main__": asyncio.run(main())