chat_grid/server/tests/test_item_use_cooldown.py

from __future__ import annotations

import json
from typing import cast

import pytest
from websockets.asyncio.server import ServerConnection

from app.server import ClientConnection, SignalingServer


def _fake_ws() -> ServerConnection:
    return cast(ServerConnection, object())


@pytest.mark.asyncio
async def test_item_use_has_global_cooldown(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "dice")
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    now_ms = 10_000

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        return

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
    monkeypatch.setattr(server.item_service, "now_ms", lambda: now_ms)

    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert send_payloads[-1].ok is True

    now_ms += 400
    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert send_payloads[-1].ok is False
    assert "cooldown" in send_payloads[-1].message.lower()

    now_ms += 700
    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert send_payloads[-1].ok is True


@pytest.mark.asyncio
async def test_radio_use_toggles_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "radio_station")
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    broadcast_payloads: list[object] = []
    now_ms = 20_000

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        broadcast_payloads.append(packet)

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
    monkeypatch.setattr(server.item_service, "now_ms", lambda: now_ms)

    assert item.params.get("enabled") is True
    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert item.params.get("enabled") is False
    assert send_payloads[-1].ok is True

    now_ms += 1200
    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert item.params.get("enabled") is True
    assert send_payloads[-1].ok is True

    assert any(getattr(packet, "type", "") == "item_upsert" for packet in broadcast_payloads)


@pytest.mark.asyncio
async def test_radio_media_fields_update_validate(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "radio_station")
    server.item_service.add_item(item)

    send_payloads: list[object] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        return

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"mediaChannel": "left"}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("mediaChannel") == "left"

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"mediaChannel": "invalid"}}),
    )
    assert send_payloads[-1].ok is False
    assert "mediachannel must be one of" in send_payloads[-1].message.lower()

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"facing": 270}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("facing") == 270

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"facing": 361}}),
    )
    assert send_payloads[-1].ok is False
    assert "facing must be between 0 and 360" in send_payloads[-1].message.lower()

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"mediaVolume": 12}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("mediaVolume") == 12

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"mediaEffect": "echo"}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("mediaEffect") == "echo"

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"emitRange": 12}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("emitRange") == 12

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"emitRange": 4}}),
    )
    assert send_payloads[-1].ok is False
    assert "emitrange must be between 5 and 20" in send_payloads[-1].message.lower()


@pytest.mark.asyncio
async def test_clock_use_reports_time_without_use_sound_packet(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "clock")
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    broadcast_payloads: list[object] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        broadcast_payloads.append(packet)

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
    monkeypatch.setattr(server.item_service, "now_ms", lambda: 30_000)
    monkeypatch.setattr(server, "_format_clock_display_time", lambda _params: "2:15 PM")

    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))

    assert send_payloads[-1].ok is True
    assert send_payloads[-1].message == f"{item.title} says 2:15 PM."
    assert not any(getattr(packet, "type", "") == "item_use_sound" for packet in broadcast_payloads)


@pytest.mark.asyncio
async def test_clock_timezone_update_validates(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "clock")
    server.item_service.add_item(item)

    send_payloads: list[object] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        return

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"timeZone": "Europe/Berlin"}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("timeZone") == "Europe/Berlin"

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"timeZone": "Invalid/Zone"}}),
    )
    assert send_payloads[-1].ok is False
    assert "timezone must be one of" in send_payloads[-1].message.lower()


@pytest.mark.asyncio
async def test_failed_wheel_use_does_not_consume_cooldown(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "wheel")
    item.params["spaces"] = ",,,"
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    now_ms = 40_000

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        return

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
    monkeypatch.setattr(server.item_service, "now_ms", lambda: now_ms)

    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert send_payloads[-1].ok is False
    assert "spaces" in send_payloads[-1].message.lower()

    item.params["spaces"] = "a,b,c"
    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert send_payloads[-1].ok is True


