openttd-client/tests/test_events.py

import asyncio
import json

import pytest
from openttd import OpenTTDAdminClient
from openttd.protocol import GameEventType, PacketAdminType


class MockTransport:
    def is_closing(self):
        return False
    def close(self):
        pass
    def write(self, data):
        return len(data)

class MockProtocol:
    def __init__(self):
        self.sent = []
    async def send_packet(self, data):
        self.sent.append(data)
        return len(data)

def decode_gamescript_payload(packet):
    """Decode the JSON payload of an AdminGamescript packet (2-byte length + 1-byte type + string)."""
    return json.loads(packet[3:].split(b"\x00")[0])

def make_admin(subscribed=True, **kwargs):
    """An admin client wired to mock transports, with the Gamescript auto-subscribe already done
    unless a test wants to observe it."""
    client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin", **kwargs)
    client._protocol = MockProtocol()
    client._transport = MockTransport()
    client._gs_subscribed = subscribed
    return client

def events_packet(*events):
    """The envelope the AdminBridge GameScript pushes event batches in."""
    return {"command": "events", "events": list(events)}

def arrival(vehicle_id=7, station_id=3, tick=100):
    return {"event": "vehicle_arrive", "tick": tick, "vehicle_id": vehicle_id,
            "station_id": station_id, "owner": 0, "vehicle_type": 0, "order_position": 0,
            "cargo": [{"cargo_id": 0, "load": 12}]}


# --- Event kinds ---

def test_event_type_values_are_the_wire_strings():
    assert GameEventType.VehicleArrive == "vehicle_arrive"
    assert GameEventType.CargoWaiting == "cargo_waiting"
    # Members are plain strings, so they can be used interchangeably with literals.
    assert str(GameEventType.VehicleDepart) == "vehicle_depart"


# --- Subscribing ---

@pytest.mark.asyncio
async def test_subscribe_events_defaults_send_only_the_command():
    client = make_admin(subscribed=False)

    task = asyncio.ensure_future(client.subscribe_events())
    await asyncio.sleep(0)

    # First use auto-subscribes to Gamescript updates, then sends the request.
    assert len(client._protocol.sent) == 2
    assert client._protocol.sent[0][2] == PacketAdminType.AdminUpdateFrequency
    assert decode_gamescript_payload(client._protocol.sent[1]) == {
        "command": "subscribe_events", "request_id": 1,
    }

    reply = {"command": "subscribe_events", "request_id": 1,
             "events": ["vehicle_arrive", "vehicle_depart"], "interval": 10}
    await client.receive_ServerGamescript(None, data=reply)
    assert await task == reply
    assert client._gs_futures == {}

@pytest.mark.asyncio
async def test_subscribe_events_encodes_every_filter():
    client = make_admin()

    task = asyncio.ensure_future(client.subscribe_events(
        events=[GameEventType.VehicleArrive, "cargo_waiting"], interval=25, company_id=0,
        vehicles=(7, 9), stations=(3,), cargo=[0, 1], min_cargo_delta=5, include_cargo=False))
    await asyncio.sleep(0)

    assert decode_gamescript_payload(client._protocol.sent[0]) == {
        "command": "subscribe_events", "request_id": 1,
        "events": ["vehicle_arrive", "cargo_waiting"], "interval": 25, "company_id": 0,
        "vehicles": [7, 9], "stations": [3], "cargo": [0, 1],
        "min_cargo_delta": 5, "include_cargo": False,
    }
    await client.receive_ServerGamescript(
        None, data={"request_id": 1, "events": ["vehicle_arrive", "cargo_waiting"]})
    assert (await task)["events"] == ["vehicle_arrive", "cargo_waiting"]

@pytest.mark.asyncio
async def test_subscribe_events_rejected_request_raises():
    client = make_admin()

    task = asyncio.ensure_future(client.subscribe_events(events=["not_an_event"]))
    await asyncio.sleep(0)
    await client.receive_ServerGamescript(
        None, data={"command": "subscribe_events", "request_id": 1,
                    "error": "unknown_event", "event": "not_an_event"})
    with pytest.raises(ValueError, match="unknown_event"):
        await task
    assert client._gs_futures == {}

@pytest.mark.asyncio
async def test_subscribe_events_timeout():
    client = make_admin()
    with pytest.raises(asyncio.TimeoutError):
        await client.subscribe_events(timeout=0.05)
    assert client._gs_futures == {}

@pytest.mark.asyncio
async def test_unsubscribe_events():
    client = make_admin()

    task = asyncio.ensure_future(client.unsubscribe_events())
    await asyncio.sleep(0)
    assert decode_gamescript_payload(client._protocol.sent[0]) == {
        "command": "unsubscribe_events", "request_id": 1,
    }

    reply = {"command": "unsubscribe_events", "request_id": 1, "events": []}
    await client.receive_ServerGamescript(None, data=reply)
    assert await task == reply


