esh-pfi-infrastructure/services/booth/tests/test_asks.py

"""Asks: session poses a multiple-choice question; operator answers in the
browser; the answer lands as a sidecar the session reads."""
import json
import pathlib

import pytest
from fastapi.testclient import TestClient

from booth.app import build_gallery, create_app, list_booths
from booth.asks import (
    ANSWER_SUFFIX,
    ASK_SUFFIX,
    AskError,
    list_asks,
    load_ask,
    normalize_ask,
    read_answer,
    write_answer,
    write_ask,
)


def _ask(booth, stem="winner", **kw):
    doc = {"prompt": "Which render wins?", "options": ["A — baseline", "B — async"]}
    doc.update(kw)
    booth.mkdir(parents=True, exist_ok=True)
    (booth / f"{stem}{ASK_SUFFIX}").write_text(json.dumps(doc))
    return booth


@pytest.fixture
def client(tmp_path):
    app = create_app(tmp_path, ttl_hours=24, start_sweeper=False)
    return TestClient(app), tmp_path


# ---- normalisation ----------------------------------------------------------


def test_normalize_string_options():
    a = normalize_ask({"prompt": " Pick ", "options": ["x", "y"]}, "s")
    assert a["prompt"] == "Pick"
    assert a["options"] == [{"id": "x", "label": "x", "detail": ""}, {"id": "y", "label": "y", "detail": ""}]
    assert a["notes"] is True and a["notes_label"] == "notes"


def test_normalize_object_options_and_flags():
    a = normalize_ask(
        {"prompt": "p", "options": [{"id": "a", "label": "A", "detail": "d"}, {"label": "B"}],
         "notes": False, "notes_label": "why"},
        "s",
    )
    assert a["options"][0] == {"id": "a", "label": "A", "detail": "d"}
    assert a["options"][1] == {"id": "B", "label": "B", "detail": ""}
    assert a["notes"] is False and a["notes_label"] == "why"


@pytest.mark.parametrize(
    "doc",
    [
        {"options": ["a", "b"]},
        {"prompt": "", "options": ["a", "b"]},
        {"prompt": "p", "options": ["only"]},
        {"prompt": "p", "options": "a,b"},
        {"prompt": "p", "options": ["a", "a"]},
        {"prompt": "p", "options": [{"id": "a"}, "b"]},
        {"prompt": "p", "options": ["a", "b"], "notes": "yes"},
        [],
    ],
)
def test_normalize_rejects(doc):
    with pytest.raises(AskError):
        normalize_ask(doc, "s")


# ---- files ------------------------------------------------------------------


def test_load_ask_reports_bad_json(tmp_path):
    (tmp_path / f"x{ASK_SUFFIX}").write_text("{not json")
    with pytest.raises(AskError):
        load_ask(tmp_path, "x")
    with pytest.raises(AskError):
        load_ask(tmp_path, "missing")


def test_list_asks_folds_answer_and_surfaces_errors(tmp_path):
    _ask(tmp_path, "one")
    _ask(tmp_path, "two")
    (tmp_path / f"broken{ASK_SUFFIX}").write_text("[]")
    (tmp_path / ".hidden.ask.json").write_text("{}")  # dotfiles never listed
    write_answer(tmp_path, "two", "B — async", "less banding", who="10.0.0.1")

    asks = list_asks(tmp_path)
    by = {a["stem"]: a for a in asks}
    assert set(by) == {"one", "two", "broken"}
    assert by["one"]["answer"] is None and by["one"]["error"] is None
    assert by["two"]["answer"]["choice"] == "B — async"
    assert by["two"]["answer"]["choice_index"] == 1
    assert by["two"]["answer"]["notes"] == "less banding"
    assert by["two"]["answer"]["answered_by"] == "10.0.0.1"
    assert by["broken"]["error"] and by["broken"]["options"] == []


def test_write_answer_validates_choice_and_is_atomic(tmp_path):
    _ask(tmp_path)
    with pytest.raises(AskError):
        write_answer(tmp_path, "winner", "C — nope")
    with pytest.raises(AskError):
        write_answer(tmp_path, "nosuch", "A — baseline")
    ans = write_answer(tmp_path, "winner", "A — baseline", "  ok \r\n")
    assert ans["notes"] == "ok"
    assert ans["answered_at"]
    assert read_answer(tmp_path, "winner") == ans
    assert not (tmp_path / f"winner{ANSWER_SUFFIX}.tmp").exists()
    # re-answer overwrites — the sidecar is the CURRENT answer, not a log
    write_answer(tmp_path, "winner", "B — async")
    assert read_answer(tmp_path, "winner")["choice_index"] == 1


