92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests de `monitor_estado.fuentes.telegram` — nucleo y aislamiento (RF-1, RF-6)."""
|
|
|
|
from datetime import date, datetime, timezone
|
|
from types import SimpleNamespace
|
|
|
|
import monitor_estado.fuentes.telegram as tg
|
|
|
|
|
|
class FakeMsg:
|
|
"""Mensaje crudo minimo compatible con lo que espera normalizar/recoger."""
|
|
|
|
def __init__(self, id, dt, text, sender_id=1, sender=None):
|
|
self.id = id
|
|
self.date = dt
|
|
self.message = text
|
|
self.sender_id = sender_id
|
|
self._sender_name = sender
|
|
self._sender = sender or "Anon"
|
|
|
|
async def get_sender(self):
|
|
return SimpleNamespace(first_name=self._sender)
|
|
|
|
|
|
class FakeClient:
|
|
"""Cliente Telethon falso (sin red)."""
|
|
|
|
def __init__(self, por_id, malos=()):
|
|
self.por_id = por_id
|
|
self.malos = set(malos)
|
|
self.conectado = False
|
|
|
|
async def connect(self):
|
|
self.conectado = True
|
|
|
|
async def is_user_authorized(self):
|
|
return True
|
|
|
|
async def disconnect(self):
|
|
self.conectado = False
|
|
|
|
async def get_dialogs(self, limit=None):
|
|
return []
|
|
|
|
async def get_entity(self, cid):
|
|
if cid in self.malos:
|
|
raise RuntimeError("sin acceso a la fuente")
|
|
return cid
|
|
|
|
def iter_messages(self, entidad, limit=None):
|
|
async def gen():
|
|
for m in self.por_id.get(entidad, []):
|
|
yield m
|
|
return gen()
|
|
|
|
|
|
def test_normalizar_filtra_dia_y_ordena():
|
|
dia = date(2026, 10, 3)
|
|
m1 = FakeMsg(2, datetime(2026, 10, 3, 15, 0, tzinfo=timezone.utc), "segundo", sender="Ana")
|
|
m2 = FakeMsg(1, datetime(2026, 10, 3, 9, 0, tzinfo=timezone.utc), "primero", sender="Ana")
|
|
m3 = FakeMsg(3, datetime(2026, 10, 2, 9, 0, tzinfo=timezone.utc), "ayer", sender="Ana")
|
|
out = tg.normalizar_mensajes([m1, m2, m3], dia, "test")
|
|
assert [r["id"] for r in out] == [1, 2] # solo hoy y ordenado
|
|
assert out[0]["quien"] == "Ana" and out[0]["fuente"] == "test"
|
|
|
|
|
|
def test_recoger_fuente_para_en_el_dia():
|
|
dia = date(2026, 10, 3)
|
|
hoy = FakeMsg(5, datetime(2026, 10, 3, 12, 0, tzinfo=timezone.utc), "hoy", sender="Ana")
|
|
ayer = FakeMsg(4, datetime(2026, 10, 2, 12, 0, tzinfo=timezone.utc), "ayer", sender="Ana")
|
|
client = FakeClient({7: [hoy, ayer]})
|
|
import asyncio
|
|
out = asyncio.run(tg.recoger_fuente(client, "oscar", 7, dia))
|
|
assert [r["id"] for r in out] == [5]
|
|
|
|
|
|
def test_recoger_todas_aisla_fuente_caida():
|
|
dia = date(2026, 10, 3)
|
|
ok = FakeMsg(1, datetime(2026, 10, 3, 12, 0, tzinfo=timezone.utc), "hola", sender="Ana")
|
|
cfg = {"telegram": {"fuentes": [{"nombre": "buena", "id": 10}, {"nombre": "mala", "id": 20}]}}
|
|
client = FakeClient({10: [ok]}, malos={20})
|
|
|
|
resultado, errores = tg.recoger_todas(cfg, dia, client_factory=lambda c: client)
|
|
assert "buena" in resultado and len(resultado["buena"]) == 1
|
|
assert len(errores) == 1 and errores[0][0] == "mala"
|
|
|
|
|
|
def test_render_mensaje():
|
|
m = {"fecha": "2026-10-03T09:12:00+02:00", "quien": "Oscar", "texto": "linea1\nlinea2"}
|
|
txt = tg.render_mensaje(m)
|
|
assert "[09:12] Oscar" in txt and "linea1" in txt and "linea2" in txt
|