announcer/tests/test_routes.py

"""Tester for HTTP-laget i announcer.py: rutene, køen og planleggeren.

Fram til nå var alt dette uprøvd, fordi `import announcer` startet tråder og
dro inn Piper. Nå starter trådene først i main(), og syntese og avspilling
byttes ut her – så kan køen kjøres ett steg om gangen med run_once().
"""

from __future__ import annotations

import json
import os
import sys
import tempfile
import threading
import time
import wave
from pathlib import Path

import jwt as pyjwt
import pytest

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

# Må settes før import: announcer.py leser konfigurasjonen på modulnivå.
_TMP = Path(tempfile.mkdtemp(prefix="announcer-routes-"))
for _k, _v in {
    "ANNOUNCER_CONFIG": "/finnes-ikke.toml",
    "ANNOUNCER_CACHE": _TMP / "cache",
    "ANNOUNCER_LOG": _TMP / "logs" / "a.jsonl",
    "ANNOUNCER_STATE": _TMP / "instance",
    "ANNOUNCER_SCHEDULE": _TMP / "scheduled.json",
    "ANNOUNCER_VOICES": _TMP / "voices",
    "ANNOUNCER_CHIMES": _TMP / "chimes",
}.items():
    os.environ[_k] = str(_v)

import announcer  # noqa: E402
import auth  # noqa: E402
import config  # noqa: E402
import schedule as sched  # noqa: E402

SECRET = "test-hemmelighet-som-er-lang-nok-for-hmac-sha256-ok"


