49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Tests de `monitor_estado.vistos` — dedup persistente (RF-8, RF-23)."""
|
|
|
|
import monitor_estado.vistos as vistos
|
|
|
|
|
|
def test_vistos_filtra_nuevos(tmp_path, monkeypatch):
|
|
"""Lo ya visto no se vuelve a contar; lo nuevo si (RF-8)."""
|
|
monkeypatch.setattr(vistos, "VISTO_DIR", tmp_path)
|
|
clave = "telegram_test"
|
|
msgs = [{"id": 1, "t": "a"}, {"id": 2, "t": "b"}]
|
|
|
|
nuevos = vistos.filtrar_nuevos(clave, msgs, id_fn=lambda m: m["id"])
|
|
assert [m["id"] for m in nuevos] == [1, 2]
|
|
|
|
# Segundo ciclo con los mismos mensajes -> nada nuevo.
|
|
assert vistos.filtrar_nuevos(clave, msgs, id_fn=lambda m: m["id"]) == []
|
|
|
|
# Llega uno nuevo -> solo ese.
|
|
mas = msgs + [{"id": 3, "t": "c"}]
|
|
assert [m["id"] for m in vistos.filtrar_nuevos(clave, mas, id_fn=lambda m: m["id"])] == [3]
|
|
|
|
|
|
def test_vistos_persiste_entre_llamadas(tmp_path, monkeypatch):
|
|
"""El estado se guarda en disco y sobrevive a una nueva lectura (RF-23)."""
|
|
monkeypatch.setattr(vistos, "VISTO_DIR", tmp_path)
|
|
clave = "correo"
|
|
vistos.filtrar_nuevos(clave, ["m1", "m2"], id_fn=lambda x: x)
|
|
|
|
assert (tmp_path / f"{clave}.json").exists()
|
|
# Nueva lectura del fichero: m1/m2 ya no son nuevos.
|
|
assert vistos.filtrar_nuevos(clave, ["m1", "m2"], id_fn=lambda x: x) == []
|
|
|
|
|
|
def test_vistos_marcar_explicito(tmp_path, monkeypatch):
|
|
"""`marcar` añade ids sin devolver nada."""
|
|
monkeypatch.setattr(vistos, "VISTO_DIR", tmp_path)
|
|
vistos.marcar("jira", ["DS-1", "DS-2"])
|
|
assert vistos.filtrar_nuevos("jira", ["DS-1"], id_fn=lambda x: x) == []
|
|
assert vistos.filtrar_nuevos("jira", ["DS-3"], id_fn=lambda x: x) == ["DS-3"]
|
|
|
|
|
|
def test_vistos_idempotente_sin_marcar(tmp_path, monkeypatch):
|
|
"""Con marcar_automatico=False, filtrar no altera el estado."""
|
|
monkeypatch.setattr(vistos, "VISTO_DIR", tmp_path)
|
|
a = vistos.filtrar_nuevos("k", [1, 2], id_fn=lambda x: x, marcar_automatico=False)
|
|
b = vistos.filtrar_nuevos("k", [1, 2], id_fn=lambda x: x, marcar_automatico=False)
|
|
assert a == [1, 2] and b == [1, 2]
|