announcer/tests/test_schedule.py

"""Tester for planlagte sendinger.

Det som er verdt å prøve her er ikke at en melding kan legges i en liste, men
de to tingene som faktisk kan svikte: at lista overlever en omstart, og at en
melding som skulle gått for en time siden *ikke* spilles når maskina kommer
tilbake.
"""

from __future__ import annotations

import json
import sys
import time
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import schedule as sched  # noqa: E402


def mk(tmp_path: Path, **kw) -> sched.Schedule:
    return sched.Schedule(tmp_path / "scheduled.json", **kw)


def add(s: sched.Schedule, *, om: float = 3600, text: str = "melding", **kw):
    """Legger til en melding «om» så mange sekunder."""
    return s.add(at=time.time() + om, text=text, voice="finn", speed=1.0,
                 chime="none", **kw)


# --------------------------------------------------------------------------
# Lagring – grunnen til at modulen finnes
# --------------------------------------------------------------------------


def test_planen_overlever_omstart(tmp_path):
    s = mk(tmp_path)
    item = add(s, om=1800, text="Pizzaen kommer", user="Kari", subject="jwt:7")
    # Ny instans, som etter en restart av containeren
    s2 = mk(tmp_path)
    assert len(s2.items) == 1
    got = s2.items[0]
    assert (got.id, got.text, got.user, got.subject) == (item.id, "Pizzaen kommer", "Kari", "jwt:7")
    assert got.status == "pending"


def test_fila_er_lesbar_json(tmp_path):
    """Poenget med fil framfor database: man skal kunne cat'e den."""
    s = mk(tmp_path)
    add(s, text="Hallo")
    raw = json.loads((tmp_path / "scheduled.json").read_text())
    assert raw["items"][0]["text"] == "Hallo"


def test_ingen_tmp_fil_ligger_igjen(tmp_path):
    s = mk(tmp_path)
    add(s)
    assert not list(tmp_path.glob("*.tmp")), "atomisk skriving skal rydde etter seg"


def test_odelagt_fil_hindrer_ikke_oppstart(tmp_path):
    (tmp_path / "scheduled.json").write_text("{dette er ikke json")
    s = mk(tmp_path)          # skal ikke kaste
    assert s.items == []
    add(s, text="virker likevel")
    assert len(mk(tmp_path).items) == 1


def test_ukjente_felt_i_fila_ignoreres(tmp_path):
    """Så en nyere versjon av fila ikke velter en eldre app."""
    (tmp_path / "scheduled.json").write_text(json.dumps({"items": [{
        "id": "abc", "at": time.time() + 60, "text": "x", "voice": "finn",
        "speed": 1.0, "chime": "none", "noe_nytt": 42,
    }]}))
    s = mk(tmp_path)
    assert len(s.items) == 1 and s.items[0].id == "abc"


def test_tom_mappe_er_greit(tmp_path):
    s = mk(tmp_path / "finnes" / "ikke" / "enda")
    add(s)
    assert len(s.items) == 1


# --------------------------------------------------------------------------
# Validering
# --------------------------------------------------------------------------


def test_tidspunkt_i_fortiden_avvises(tmp_path):
    s = mk(tmp_path)
    with pytest.raises(ValueError, match="passert"):
        add(s, om=-3600)


def test_tidspunkt_rett_bak_oss_godtas(tmp_path):
    """Innenfor nådefristen; ellers ville «send nå» vært en kappløpssituasjon."""
    s = mk(tmp_path, grace=120)
    assert add(s, om=-30) is not None


def test_for_langt_fram_avvises(tmp_path):
    s = mk(tmp_path, max_ahead=30 * 86400)
    with pytest.raises(ValueError, match="dager"):
        add(s, om=400 * 86400)        # typisk tastefeil i årstall


# --------------------------------------------------------------------------
# due(): skal den sendes, eller gikk den glipp av?
# --------------------------------------------------------------------------


def test_ikke_forfalt_blir_liggende(tmp_path):
    s = mk(tmp_path)
    add(s, om=600)
    send, missed = s.due()
    assert (send, missed) == ([], [])