def tiny_wav(path: Path) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    with wave.open(str(path), "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(8000)
        w.writeframes(b"\x00\x00" * 80)
    return path


@pytest.fixture
def app(tmp_path, monkeypatch):
    """Frisk kø, frisk plan, falsk syntese og «avspilling» som bare lykkes."""
    monkeypatch.setattr(announcer, "LOG_PATH", tmp_path / "a.jsonl")
    monkeypatch.setattr(announcer, "CACHE_DIR", tmp_path / "cache")
    monkeypatch.setattr(announcer, "schedule", sched.Schedule(tmp_path / "s.json", grace=120))
    monkeypatch.setattr(announcer, "player", announcer.Player(on_finished=announcer._scheduled_finished))
    monkeypatch.setattr(announcer, "synthesize",
                        lambda text, voice, speed, chime: tiny_wav(tmp_path / "cache" / "x.wav"))
    monkeypatch.setattr(announcer, "player_cmd", lambda path: ["true"])
    auth.reconfigure(config.load("/finnes-ikke.toml"))
    announcer.app.config["TESTING"] = True
    return announcer.app


@pytest.fixture
def c(app):
    return app.test_client()


def audit_lines(tmp_path) -> list[dict]:
    p = tmp_path / "a.jsonl"
    return [json.loads(l) for l in p.read_text().splitlines()] if p.exists() else []


def with_auth(tmp_path, toml: str) -> None:
    p = tmp_path / "c.toml"
    p.write_text(f'[paths]\nstate = "{tmp_path}"\n\n{toml}')
    auth.reconfigure(config.load(p))


def token(sub: str, name: str, roles: list[str]) -> dict:
    t = pyjwt.encode({"iss": "https://crew.test", "aud": "announcer", "sub": sub,
                      "name": name, "roles": roles, "exp": time.time() + 600},
                     SECRET, algorithm="HS256")
    return {"Authorization": f"Bearer {t}"}


JWT_CFG = f'[auth.jwt]\nsecret = "{SECRET}"\naudience = "announcer"\nissuer = "https://crew.test"\n'


# --------------------------------------------------------------------------
# Grunnleggende
# --------------------------------------------------------------------------


def test_sida_og_konfig(c):
    assert c.get("/").status_code == 200
    d = c.get("/api/config").get_json()
    assert d["app"]["title"] == "Announcer"
    assert d["auth"]["enabled"] is False
    assert "schedule" in d["auth"]["can"]


def test_ukjent_api_rute_svarer_json(c):
    r = c.get("/api/finnes-ikke")
    assert r.status_code == 404
    assert r.get_json()["error"] == "Finnes ikke"


@pytest.mark.parametrize("body, feil", [
    ({}, "Tom melding"),
    ({"text": "   "}, "Tom melding"),
    ({"text": "x" * 601}, "Maks 600 tegn"),
    ({"text": "hei", "voice": "finnes-ikke"}, "Ukjent stemme"),
    ({"text": "hei", "chime": "finnes-ikke"}, "Ukjent jingle"),
])
def test_ugyldig_melding_gir_400(c, body, feil):
    r = c.post("/api/announce", json=body)
    assert r.status_code == 400
    assert r.get_json()["error"] == feil


def test_hastighet_klemmes_inn(c):
    c.post("/api/announce", json={"text": "fort", "speed": 9})
    c.post("/api/announce", json={"text": "sakte", "speed": 0.1})
    assert [i.speed for i in announcer.player.history] == [1.6, 0.6]


def test_whitespace_normaliseres(c):
    c.post("/api/announce", json={"text": "  hei   på\n\ndeg  "})
    assert announcer.player.history[0].text == "hei på deg"


# --------------------------------------------------------------------------
# Køen
# --------------------------------------------------------------------------


def test_announce_legger_i_ko_og_spiller(c, tmp_path):
    r = c.post("/api/announce", json={"text": "Hallo", "chime": "none"})
    assert r.status_code == 200
    assert r.get_json()["status"] == "queued"

    item = announcer.player.run_once(timeout=1)
    assert item.status == "done"
    actions = [l["action"] for l in audit_lines(tmp_path)]
    assert actions == ["announce", "played"]
    assert audit_lines(tmp_path)[1]["status"] == "done"


def test_feilet_avspilling_markeres(c, monkeypatch):
    monkeypatch.setattr(announcer, "player_cmd", lambda path: ["false"])
    c.post("/api/announce", json={"text": "x"})
    assert announcer.player.run_once(timeout=1).status == "failed"


def test_syntesefeil_markeres(c, monkeypatch):
    def boom(*a):
        raise RuntimeError("edge-tts nede")
    monkeypatch.setattr(announcer, "synthesize", boom)
    c.post("/api/announce", json={"text": "x"})
    item = announcer.player.run_once(timeout=1)
    assert item.status == "failed"
    assert "edge-tts nede" in item.error


def test_stopp_under_syntese_hindrer_avspilling(c, monkeypatch):
    """Bug nr. 1 fra gjennomgangen: Stopp gjorde ingenting før lyden var
    laget, og meldingen gikk ut likevel. Nå skal den aldri nå avspilling."""
    spilt = []
    def synth_som_blir_stoppet(text, voice, speed, chime):
        announcer.player.stop()          # noen trykker Stopp midt i syntesen
        return tiny_wav(announcer.CACHE_DIR / "x.wav")
    monkeypatch.setattr(announcer, "synthesize", synth_som_blir_stoppet)
    monkeypatch.setattr(announcer, "player_cmd", lambda p: spilt.append(p) or ["true"])

    c.post("/api/announce", json={"text": "feil melding"})
    item = announcer.player.run_once(timeout=1)
    assert item.status == "stopped"
    assert spilt == [], "avspilling skulle aldri startet"


def test_stopp_under_avspilling(c, monkeypatch):
    monkeypatch.setattr(announcer, "player_cmd", lambda p: ["sleep", "5"])
    c.post("/api/announce", json={"text": "lang"})
    t = threading.Thread(target=announcer.player.run_once, kwargs={"timeout": 1})
    t.start()
    for _ in range(50):                   # vent til prosessen faktisk kjører
        if announcer.player.proc is not None:
            break
        time.sleep(0.02)
    r = c.post("/api/stop")
    t.join(timeout=3)
    assert r.get_json() == {"ok": True}
    assert announcer.player.history[0].status == "stopped"
    assert not t.is_alive(), "sleep 5 skulle vært drept"


def test_tom_ko(c):
    for i in range(3):
        c.post("/api/announce", json={"text": f"m{i}"})
    assert c.post("/api/clear").status_code == 200
    assert [i.status for i in announcer.player.history] == ["stopped"] * 3
    assert announcer.player.run_once(timeout=0.2) is None, "køen skal være tom"
    assert c.get("/api/status").get_json()["queued"] == []


def test_status_viser_ko_og_historikk(c):
    c.post("/api/announce", json={"text": "en"})
    c.post("/api/announce", json={"text": "to"})
    announcer.player.run_once(timeout=1)
    s = c.get("/api/status").get_json()
    assert s["current"] is None
    assert [i["text"] for i in s["queued"]] == ["to"]
    assert [i["text"] for i in s["history"]] == ["en"]
    assert "scheduled" in s and "now" in s


def test_ip_vises_bare_for_den_som_kan_tomme_koen(c, tmp_path):
    with_auth(tmp_path, JWT_CFG)
    crew = token("7", "Kari", ["crew"])
    adm = token("1", "Kent", ["admin"])
    c.post("/api/announce", json={"text": "x"}, headers=crew)
    assert c.get("/api/status", headers=crew).get_json()["queued"][0]["sender"] == ""
    assert c.get("/api/status", headers=adm).get_json()["queued"][0]["sender"] == "127.0.0.1"


def test_401_i_pin_modus_sier_logg_inn(c, tmp_path):
    with_auth(tmp_path, '[auth.pin]\npin = "1234"\n')
    r = c.post("/api/announce", json={"text": "x"})
    assert r.status_code == 401
    assert r.get_json() == {"error": "Feil eller manglende PIN.", "login": True}
    assert c.get("/api/status").status_code == 401


# --------------------------------------------------------------------------
# Planlagte sendinger gjennom HTTP og planleggeren
# --------------------------------------------------------------------------


def test_planlegging_og_fyring(c, tmp_path):
    at = time.time() + 600
    r = c.post("/api/schedule", json={"text": "Senere", "chime": "none", "at": at})
    assert r.status_code == 200
    sid = r.get_json()["id"]
    assert [i["id"] for i in c.get("/api/status").get_json()["scheduled"]["pending"]] == [sid]

    # Ikke forfalt: ingenting skjer
    announcer.scheduler_tick()
    assert announcer.player.history == []

    # Skru klokka fram: forfalt, innenfor nådefristen
    announcer.schedule.get(sid).at = time.time() - 10
    announcer.scheduler_tick()
    assert announcer.schedule.get(sid).status == "queued"
    assert announcer.player.history[0].schedule_id == sid

    # Bug nr. 2: statusen skal følge avspillingen helt i mål
    announcer.player.run_once(timeout=1)
    assert announcer.schedule.get(sid).status == "sent"
    assert [l["action"] for l in audit_lines(tmp_path)] == ["schedule", "scheduled_fire", "played"]


def test_planlagt_som_feiler_i_avspilling_star_som_feilet(c, monkeypatch):
    monkeypatch.setattr(announcer, "player_cmd", lambda p: ["false"])
    sid = c.post("/api/schedule", json={"text": "x", "at": time.time() + 600}).get_json()["id"]
    announcer.schedule.get(sid).at = time.time() - 10
    announcer.scheduler_tick()
    announcer.player.run_once(timeout=1)
    assert announcer.schedule.get(sid).status == "failed"


def test_planlagt_som_gikk_glipp_av_spilles_ikke(c, tmp_path):
    sid = c.post("/api/schedule", json={"text": "for sent", "at": time.time() + 600}).get_json()["id"]
    announcer.schedule.get(sid).at = time.time() - 3600
    announcer.scheduler_tick()
    assert announcer.schedule.get(sid).status == "missed"
    assert announcer.player.history == []
    assert [l["action"] for l in audit_lines(tmp_path)] == ["schedule", "missed"]


@pytest.mark.parametrize("body, feil", [
    ({"text": "x"}, "Mangler eller ugyldig tidspunkt"),
    ({"text": "x", "at": "snart"}, "Mangler eller ugyldig tidspunkt"),
    ({"text": "x", "at": 1}, "Tidspunktet er passert"),
    ({"text": "x", "at": time.time() + 400 * 86400}, "Mer enn 30 dager fram i tid"),
    ({"text": "", "at": time.time() + 60}, "Tom melding"),
])
def test_ugyldig_planlegging(c, body, feil):
    r = c.post("/api/schedule", json=body)
    assert r.status_code == 400
    assert r.get_json()["error"] == feil


def test_avbryt_egen_planlagt(c, tmp_path):
    sid = c.post("/api/schedule", json={"text": "x", "at": time.time() + 600}).get_json()["id"]
    assert c.post("/api/schedule/cancel", json={"id": sid}).status_code == 200
    assert announcer.schedule.get(sid).status == "cancelled"
    assert c.post("/api/schedule/cancel", json={"id": sid}).status_code == 404
    assert c.post("/api/schedule/cancel", json={"id": "nope"}).status_code == 404


def test_andres_planlagte_krever_clear(c, tmp_path):
    with_auth(tmp_path, JWT_CFG)
    kari = token("7", "Kari", ["crew"])
    ola = token("8", "Ola", ["crew"])
    kent = token("1", "Kent", ["admin"])
    sid = c.post("/api/schedule", json={"text": "x", "at": time.time() + 600}, headers=kari).get_json()["id"]

    r = c.post("/api/schedule/cancel", json={"id": sid}, headers=ola)
    assert r.status_code == 403
    assert "Kari" in r.get_json()["error"]
    assert c.post("/api/schedule/cancel", json={"id": sid}, headers=kari).status_code == 200

    sid2 = c.post("/api/schedule", json={"text": "y", "at": time.time() + 600}, headers=kari).get_json()["id"]
    assert c.post("/api/schedule/cancel", json={"id": sid2}, headers=kent).status_code == 200


def test_schedule_krever_rettigheten(c, tmp_path):
    with_auth(tmp_path, JWT_CFG + '\n[roles]\nlytter = ["view", "announce"]\n')
    h = token("9", "Lytter", ["lytter"])
    assert c.post("/api/announce", json={"text": "x"}, headers=h).status_code == 200
    assert c.post("/api/schedule", json={"text": "x", "at": time.time() + 60}, headers=h).status_code == 403


# --------------------------------------------------------------------------
# Oppstart
# --------------------------------------------------------------------------


def test_uskrivbar_datamappe_gir_klar_melding(tmp_path, monkeypatch):
    locked = tmp_path / "laast"
    locked.mkdir()
    locked.chmod(0o500)
    monkeypatch.setattr(announcer, "CACHE_DIR", locked / "cache")
    with pytest.raises(SystemExit) as e:
        announcer.check_writable()
    assert "PUID" in str(e.value) and "chown" in str(e.value)
    locked.chmod(0o700)