def test_write_answer_drops_notes_when_ask_disables_them(tmp_path):
    _ask(tmp_path, notes=False)
    assert write_answer(tmp_path, "winner", "A — baseline", "ignored")["notes"] == ""


def test_write_ask_roundtrip_and_stem_guard(tmp_path):
    p = write_ask(tmp_path / "b", "pick", "Pick one", ["x", {"id": "y", "label": "Y"}], notes=False)
    assert p.name == f"pick{ASK_SUFFIX}"
    a = load_ask(tmp_path / "b", "pick")
    assert [o["id"] for o in a["options"]] == ["x", "y"] and a["notes"] is False
    for bad in ("../x", ".hidden", "a/b", ""):
        with pytest.raises(AskError):
            write_ask(tmp_path / "b", bad, "p", ["a", "b"])
    with pytest.raises(AskError):
        write_ask(tmp_path / "b", "ok", "p", ["solo"])


# ---- gallery + index integration -------------------------------------------


def test_gallery_hides_ask_and_answer_files(tmp_path):
    b = _ask(tmp_path / "b")
    (b / "a.png").write_bytes(b"x")
    write_answer(b, "winner", "A — baseline")
    names = {it["name"] for it in build_gallery(b)}
    assert names == {"a.png"}


def test_list_booths_counts_open_asks(tmp_path):
    b = _ask(tmp_path / "b", "one")
    _ask(b, "two")
    write_answer(b, "two", "A — baseline")
    (tmp_path / "plain").mkdir()
    by = {x["name"]: x for x in list_booths(tmp_path, 3600)}
    assert by["b"]["asks_open"] == 1 and by["b"]["asks_total"] == 2
    assert by["plain"]["asks_open"] == 0 and by["plain"]["asks_total"] == 0
    assert by["b"]["count"] == 0  # ask/answer files are not "items"


# ---- routes -----------------------------------------------------------------


def test_booth_page_renders_open_ask_as_form(client):
    c, data = client
    _ask(data / "b")
    html = c.get("/b/b/").text
    assert "Which render wins?" in html
    assert 'type="radio"' in html and 'name="choice"' in html
    assert 'value="B — async"' in html
    assert 'action="/b/b/answer"' in html
    assert "<textarea" in html
    assert "1 open ask" in html


def test_answer_route_writes_sidecar_and_page_shows_it(client):
    c, data = client
    _ask(data / "b")
    r = c.post("/b/b/answer", data={"ask": "winner", "choice": "B — async", "notes": "less banding"},
               follow_redirects=False)
    assert r.status_code == 303 and r.headers["location"] == "/b/b/#ask-winner"
    ans = json.loads((data / "b" / f"winner{ANSWER_SUFFIX}").read_text())
    assert ans["choice"] == "B — async" and ans["notes"] == "less banding"
    assert ans["answered_by"]  # TestClient's client addr
    html = c.get("/b/b/").text
    assert "answered" in html and "less banding" in html
    assert "1 open ask" not in html
    # the sidecar is fetchable over HTTP for remote sessions
    assert c.get("/b/b/winner.answer.json").json()["choice"] == "B — async"


def test_answer_route_rejects_bad_choice_and_unknown_ask(client):
    c, data = client
    _ask(data / "b")
    assert c.post("/b/b/answer", data={"ask": "winner", "choice": "Z"}).status_code == 400
    assert c.post("/b/b/answer", data={"ask": "nosuch", "choice": "A — baseline"}).status_code == 404
    assert c.post("/b/b/answer", data={"ask": "../x", "choice": "A — baseline"}).status_code == 404
    assert not (data / "b" / f"winner{ANSWER_SUFFIX}").exists()


def test_answer_json_404s_until_answered(client):
    c, data = client
    _ask(data / "b")
    assert c.get("/b/b/winner.answer.json").status_code == 404


def test_notes_field_hidden_when_disabled(client):
    c, data = client
    _ask(data / "b", notes=False)
    assert "<textarea" not in c.get("/b/b/").text


def test_index_card_shows_open_ask_badge(client):
    c, data = client
    _ask(data / "b")
    html = c.get("/").text
    assert "1 ask" in html