Add vehicle listing support to admin client
All checks were successful
Continuous Integration / lint-and-security (pull_request) Successful in 31s
Continuous Integration / tests-and-coverage (pull_request) Successful in 25s

The Admin Network has no native packet for listing individual vehicles,
so list_vehicles() sends a "list_vehicles" command over the existing
GameScript JSON channel and relies on a companion server-side script to
reply with vehicle data via ServerGamescript. Requires subscribing to
Gamescript updates (documented in docs/PROTOCOL.md) to receive the reply.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-16 20:44:44 +02:00
parent 309f1da762
commit 0a8271d57c
6 changed files with 67 additions and 1 deletions

View File

@@ -1,10 +1,15 @@
import pytest
import asyncio
import json
import os
import monocypher
from openttd import OpenTTDAdminClient
from openttd.protocol import PacketAdminType, AdminUpdateType, AdminUpdateFrequency
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])
class MockTransport:
def __init__(self):
self._closing = False
@@ -27,6 +32,20 @@ def test_admin_packet_types():
assert PacketAdminType.ServerWelcome == 104
assert PacketAdminType.ServerAuthRequest == 128
@pytest.mark.asyncio
async def test_admin_list_vehicles():
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")
proto = MockProtocol()
client._protocol = proto
client._transport = MockTransport()
await client.list_vehicles()
await client.list_vehicles(company_id=2)
assert len(proto.sent) == 2
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_client_connect_and_actions(monkeypatch):
client = OpenTTDAdminClient("127.0.0.1", port=3977, admin_name="TestAdmin")

View File

@@ -267,6 +267,40 @@ async def test_e2e_admin_send_gamescript_multiple_inputs(connected_admin):
await connected_admin.send_gamescript({"command": "ping", "sequence": 1})
await asyncio.sleep(0.5)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_list_vehicles_all_companies(connected_admin):
# Public function: list_vehicles()
# 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_vehicles()
await asyncio.sleep(0.5)
assert not connected_admin.shutdown_event.is_set()
assert len(responses) >= 1
assert "vehicles" in responses[-1]
assert isinstance(responses[-1]["vehicles"], list)
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_e2e_admin_list_vehicles_specific_company(connected_admin):
# Public function: list_vehicles()
# 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_vehicles(company_id=0)
await asyncio.sleep(0.5)
assert not connected_admin.shutdown_event.is_set()
assert len(responses) >= 1
assert "vehicles" in responses[-1]
assert isinstance(responses[-1]["vehicles"], list)
# --- Protocol Public Functions ---