fn_registry/python/functions/infra/drain_fleet_events_test.py

"""Tests para drain_fleet_events."""

import json

from drain_fleet_events import drain_fleet_events


def _write_queue(path, events):
    """Escribe una lista de eventos como JSONL en path."""
    with open(path, "w", encoding="utf-8") as fh:
        for ev in events:
            fh.write(json.dumps(ev) + "\n")


def _make_event(to, urgent=False, session_id="uuid", from_state="TRABAJANDO"):
    return {
        "ts": 1750445000000,
        "session_id": session_id,
        "pid": 1234,
        "from": from_state,
        "to": to,
        "goal": "objetivo de prueba",
        "phase": "preguntando",
        "urgent": urgent,
    }


def test_cola_con_eventos_cursor_cero_devuelve_todos_y_avanza(tmp_path):
    events_path = tmp_path / "events.jsonl"
    cursor_path = tmp_path / "cursor"
    eventos = [
        _make_event("RECLAMA", urgent=True),
        _make_event("DICE_TERMINADO"),
        _make_event("ESTANCADO"),
    ]
    _write_queue(events_path, eventos)

    result = drain_fleet_events(str(events_path), str(cursor_path))

    assert result["total_new"] == 3
    assert len(result["events"]) == 3
    assert result["cursor"] == 3
    assert "reset" not in result
    # El cursor se persistio en disco.
    assert cursor_path.read_text().strip() == "3"


def test_segunda_llamada_cero_nuevos(tmp_path):
    events_path = tmp_path / "events.jsonl"
    cursor_path = tmp_path / "cursor"
    _write_queue(events_path, [_make_event("RECLAMA"), _make_event("ESTANCADO")])

    first = drain_fleet_events(str(events_path), str(cursor_path))
    assert first["total_new"] == 2

    second = drain_fleet_events(str(events_path), str(cursor_path))
    assert second["total_new"] == 0
    assert second["events"] == []
    assert second["by_classification"] == {}
    assert second["urgent"] == []
    assert second["cursor"] == 2


def test_advance_false_no_mueve_cursor(tmp_path):
    events_path = tmp_path / "events.jsonl"
    cursor_path = tmp_path / "cursor"
    _write_queue(events_path, [_make_event("RECLAMA"), _make_event("GONE")])

    peek1 = drain_fleet_events(str(events_path), str(cursor_path), advance=False)
    assert peek1["total_new"] == 2
    # No se creo archivo de cursor.
    assert not cursor_path.exists()

    # Una segunda lectura con peek vuelve a ver los mismos eventos.
    peek2 = drain_fleet_events(str(events_path), str(cursor_path), advance=False)
    assert peek2["total_new"] == 2
    assert not cursor_path.exists()


def test_archivo_ausente_vacio_sin_crash(tmp_path):
    events_path = tmp_path / "no_existe.jsonl"
    cursor_path = tmp_path / "cursor"

    result = drain_fleet_events(str(events_path), str(cursor_path))

    assert result["total_new"] == 0
    assert result["events"] == []
    assert result["by_classification"] == {}
    assert result["urgent"] == []
    assert result["cursor"] == 0


def test_json_invalido_se_salta(tmp_path):
    events_path = tmp_path / "events.jsonl"
    cursor_path = tmp_path / "cursor"
    with open(events_path, "w", encoding="utf-8") as fh:
        fh.write(json.dumps(_make_event("RECLAMA")) + "\n")
        fh.write("esto no es json valido {{{\n")
        fh.write("\n")  # linea en blanco
        fh.write("   \n")  # linea solo espacios
        fh.write(json.dumps(_make_event("ESTANCADO")) + "\n")

    result = drain_fleet_events(str(events_path), str(cursor_path))

    # Solo 2 eventos validos, pero el cursor avanza al total de lineas (5).
    assert result["total_new"] == 2
    assert result["cursor"] == 5
    assert [e["to"] for e in result["events"]] == ["RECLAMA", "ESTANCADO"]


def test_agrupacion_por_to_y_urgent(tmp_path):
    events_path = tmp_path / "events.jsonl"
    cursor_path = tmp_path / "cursor"
    eventos = [
        _make_event("RECLAMA", urgent=True, session_id="a"),
        _make_event("RECLAMA", urgent=False, session_id="b"),
        _make_event("DICE_TERMINADO", urgent=False, session_id="c"),
        _make_event("ESTANCADO", urgent=True, session_id="d"),
    ]
    _write_queue(events_path, eventos)

    result = drain_fleet_events(str(events_path), str(cursor_path))

    bc = result["by_classification"]
    assert set(bc.keys()) == {"RECLAMA", "DICE_TERMINADO", "ESTANCADO"}
    assert len(bc["RECLAMA"]) == 2
    assert len(bc["DICE_TERMINADO"]) == 1
    assert len(bc["ESTANCADO"]) == 1

    assert len(result["urgent"]) == 2
    urgent_sessions = {e["session_id"] for e in result["urgent"]}
    assert urgent_sessions == {"a", "d"}


def test_cursor_mayor_que_lineas_reinicia(tmp_path):
    events_path = tmp_path / "events.jsonl"
    cursor_path = tmp_path / "cursor"
    # Cola con 2 lineas pero cursor apuntando a 99 (cola rotada/truncada).
    _write_queue(events_path, [_make_event("RECLAMA"), _make_event("GONE")])
    cursor_path.write_text("99")

    result = drain_fleet_events(str(events_path), str(cursor_path))

    assert result.get("reset") is True
    assert result["total_new"] == 2
    assert result["cursor"] == 2
    assert cursor_path.read_text().strip() == "2"