# --- Receiving events ---

@pytest.mark.asyncio
async def test_event_batch_reaches_on_event_and_bypasses_on_gamescript():
    client = make_admin()
    seen, other = [], []
    client.on_event = seen.append
    client.on_gamescript = other.append

    depart = {"event": "vehicle_depart", "tick": 120, "vehicle_id": 7,
              "station_id": 3, "dwell": 74}
    await client.receive_ServerGamescript(None, data=events_packet(arrival(), depart))

    assert [event["event"] for event in seen] == ["vehicle_arrive", "vehicle_depart"]
    assert other == []

@pytest.mark.asyncio
async def test_non_event_gamescript_payloads_still_reach_on_gamescript():
    client = make_admin()
    seen, other = [], []
    client.on_event = seen.append
    client.on_gamescript = other.append

    # A reply that merely mentions events, and a malformed batch, are not event batches.
    await client.receive_ServerGamescript(None, data={"vehicles": []})
    await client.receive_ServerGamescript(None, data={"command": "events", "events": "nope"})

    assert seen == []
    assert other == [{"vehicles": []}, {"command": "events", "events": "nope"}]

@pytest.mark.asyncio
async def test_events_are_buffered_until_awaited():
    client = make_admin()
    await client.receive_ServerGamescript(None, data=events_packet(arrival(station_id=3),
                                                                   arrival(station_id=4)))

    # The oldest matching event is handed out first, and only once.
    first = await client.wait_for_event("vehicle_arrive")
    second = await client.wait_for_event(GameEventType.VehicleArrive)
    assert (first["station_id"], second["station_id"]) == (3, 4)
    assert len(client._event_buffer) == 0

@pytest.mark.asyncio
async def test_wait_for_event_resolves_on_arrival():
    client = make_admin()
    task = asyncio.ensure_future(client.wait_for_event("cargo_waiting", timeout=5.0))
    await asyncio.sleep(0)

    cargo = {"event": "cargo_waiting", "tick": 300, "station_id": 3,
             "cargo_id": 0, "waiting": 25, "delta": 15}
    await client.receive_ServerGamescript(None, data=events_packet(arrival(), cargo))

    assert await task == cargo
    # The event that did not match the waiter is still buffered for the next caller.
    assert await client.wait_for_event() == arrival()
    assert client._event_waiters == []

@pytest.mark.asyncio
async def test_wait_for_event_accepts_several_kinds():
    client = make_admin()
    task = asyncio.ensure_future(
        client.wait_for_event({GameEventType.VehicleArrive, GameEventType.VehicleDepart}))
    await asyncio.sleep(0)

    await client.receive_ServerGamescript(
        None, data=events_packet({"event": "town_founded", "tick": 5, "town_id": 2}, arrival()))
    assert (await task)["event"] == "vehicle_arrive"

@pytest.mark.asyncio
async def test_wait_for_event_ignores_non_dict_events_when_filtering():
    client = make_admin()
    task = asyncio.ensure_future(client.wait_for_event("vehicle_arrive", timeout=0.05))
    await asyncio.sleep(0)

    # A malformed entry inside a batch must not satisfy a filtered waiter.
    await client.receive_ServerGamescript(None, data=events_packet("garbage"))
    with pytest.raises(asyncio.TimeoutError):
        await task
    assert list(client._event_buffer) == ["garbage"]

@pytest.mark.asyncio
async def test_wait_for_event_timeout_deregisters_the_waiter():
    client = make_admin()
    with pytest.raises(asyncio.TimeoutError):
        await client.wait_for_event("vehicle_arrive", timeout=0.05)
    assert client._event_waiters == []

@pytest.mark.asyncio
async def test_already_resolved_waiters_are_skipped():
    client = make_admin()
    # A waiter whose future completed but that has not been cleaned up yet must not swallow
    # the event; it goes to the next live waiter instead.
    stale = asyncio.get_running_loop().create_future()
    stale.set_result("already done")
    live = asyncio.get_running_loop().create_future()
    client._event_waiters = [(None, stale), (None, live)]

    await client.receive_ServerGamescript(None, data=events_packet(arrival()))
    assert live.result() == arrival()
    assert len(client._event_buffer) == 0

@pytest.mark.asyncio
async def test_event_buffer_is_bounded():
    client = make_admin(event_buffer_size=2)
    await client.receive_ServerGamescript(None, data=events_packet(
        arrival(station_id=1), arrival(station_id=2), arrival(station_id=3)))

    # The oldest event is dropped rather than growing the buffer without bound.
    assert [event["station_id"] for event in client._event_buffer] == [2, 3]

@pytest.mark.asyncio
async def test_disconnect_fails_pending_event_waiters():
    client = make_admin()
    task = asyncio.ensure_future(client.wait_for_event())
    await asyncio.sleep(0)

    client.disconnect(None)
    with pytest.raises(ConnectionError):
        await task
    assert client._event_waiters == []