Merge pull request 'Add station listing and cargo queries to admin client' (#21) from claude/station-realtime-planned-data-0fc51a into main

Reviewed-on: #21

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
 

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": <tile>, "owner": <company_id>, "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).
 

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."""

main_admin.py

@@ -50,6 +50,41 @@ 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()

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")

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 ---