def test_forfalt_innenfor_nadefrist_sendes(tmp_path):
    s = mk(tmp_path, grace=120)
    item = add(s, om=600)
    # Som om klokka har gått fram til litt etter planlagt tid.
    send, missed = s.due(now=item.at + 30)
    assert [i.id for i in send] == [item.id]
    assert missed == []


def test_for_sent_gar_glipp_av_og_spilles_ikke(tmp_path):
    """Kjernen i det Kent valgte: hopp over, men vis det.

    Forløpet er: meldingen ble lagt inn i god tid, maskina var av da den
    skulle gått, og nå er den tilbake en time for sent. «Turneringen starter
    om ti minutter» skal ikke ut på anlegget da."""
    s = mk(tmp_path, grace=120)
    item = add(s, om=600)
    send, missed = s.due(now=item.at + 3600)
    assert send == []
    assert [i.id for i in missed] == [item.id]


def test_due_endrer_ingenting_selv(tmp_path):
    """announcer.py markerer først etter at meldingen faktisk er i kø, så en
    melding ikke kan stå som «sendt» uten å ha blitt det."""
    s = mk(tmp_path)
    item = add(s, om=600)
    s.due(now=item.at + 10)
    assert mk(tmp_path).items[0].status == "pending"


def test_avbrutt_forfaller_ikke(tmp_path):
    s = mk(tmp_path)
    item = add(s, om=600)
    s.cancel(item.id)
    assert s.due(now=item.at + 10) == ([], [])


def test_sendt_forfaller_ikke_igjen(tmp_path):
    s = mk(tmp_path)
    item = add(s, om=600)
    s.mark(item.id, "sent", "kø #4")
    assert s.due(now=item.at + 10) == ([], [])


# --------------------------------------------------------------------------
# Avbryting og markering
# --------------------------------------------------------------------------


def test_avbryting_lagres(tmp_path):
    s = mk(tmp_path)
    item = add(s)
    assert s.cancel(item.id) is not None
    assert mk(tmp_path).items[0].status == "cancelled"


def test_avbryting_to_ganger_gir_none(tmp_path):
    s = mk(tmp_path)
    item = add(s)
    s.cancel(item.id)
    assert s.cancel(item.id) is None


def test_avbryting_av_ukjent_id_gir_none(tmp_path):
    assert mk(tmp_path).cancel("finnes-ikke") is None


def test_mark_lagres(tmp_path):
    s = mk(tmp_path)
    item = add(s)
    s.mark(item.id, "missed", "maskina var av")
    got = mk(tmp_path).items[0]
    assert (got.status, got.note) == ("missed", "maskina var av")


# --------------------------------------------------------------------------
# Visning
# --------------------------------------------------------------------------


def test_snapshot_sorterer_ventende_kronologisk(tmp_path):
    s = mk(tmp_path)
    add(s, om=7200, text="sist")
    add(s, om=600, text="først")
    add(s, om=3600, text="midt")
    assert [i["text"] for i in s.snapshot()["pending"]] == ["først", "midt", "sist"]


def test_snapshot_skiller_ventende_fra_ferdige(tmp_path):
    s = mk(tmp_path)
    a = add(s, om=600, text="venter")
    add(s, om=1200, text="venter også")
    s.mark(a.id, "sent")
    snap = s.snapshot()
    assert [i["text"] for i in snap["pending"]] == ["venter også"]
    assert [i["text"] for i in snap["done"]] == ["venter"]


def test_subject_lekker_ikke_ut_til_klienten(tmp_path):
    """Navnet skal vises, bruker-id-en er intern."""
    s = mk(tmp_path)
    add(s, user="Kari", subject="jwt:7")
    pub = s.snapshot()["pending"][0]
    assert pub["user"] == "Kari"
    assert "subject" not in pub


def test_ferdige_hoper_seg_ikke_opp(tmp_path):
    s = mk(tmp_path)
    for i in range(sched.KEEP_DONE + 20):
        item = add(s, om=600 + i, text=f"nr {i}")
        s.mark(item.id, "sent")
    assert len([i for i in s.items if i.status != "pending"]) == sched.KEEP_DONE