@pytest.mark.asyncio
async def test_widget_update_and_use(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "widget")
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    broadcast_payloads: list[object] = []
    now_ms = 50_000

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        broadcast_payloads.append(packet)

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
    monkeypatch.setattr(server.item_service, "now_ms", lambda: now_ms)

    await server._handle_message(
        client,
        json.dumps(
            {
                "type": "item_update",
                "itemId": item.id,
                "params": {
                    "directional": True,
                    "facing": 123.4,
                    "emitRange": 7,
                    "emitVolume": 42,
                    "emitSoundSpeed": 25,
                    "emitSoundTempo": 60,
                    "emitEffect": "reverb",
                    "emitEffectValue": 63.2,
                    "useSound": "ping.ogg",
                    "emitSound": "https://example.com/ambient.ogg",
                },
            }
        ),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("directional") is True
    assert item.params.get("facing") == 123
    assert item.params.get("emitRange") == 7
    assert item.params.get("emitVolume") == 42
    assert item.params.get("emitSoundSpeed") == 25
    assert item.params.get("emitSoundTempo") == 60
    assert item.params.get("emitEffect") == "reverb"
    assert item.params.get("emitEffectValue") == 63.2
    assert item.params.get("useSound") == "sounds/ping.ogg"
    assert item.params.get("emitSound") == "https://example.com/ambient.ogg"

    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert send_payloads[-1].ok is True
    assert item.params.get("enabled") is False
    assert any(getattr(packet, "type", "") == "item_use_sound" for packet in broadcast_payloads)

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"emitRange": 21}}),
    )
    assert send_payloads[-1].ok is False
    assert "emitrange must be between 1 and 20" in send_payloads[-1].message.lower()

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"emitSoundSpeed": 101}}),
    )
    assert send_payloads[-1].ok is False
    assert "emitsoundspeed must be between 0 and 100" in send_payloads[-1].message.lower()

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"emitSoundTempo": 101}}),
    )
    assert send_payloads[-1].ok is False
    assert "emitsoundtempo must be between 0 and 100" in send_payloads[-1].message.lower()


@pytest.mark.asyncio
async def test_piano_update_and_use(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "piano")
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    broadcast_payloads: list[object] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        broadcast_payloads.append(packet)

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)

    await server._handle_message(
        client,
        json.dumps(
            {
                "type": "item_update",
                "itemId": item.id,
                "params": {
                    "instrument": "drum_kit",
                    "emitRange": 12,
                },
            }
        ),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("instrument") == "drum_kit"
    assert item.params.get("voiceMode") == "poly"
    assert item.params.get("octave") == 0
    assert item.params.get("attack") == 1
    assert item.params.get("decay") == 22
    assert item.params.get("release") == 12
    assert item.params.get("brightness") == 68
    assert item.params.get("emitRange") == 12

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"instrument": "nintendo"}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("instrument") == "nintendo"
    assert item.params.get("voiceMode") == "poly"
    assert item.params.get("octave") == 0
    assert item.params.get("attack") == 1
    assert item.params.get("decay") == 24
    assert item.params.get("release") == 15
    assert item.params.get("brightness") == 85

    await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
    assert send_payloads[-1].ok is True
    assert "begin playing" in send_payloads[-1].message.lower()
    assert not any(getattr(packet, "type", "") == "item_use_sound" for packet in broadcast_payloads)

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"instrument": "banjo"}}),
    )
    assert send_payloads[-1].ok is False
    assert "instrument must be one of" in send_payloads[-1].message.lower()

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"voiceMode": "mono", "octave": -2}}),
    )
    assert send_payloads[-1].ok is True
    assert item.params.get("voiceMode") == "mono"
    assert item.params.get("octave") == -2

    await server._handle_message(
        client,
        json.dumps({"type": "item_update", "itemId": item.id, "params": {"octave": 3}}),
    )
    assert send_payloads[-1].ok is False
    assert "octave must be between -2 and 2" in send_payloads[-1].message.lower()


