From 81a4d9333dd6b7152ea6962a834b0fa5454a4540 Mon Sep 17 00:00:00 2001 From: kovagoadi Date: Thu, 23 Jul 2026 21:48:20 +0200 Subject: [PATCH] Add station listing and cargo queries to admin client Extends the AdminBridge GameScript JSON channel (the same relay used by list_vehicles/get_timetable) with station support: - list_stations(): enumerate stations, fire-and-forget like list_vehicles(). - get_station(): authoritative per-cargo snapshot of a station's live state, with both the real-time waiting amount (GSStation.GetCargoWaiting) and the planned cargodist link-graph flow (GetCargoPlanned), plus rating. - get_station_cargo(): break one cargo type down by source station and by next hop (the cargodist routing destination) for both waiting and planned amounts, with optional from_station/via_station filters. All three use stock GameScript API (no server patch, unlike timetables). Refactors the shared GS request/reply correlation out of get_timetable and get_station into a _gs_query() helper. Companion handlers must be added to the server-side AdminBridge GameScript (not tracked in this repo). Includes unit + e2e tests, a worked demo in main_admin.py, and protocol docs. Co-Authored-By: Claude Opus 4.8 --- README.md | 3 + docs/PROTOCOL.md | 20 +++++++ lib/openttd/client.py | 123 ++++++++++++++++++++++++++++++++++------ main_admin.py | 37 +++++++++++- tests/test_admin.py | 125 +++++++++++++++++++++++++++++++++++++++++ tests/test_e2e.py | 127 ++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 418 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index fcc2a9e..22ea7c9 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,9 @@ A high-performance, Object-Oriented Python client for OpenTTD servers, specifica - **Vehicle Listing:** Query vehicle data via the Admin GameScript channel with `list_vehicles()`. - **Vehicle Timetables:** Read and modify a vehicle's timetable (`change_timetable()`, `autofill_timetable()`, `set_timetable_start()`, `set_vehicle_on_time()`, `get_vehicle_timetable()`) via real game-protocol commands. - **Authoritative Timetable Reads:** `OpenTTDAdminClient.get_timetable()` fetches the real, current timetable of any vehicle from the running game (via a patched GameScript API + the AdminBridge GS) — no company join needed, works for timetables set before connecting. +- **Station Listing:** Enumerate stations via the Admin GameScript channel with `list_stations()`. +- **Station Cargo Snapshots:** `OpenTTDAdminClient.get_station()` returns a station's live per-cargo state from the running game — both the **real-time** amount waiting and the **planned** flow through the cargodist link graph — over the AdminBridge GS (stock GameScript API, no server patch needed). +- **Cargo Flow Breakdown:** `OpenTTDAdminClient.get_station_cargo()` breaks one cargo type down by **source station** and **next hop** (routing destination) for both waiting (real-time) and planned amounts, with optional `from_station`/`via_station` filters. ## 🛠 Setup diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 1a4af33..5170dd5 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -42,6 +42,26 @@ The stock GameScript API has no timetable getters, so this project patches the s Replies carrying a `request_id` that matches a pending request resolve that request and are **not** delivered to the `on_gamescript` callback; all other `ServerGamescript` traffic reaches the callback unchanged. The same `update_frequency` subscription requirement applies (`get_timetable()` subscribes automatically on first use). Since GameScripts do not tick while the game is paused, a query against a paused server times out (`asyncio.TimeoutError`). +### Station Listing +Like vehicles, the Admin Network has no native packet for enumerating individual stations (`ServerCompanyStats` only reports an aggregate per-company station count). `list_stations()` sends a `list_stations` command over the same GameScript JSON channel and the companion AdminBridge GameScript replies with station data through `ServerGamescript` (`{"command": "list_stations", "stations": [{"id", "name", ...}, ...]}`). It is fire-and-forget, so the reply is delivered to the `on_gamescript` callback — subscribe to `Gamescript` updates first, exactly as for `list_vehicles()`. An optional `company_id` field scopes the list to one company. + +### Station Query +`get_station()` fetches an authoritative snapshot of one station's live cargo state over the same GameScript JSON channel, awaiting the correlated reply — the station analogue of `get_timetable()`. Unlike timetables, the getters it relies on (`GSStation.GetCargoWaiting`, `GetCargoPlanned`, `GetCargoRating`) are part of the **stock** GameScript API, so this needs no server patch. The per-cargo reply exposes both the **real-time** amount currently waiting and the **planned** amount routed through the station by the cargodist link graph. + +- **Request:** `{"command": "get_station", "station_id": N, "request_id": X}` — `request_id` is the same client-side monotonic counter used by `get_timetable()`, matching the reply to the awaiting caller. +- **Reply (success):** `{"command": "get_station", "station_id": N, "request_id": X, "name": ..., "location": , "owner": , "cargo": [{"cargo_id", "waiting", "planned", "rating"}, ...]}` — `waiting` is the real-time units at the station (`GetCargoWaiting`), `planned` is the link-graph planned flow (`GetCargoPlanned`, 0 when cargo distribution is off for that cargo), and `rating` is the acceptance rating as a percentage (0-100, `GetCargoRating`) or `null` when the station has no rating for that cargo yet. Only cargo the station has handled appears. +- **Reply (error):** same envelope with an `"error"` field instead of the data: `"invalid_station"` (no such station) or `"response_too_large"`. `get_station()` raises `ValueError` for these. + +Correlation, the `update_frequency` subscription requirement (auto-subscribed on first use), and the paused-game timeout behave exactly as described for the Timetable Query above. + +### Station Cargo Flow Breakdown +`get_station_cargo()` drills into a single cargo type at one station and returns how its **waiting** (real-time) and **planned** amounts split across the cargo distribution (cargodist) link graph. Cargodist tags every unit with a **source** station (`from`, where it was first loaded) and a **next hop** (`via`, the next station it travels to toward its final destination). There is no per-station store of the *final* destination — the routing destination is the next hop — so the breakdown is offered along those two axes. The GS reads them with the stock `GSStation.GetCargoWaiting{From,Via,FromVia}` / `GetCargoPlanned{From,Via,FromVia}` scalars and the `GSStationList_Cargo{Waiting,Planned}By{From,Via}` (and `…ViaByFrom` / `…FromByVia`) list classes — again no server patch. + +- **Request:** `{"command": "get_station_cargo", "station_id": N, "cargo_id": C, "request_id": X}`, optionally with `"from_station"` and/or `"via_station"` filters. +- **Reply (success):** `{"command": "get_station_cargo", "station_id": N, "cargo_id": C, "request_id": X, "waiting": ..., "planned": ..., "waiting_by_from": [{"station", "amount"}, ...], "planned_by_from": [...], "waiting_by_via": [...], "planned_by_via": [...]}`. `waiting`/`planned` are the (filtered) totals; each `*_by_from` list groups by source station and each `*_by_via` list groups by next hop (zero-amount entries omitted). A `station` of `65535` (`STATION_INVALID`) means the source was deleted or — as a next hop — the cargo has no onward routing / is consumed here (also the only next hop for cargo using manual, non-cargodist distribution). Any supplied `from_station`/`via_station` filter is echoed back. +- **Filters:** `via_station` restricts the query (and the `*_by_from` breakdowns) to cargo whose next hop is that station; `from_station` restricts it (and the `*_by_via` breakdowns) to cargo from that source; supplying both makes `waiting`/`planned` the exact source-and-next-hop amount. Pass `65535` to target `STATION_INVALID`. +- **Reply (error):** same envelope with an `"error"` field: `"invalid_station"`, `"invalid_cargo"`, `"invalid_from_station"`/`"invalid_via_station"` (a filter that is neither a valid station nor `STATION_INVALID`), or `"response_too_large"`. `get_station_cargo()` raises `ValueError` for these. + ## Vehicle Timetables (Game Port DoCommands) Unlike vehicle listing, timetables have no stock GameScript API surface (this project adds read-only getters via a server patch — see "Timetable Query" above; writing still has none). Reading and modifying them requires real engine commands (`DoCommand`s) sent over the **game port** (TCP 3979) via `ClientCommand`/`ServerCommand` packets, not the Admin Network. This section covers the wire format; for how to call the methods and what each parameter means, see the [Vehicle Timetables Usage Guide](TIMETABLES.md). diff --git a/lib/openttd/client.py b/lib/openttd/client.py index bba8219..5e7271f 100644 --- a/lib/openttd/client.py +++ b/lib/openttd/client.py @@ -454,6 +454,49 @@ class OpenTTDAdminClient: payload["company_id"] = company_id await self.send_gamescript(payload) + async def list_stations(self, company_id=None): + """Request a list of stations via GameScript. company_id=None for all companies. + + Like list_vehicles(), this is fire-and-forget: the AdminBridge GameScript replies with a + {"stations": [...]} envelope delivered to the on_gamescript callback, so subscribe to + Gamescript updates first (update_frequency(Gamescript, Automatic)) or the reply is dropped. + For a station's live cargo detail (waiting vs planned), use get_station(). + """ + payload = {"command": "list_stations"} + if company_id is not None: + payload["company_id"] = company_id + await self.send_gamescript(payload) + + async def _gs_query(self, payload, timeout, context): + """Send a GameScript request and await its correlated reply. + + Assigns a fresh request_id, registers a future the ServerGamescript handler resolves when + the matching reply arrives, and (on first use) subscribes to Gamescript updates so the + server actually forwards the reply. `payload` is the request dict without request_id; + `context` is a label used in the ValueError raised on a GameScript-reported error. + + Raises asyncio.TimeoutError if no reply arrives within `timeout`, ValueError on an error + reply, and ConnectionError if the admin connection drops while waiting. + """ + from .protocol import AdminUpdateType, AdminUpdateFrequency + if not self._gs_subscribed: + await self.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic) + self._gs_subscribed = True + self._gs_request_id += 1 + rid = self._gs_request_id + fut = asyncio.get_running_loop().create_future() + self._gs_futures[rid] = fut + request = dict(payload) + request["request_id"] = rid + try: + await self.send_gamescript(request) + data = await asyncio.wait_for(fut, timeout) + finally: + self._gs_futures.pop(rid, None) + if "error" in data: + raise ValueError(f"{context}: {data['error']}") + return data + async def get_timetable(self, vehicle_id, timeout=5.0): """Fetch an authoritative timetable snapshot for a vehicle via the AdminBridge GameScript. @@ -470,22 +513,70 @@ class OpenTTDAdminClient: ValueError on a GameScript-reported error (invalid_vehicle, response_too_large), and ConnectionError if the admin connection drops while waiting. """ - from .protocol import AdminUpdateType, AdminUpdateFrequency - if not self._gs_subscribed: - await self.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic) - self._gs_subscribed = True - self._gs_request_id += 1 - rid = self._gs_request_id - fut = asyncio.get_running_loop().create_future() - self._gs_futures[rid] = fut - try: - await self.send_gamescript({"command": "get_timetable", "vehicle_id": vehicle_id, "request_id": rid}) - data = await asyncio.wait_for(fut, timeout) - finally: - self._gs_futures.pop(rid, None) - if "error" in data: - raise ValueError(f"get_timetable({vehicle_id}): {data['error']}") - return data + return await self._gs_query( + {"command": "get_timetable", "vehicle_id": vehicle_id}, timeout, + f"get_timetable({vehicle_id})") + + async def get_station(self, station_id, timeout=5.0): + """Fetch an authoritative snapshot of a station's live cargo state via the AdminBridge GameScript. + + This queries the real game state (like get_timetable() does for vehicles): it works for any + existing station regardless of when it was built or when this client connected. Auto-subscribes + to Gamescript updates on first use; if you manage update frequencies yourself, ensure + update_frequency(Gamescript, Automatic) is active before calling. + + Returns a dict with station-level keys (name, location, owner) and a "cargo" list of per-cargo + dicts. Each cargo dict carries both the real-time and the planned amounts: + - "waiting": units currently sitting at the station (real-time, GSStation.GetCargoWaiting) + - "planned": units planned to move through it per the cargodist link graph + (GSStation.GetCargoPlanned); 0 when cargo distribution is not enabled for that cargo + - "rating": the station's acceptance rating for the cargo as a percentage (0-100), + or None if the station has no rating for that cargo yet + Only cargo types the station has ever handled appear in the list. + + Raises asyncio.TimeoutError if no reply arrives (e.g. game paused, GS not loaded), + ValueError on a GameScript-reported error (invalid_station, response_too_large), and + ConnectionError if the admin connection drops while waiting. + """ + return await self._gs_query( + {"command": "get_station", "station_id": station_id}, timeout, + f"get_station({station_id})") + + async def get_station_cargo(self, station_id, cargo_id, from_station=None, via_station=None, timeout=5.0): + """Fetch a per-source / per-next-hop breakdown of one cargo at a station via the AdminBridge GS. + + Where get_station() reports each cargo's totals, this drills into a single cargo type and + shows how the waiting (real-time) and planned amounts split across the cargo distribution + (cargodist) link graph. Cargodist tracks every unit by its source station (where it was + first loaded) and its next hop (the next station it heads to on the way to its final + destination); there is no separate "final destination" store, so the routing destination is + the next hop ("via"). + + Returns a dict with the (optionally filtered) totals "waiting" and "planned", plus four + breakdown lists, each a list of {"station": id, "amount": n} entries (zero amounts omitted): + - "waiting_by_from" / "planned_by_from": grouped by source station + - "waiting_by_via" / "planned_by_via": grouped by next hop (routing destination) + A station id of 65535 (STATION_INVALID) marks cargo whose source was deleted or, as a next + hop, cargo with no onward routing / to be consumed at this station (also the sole next hop + for cargo types using manual, non-cargodist distribution). + + Optional filters narrow the query: + - from_station: only cargo originating at this source station. + - via_station: only cargo whose next hop is this station. + Passing from_station restricts the by_via breakdown to that source (and the totals to it); + passing via_station restricts the by_from breakdown to that next hop; passing both makes the + totals the exact source+next-hop amount. Pass 65535 for either to target STATION_INVALID. + + Raises asyncio.TimeoutError if no reply arrives (e.g. game paused, GS not loaded), + ValueError on a GameScript-reported error (invalid_station, invalid_cargo, + response_too_large), and ConnectionError if the admin connection drops while waiting. + """ + payload = {"command": "get_station_cargo", "station_id": station_id, "cargo_id": cargo_id} + if from_station is not None: + payload["from_station"] = from_station + if via_station is not None: + payload["via_station"] = via_station + return await self._gs_query(payload, timeout, f"get_station_cargo({station_id}, {cargo_id})") async def send_gamescript(self, json_data): """Send a JSON string to the GameScript.""" diff --git a/main_admin.py b/main_admin.py index 738b3f4..1d4f05a 100644 --- a/main_admin.py +++ b/main_admin.py @@ -49,7 +49,42 @@ async def run_admin(): print("--- Requesting vehicle info via GameScript ---") await admin.list_vehicles() - + + # Capture station-list replies (delivered to on_gamescript, like list_vehicles) while + # still logging every other GameScript message. + stations = [] + def gamescript_capture(data): + if isinstance(data, dict) and "stations" in data: + stations.append(data["stations"]) + gamescript_logger(data) + admin.on_gamescript = gamescript_capture + + print("--- Requesting station info via GameScript ---") + await admin.list_stations() + await asyncio.sleep(1) + + # Fetch one station's authoritative live cargo (real-time waiting + planned). + if stations and stations[-1]: + sid = stations[-1][0]["id"] + try: + data = await admin.get_station(sid, timeout=10.0) + print(f"--- Station {sid} ({data.get('name')}) cargo: real-time waiting vs planned ---") + for cargo in data.get("cargo", []): + print(f" cargo {cargo['cargo_id']}: waiting={cargo['waiting']} " + f"planned={cargo['planned']} rating={cargo['rating']}") + + # Break the first cargo down by source station and by next hop (routing destination). + if data.get("cargo"): + cid = data["cargo"][0]["cargo_id"] + flow = await admin.get_station_cargo(sid, cid, timeout=10.0) + print(f"--- Station {sid} cargo {cid} flow breakdown (station 65535 = none/deleted) ---") + print(f" waiting by source: {flow['waiting_by_from']}") + print(f" waiting by next hop: {flow['waiting_by_via']}") + print(f" planned by source: {flow['planned_by_from']}") + print(f" planned by next hop: {flow['planned_by_via']}") + except Exception as e: + print(f"!!! station query failed: {e}") + await asyncio.sleep(5) print("--- Quitting ---") await admin.quit() diff --git a/tests/test_admin.py b/tests/test_admin.py index d01913e..64af09d 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -46,6 +46,20 @@ async def test_admin_list_vehicles(): assert decode_gamescript_payload(proto.sent[0]) == {"command": "list_vehicles"} assert decode_gamescript_payload(proto.sent[1]) == {"command": "list_vehicles", "company_id": 2} +@pytest.mark.asyncio +async def test_admin_list_stations(): + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") + proto = MockProtocol() + client._protocol = proto + client._transport = MockTransport() + + await client.list_stations() + await client.list_stations(company_id=2) + + assert len(proto.sent) == 2 + assert decode_gamescript_payload(proto.sent[0]) == {"command": "list_stations"} + assert decode_gamescript_payload(proto.sent[1]) == {"command": "list_stations", "company_id": 2} + @pytest.mark.asyncio async def test_admin_client_connect_and_actions(monkeypatch): client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") @@ -258,6 +272,117 @@ async def test_admin_get_timetable_disconnect_fails_pending(): await task assert client._gs_futures == {} +@pytest.mark.asyncio +async def test_admin_get_station_request_and_response(): + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") + proto = MockProtocol() + client._protocol = proto + client._transport = MockTransport() + + task = asyncio.ensure_future(client.get_station(3)) + await asyncio.sleep(0) # let the task send the request + + # First use auto-subscribes to Gamescript updates, then sends the query. + assert len(proto.sent) == 2 + assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency + assert decode_gamescript_payload(proto.sent[1]) == { + "command": "get_station", "station_id": 3, "request_id": 1, + } + + response = {"command": "get_station", "station_id": 3, "request_id": 1, + "name": "Test Central", "location": 12345, "owner": 0, + "cargo": [{"cargo_id": 0, "waiting": 42, "planned": 17, "rating": 71}]} + await client.receive_ServerGamescript(None, data=response) + assert await task == response + assert client._gs_futures == {} + +@pytest.mark.asyncio +async def test_admin_get_station_error_response(): + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") + client._protocol = MockProtocol() + client._transport = MockTransport() + + task = asyncio.ensure_future(client.get_station(65535)) + await asyncio.sleep(0) + await client.receive_ServerGamescript( + None, data={"command": "get_station", "station_id": 65535, + "request_id": 1, "error": "invalid_station"}) + with pytest.raises(ValueError, match="invalid_station"): + await task + assert client._gs_futures == {} + +@pytest.mark.asyncio +async def test_admin_get_station_timeout(): + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") + client._protocol = MockProtocol() + client._transport = MockTransport() + + with pytest.raises(asyncio.TimeoutError): + await client.get_station(3, timeout=0.05) + assert client._gs_futures == {} + +@pytest.mark.asyncio +async def test_admin_get_station_cargo_request_and_response(): + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") + proto = MockProtocol() + client._protocol = proto + client._transport = MockTransport() + + task = asyncio.ensure_future(client.get_station_cargo(3, 0)) + await asyncio.sleep(0) # let the task send the request + + # First use auto-subscribes to Gamescript updates, then sends the query. + assert len(proto.sent) == 2 + assert proto.sent[0][2] == PacketAdminType.AdminUpdateFrequency + assert decode_gamescript_payload(proto.sent[1]) == { + "command": "get_station_cargo", "station_id": 3, "cargo_id": 0, "request_id": 1, + } + + response = {"command": "get_station_cargo", "station_id": 3, "cargo_id": 0, "request_id": 1, + "waiting": 60, "planned": 40, + "waiting_by_from": [{"station": 5, "amount": 25}, {"station": 6, "amount": 35}], + "planned_by_from": [{"station": 5, "amount": 40}], + "waiting_by_via": [{"station": 7, "amount": 60}], + "planned_by_via": [{"station": 7, "amount": 40}]} + await client.receive_ServerGamescript(None, data=response) + assert await task == response + assert client._gs_futures == {} + +@pytest.mark.asyncio +async def test_admin_get_station_cargo_with_filters_encoding(): + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") + proto = MockProtocol() + client._protocol = proto + client._transport = MockTransport() + client._gs_subscribed = True # skip the auto-subscribe so only the query is sent + + task = asyncio.ensure_future(client.get_station_cargo(3, 0, from_station=6, via_station=7)) + await asyncio.sleep(0) + + assert len(proto.sent) == 1 + assert decode_gamescript_payload(proto.sent[0]) == { + "command": "get_station_cargo", "station_id": 3, "cargo_id": 0, + "from_station": 6, "via_station": 7, "request_id": 1, + } + await client.receive_ServerGamescript( + None, data={"request_id": 1, "waiting": 12, "planned": 8}) + assert (await task)["waiting"] == 12 + +@pytest.mark.asyncio +async def test_admin_get_station_cargo_error_response(): + client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") + client._protocol = MockProtocol() + client._transport = MockTransport() + + task = asyncio.ensure_future(client.get_station_cargo(3, 999)) + await asyncio.sleep(0) + await client.receive_ServerGamescript( + None, data={"command": "get_station_cargo", "station_id": 3, "cargo_id": 999, + "request_id": 1, "error": "invalid_cargo"}) + with pytest.raises(ValueError, match="invalid_cargo"): + await task + assert client._gs_futures == {} + @pytest.mark.asyncio async def test_admin_gamescript_passthrough_unmatched(): client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin") diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 3c8bfb9..85cffd0 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -464,6 +464,133 @@ async def test_e2e_admin_get_timetable_invalid_vehicle(connected_admin): with pytest.raises(ValueError, match="invalid_vehicle"): await connected_admin.get_timetable(65535, timeout=10.0) +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_list_stations_all_companies(connected_admin): + # Public function: list_stations() + # Input 1: all companies (no company_id) + responses = [] + connected_admin.on_gamescript = lambda data: responses.append(data) + await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic) + + await connected_admin.list_stations() + await asyncio.sleep(0.5) + + assert not connected_admin.shutdown_event.is_set() + assert len(responses) >= 1 + assert "stations" in responses[-1] + assert isinstance(responses[-1]["stations"], list) + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_list_stations_specific_company(connected_admin): + # Public function: list_stations() + # Input 2: specific company_id + responses = [] + connected_admin.on_gamescript = lambda data: responses.append(data) + await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic) + + await connected_admin.list_stations(company_id=0) + await asyncio.sleep(0.5) + + assert not connected_admin.shutdown_event.is_set() + assert len(responses) >= 1 + assert "stations" in responses[-1] + assert isinstance(responses[-1]["stations"], list) + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_get_station_valid_station(connected_admin): + # Public function: get_station() + # Input 1: a real station id discovered via list_stations + responses = [] + connected_admin.on_gamescript = lambda data: responses.append(data) + await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic) + await connected_admin.list_stations() + await asyncio.sleep(0.5) + + assert len(responses) >= 1 and "stations" in responses[-1] + stations = responses[-1]["stations"] + if not stations: + pytest.skip("No stations on the test server to query.") + sid = stations[0]["id"] + + data = await connected_admin.get_station(sid, timeout=10.0) + assert data["station_id"] == sid + assert "cargo" in data + assert isinstance(data["cargo"], list) + for cargo in data["cargo"]: + for key in ("cargo_id", "waiting", "planned", "rating"): + assert key in cargo + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_get_station_invalid_station(connected_admin): + # Public function: get_station() + # Input 2: an id no station can have -> GameScript reports invalid_station + with pytest.raises(ValueError, match="invalid_station"): + await connected_admin.get_station(65535, timeout=10.0) + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_get_station_cargo_breakdown(connected_admin): + # Public function: get_station_cargo() + # Input 1: a real station + a cargo it has handled, discovered via list_stations/get_station + responses = [] + connected_admin.on_gamescript = lambda data: responses.append(data) + await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic) + await connected_admin.list_stations() + await asyncio.sleep(0.5) + + assert responses and "stations" in responses[-1] + stations = responses[-1]["stations"] + if not stations: + pytest.skip("No stations on the test server to query.") + + # Find a station/cargo pair that actually has cargo data. + target = None + for st in stations: + detail = await connected_admin.get_station(st["id"], timeout=10.0) + if detail["cargo"]: + target = (st["id"], detail["cargo"][0]["cargo_id"]) + break + if target is None: + pytest.skip("No station with handled cargo to break down.") + sid, cid = target + + data = await connected_admin.get_station_cargo(sid, cid, timeout=10.0) + assert data["station_id"] == sid and data["cargo_id"] == cid + for key in ("waiting", "planned", + "waiting_by_from", "planned_by_from", "waiting_by_via", "planned_by_via"): + assert key in data + for key in ("waiting_by_from", "planned_by_from", "waiting_by_via", "planned_by_via"): + assert isinstance(data[key], list) + for entry in data[key]: + assert "station" in entry and "amount" in entry + + # Input 2: the same query narrowed by a next-hop (via) filter is accepted and echoes it back. + filtered = await connected_admin.get_station_cargo(sid, cid, via_station=sid, timeout=10.0) + assert filtered["via_station"] == sid + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_e2e_admin_get_station_cargo_invalid_cargo(connected_admin): + # Public function: get_station_cargo() + # A cargo id no cargo can have -> GameScript reports invalid_cargo. Needs a valid station. + responses = [] + connected_admin.on_gamescript = lambda data: responses.append(data) + await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic) + await connected_admin.list_stations() + await asyncio.sleep(0.5) + + assert responses and "stations" in responses[-1] + stations = responses[-1]["stations"] + if not stations: + pytest.skip("No stations on the test server to query.") + + with pytest.raises(ValueError, match="invalid_cargo"): + await connected_admin.get_station_cargo(stations[0]["id"], 250, timeout=10.0) + # --- Protocol Public Functions --- -- 2.49.1