82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests de `monitor_estado.ciclo` — aislamiento de fuentes y ciclo completo (RF-6, RF-9)."""
|
|
|
|
from datetime import date
|
|
|
|
import monitor_estado.archivo as archivo
|
|
import monitor_estado.ciclo as ciclo
|
|
import monitor_estado.vistos as vistos
|
|
|
|
|
|
class _EmptyJira:
|
|
def search(self, jql, fields=None, max_results=50):
|
|
return {"issues": []}
|
|
|
|
|
|
def _preparar_tmp(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(archivo, "MONITOR_DIR", tmp_path)
|
|
monkeypatch.setattr(vistos, "VISTO_DIR", tmp_path / "visto")
|
|
|
|
|
|
def test_aislamiento_fuente_fallida_continua(tmp_path, monkeypatch):
|
|
"""Si todas las fuentes fallan, el ciclo termina y registra los errores (RF-6)."""
|
|
_preparar_tmp(tmp_path, monkeypatch)
|
|
|
|
def bad_tg(cfg, hoy):
|
|
raise RuntimeError("telegram caido")
|
|
|
|
def bad_correo(horas=24, limite=50):
|
|
raise RuntimeError("gmail caido")
|
|
|
|
class BadJira:
|
|
def search(self, *a, **k):
|
|
raise RuntimeError("jira caido")
|
|
|
|
res = ciclo.ejecutar(
|
|
cfg={"retencion_dias": 30, "correo": {"horas": 24}},
|
|
deps={
|
|
"telegram_recoger": bad_tg,
|
|
"correo_recoger": bad_correo,
|
|
"jira_cliente": BadJira(),
|
|
"ia": lambda p: '{"resumen_md": "x", "requiere_atencion": []}',
|
|
"avisar": lambda r, cfg=None, _runner=None: False,
|
|
},
|
|
)
|
|
assert len(res["errores"]) == 3
|
|
assert res["fallo_ia"] is False
|
|
assert (tmp_path / "ultimo_ciclo.json").exists()
|
|
|
|
|
|
def test_ciclo_archiva_y_avisa(tmp_path, monkeypatch):
|
|
"""Un ciclo con novedad archiva el dia y avisa si hay atencion (RF-9, RF-14)."""
|
|
_preparar_tmp(tmp_path, monkeypatch)
|
|
emitidos = []
|
|
|
|
def tg(cfg, hoy):
|
|
return ({"oscar_rodriguez": [{"id": 1, "fecha": "2026-10-03T09:00:00", "quien": "Oscar", "texto": "incidencia"}]}, [])
|
|
|
|
def correo(horas=24, limite=50):
|
|
return []
|
|
|
|
def ia(prompt):
|
|
return '{"resumen_md": "hay incidencia", "requiere_atencion": [{"fuente": "telegram", "texto": "incidencia"}]}'
|
|
|
|
def avisar(res, cfg=None, _runner=None):
|
|
emitidos.append(res.get("requiere_atencion"))
|
|
return True
|
|
|
|
res = ciclo.ejecutar(
|
|
cfg={"retencion_dias": 30, "correo": {"horas": 24}},
|
|
hoy=date(2026, 10, 3),
|
|
deps={
|
|
"telegram_recoger": tg,
|
|
"correo_recoger": correo,
|
|
"jira_cliente": _EmptyJira(),
|
|
"ia": ia,
|
|
"avisar": avisar,
|
|
},
|
|
)
|
|
assert (tmp_path / "telegram" / "oscar_rodriguez" / "2026-10-03.md").exists()
|
|
assert res["aviso_emitido"] is True
|
|
assert emitidos and emitidos[0]
|