@pytest.mark.asyncio
async def test_piano_note_packet_broadcasts(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws_sender = _fake_ws()
    sender = ClientConnection(websocket=ws_sender, id="u1", nickname="tester", x=5, y=6)
    ws_other = _fake_ws()
    other = ClientConnection(websocket=ws_other, id="u2", nickname="listener", x=7, y=6)
    server.clients[ws_sender] = sender
    server.clients[ws_other] = other
    item = server.item_service.default_item(sender, "piano")
    item.params["instrument"] = "organ"
    item.params["attack"] = 20
    item.params["decay"] = 60
    item.params["emitRange"] = 12
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    broadcast_payloads: list[object] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        broadcast_payloads.append(packet)

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)

    await server._handle_message(
        sender,
        json.dumps({"type": "item_piano_note", "itemId": item.id, "keyId": "KeyA", "midi": 60, "on": True}),
    )

    assert not send_payloads
    assert broadcast_payloads
    packet = broadcast_payloads[-1]
    assert getattr(packet, "type", "") == "item_piano_note"
    assert getattr(packet, "itemId", "") == item.id
    assert getattr(packet, "instrument", "") == "organ"
    assert getattr(packet, "voiceMode", "") == "poly"
    assert getattr(packet, "octave", 999) == 0
    assert getattr(packet, "attack", -1) == 20
    assert getattr(packet, "decay", -1) == 60
    assert getattr(packet, "release", -1) == 35
    assert getattr(packet, "brightness", -1) == 55
    assert getattr(packet, "emitRange", -1) == 12


@pytest.mark.asyncio
async def test_piano_note_key_cap(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws_sender = _fake_ws()
    sender = ClientConnection(websocket=ws_sender, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws_sender] = sender
    item = server.item_service.default_item(sender, "piano")
    server.item_service.add_item(item)

    broadcast_payloads: list[object] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        return

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        broadcast_payloads.append(packet)

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)

    for index in range(12):
        await server._handle_message(
            sender,
            json.dumps({"type": "item_piano_note", "itemId": item.id, "keyId": f"Key{index}", "midi": 60, "on": True}),
        )
    assert len(broadcast_payloads) == 12

    # 13th distinct held key is dropped by cap.
    await server._handle_message(
        sender,
        json.dumps({"type": "item_piano_note", "itemId": item.id, "keyId": "KeyOverflow", "midi": 60, "on": True}),
    )
    assert len(broadcast_payloads) == 12


@pytest.mark.asyncio
async def test_piano_recording_toggle_and_save(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "piano")
    server.item_service.add_item(item)

    send_payloads: list[object] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        return

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)

    await server._handle_message(
        client,
        json.dumps({"type": "item_piano_recording", "itemId": item.id, "action": "toggle_record"}),
    )
    assert send_payloads[-1].ok is True
    assert item.id in server.piano_recording_state_by_item

    await server._handle_message(
        client,
        json.dumps({"type": "item_piano_note", "itemId": item.id, "keyId": "KeyA", "midi": 60, "on": True}),
    )
    await server._handle_message(
        client,
        json.dumps({"type": "item_piano_note", "itemId": item.id, "keyId": "KeyA", "midi": 60, "on": False}),
    )
    await server._handle_message(
        client,
        json.dumps({"type": "item_piano_recording", "itemId": item.id, "action": "toggle_record"}),
    )
    assert send_payloads[-1].ok is True
    assert item.id not in server.piano_recording_state_by_item
    song_id = item.params.get("songId")
    assert isinstance(song_id, str)
    payload = server.item_service.piano_songs.get(song_id)
    assert isinstance(payload, dict)
    keys = payload.get("keys")
    events = payload.get("events")
    assert isinstance(keys, list) and "KeyA" in keys
    assert isinstance(events, list) and len(events) >= 2


@pytest.mark.asyncio
async def test_piano_playback_starts_task(monkeypatch: pytest.MonkeyPatch) -> None:
    server = SignalingServer("127.0.0.1", 8765, None, None)
    ws = _fake_ws()
    client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
    server.clients[ws] = client
    item = server.item_service.default_item(client, "piano")
    item.params["recording"] = [{"t": 0, "keyId": "KeyA", "midi": 60, "on": True}]
    server.item_service.add_item(item)

    send_payloads: list[object] = []
    playback_started: list[str] = []

    async def fake_send(websocket: ServerConnection, packet: object) -> None:
        send_payloads.append(packet)

    async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
        return

    async def fake_start_playback(current_item) -> None:
        playback_started.append(current_item.id)

    monkeypatch.setattr(server, "_send", fake_send)
    monkeypatch.setattr(server, "_broadcast", fake_broadcast)
    monkeypatch.setattr(server, "_start_piano_playback", fake_start_playback)

    await server._handle_message(
        client,
        json.dumps({"type": "item_piano_recording", "itemId": item.id, "action": "playback"}),
    )
    assert send_payloads[-1].ok is True
    task = server.piano_playback_tasks_by_item.get(item.id)
    assert task is not None
    await task
    assert playback_started == [item.id]