kovagoadi
81a4d9333d
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 <noreply@anthropic.com>
651 lines
27 KiB
Python
651 lines
27 KiB
Python
import asyncio
|
|
import pytest
|
|
import pytest_asyncio
|
|
import sys
|
|
import os
|
|
import random
|
|
# Add lib to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'lib'))
|
|
|
|
from openttd import OpenTTDClient, OpenTTDAdminClient
|
|
from openttd.protocol import (
|
|
OpenTTDProtocol,
|
|
OpenTTDAdminProtocol,
|
|
AdminUpdateType,
|
|
AdminUpdateFrequency,
|
|
PacketGameType,
|
|
ModifyTimetableFlags
|
|
)
|
|
|
|
# These identify a vehicle/order that already exists in the local dev server's persisted
|
|
# save (company 0, unprotected, owns vehicle 7 with 2 orders) -- required for the timetable
|
|
# command tests below, since DoCommands are rejected unless issued by the owning company.
|
|
TIMETABLE_COMPANY_ID = 0
|
|
TIMETABLE_VEHICLE_ID = 7
|
|
TIMETABLE_ORDER_POSITION = 0
|
|
|
|
|
|
# --- Pytest Fixtures ---
|
|
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def connected_admin(server_config):
|
|
"""Fixture to yield a connected and authenticated OpenTTDAdminClient."""
|
|
admin_name = f"E2E_Admin_{random.randint(1000, 9999)}"
|
|
admin = OpenTTDAdminClient(
|
|
host=server_config["host"],
|
|
port=server_config["admin_port"],
|
|
admin_name=admin_name
|
|
)
|
|
await admin.connect(admin_password=server_config["password"], secure=True)
|
|
await asyncio.wait_for(admin.joined.wait(), timeout=10.0)
|
|
yield admin
|
|
if hasattr(admin, '_transport') and not admin.shutdown_event.is_set():
|
|
await admin.quit()
|
|
|
|
@pytest_asyncio.fixture
|
|
async def connected_client(server_config):
|
|
"""Fixture to yield a connected and joined spectator OpenTTDClient."""
|
|
client_name = f"E2E_Player_{random.randint(1000, 9999)}"
|
|
client = OpenTTDClient(
|
|
host=server_config["host"],
|
|
port=server_config["game_port"],
|
|
username=client_name
|
|
)
|
|
await client.connect(server_password=server_config["password"])
|
|
await client.join_company(company_id=255, company_password="")
|
|
await asyncio.wait_for(client.joined.wait(), timeout=15.0)
|
|
yield client
|
|
if hasattr(client, '_transport') and not client.shutdown_event.is_set():
|
|
await client.quit()
|
|
|
|
@pytest_asyncio.fixture
|
|
async def connected_owner_client(server_config):
|
|
"""Fixture to yield a client joined to TIMETABLE_COMPANY_ID (owns a real vehicle for command tests)."""
|
|
client_name = f"E2E_Owner_{random.randint(1000, 9999)}"
|
|
client = OpenTTDClient(
|
|
host=server_config["host"],
|
|
port=server_config["game_port"],
|
|
username=client_name
|
|
)
|
|
await client.connect(server_password=server_config["password"])
|
|
await client.join_company(company_id=TIMETABLE_COMPANY_ID, company_password="")
|
|
await asyncio.wait_for(client.joined.wait(), timeout=15.0)
|
|
yield client
|
|
if hasattr(client, '_transport') and not client.shutdown_event.is_set():
|
|
await client.quit()
|
|
|
|
|
|
# ==============================================================================
|
|
# --- End-to-End Tests (Covering all public functions with multiple inputs) ---
|
|
# ==============================================================================
|
|
|
|
# --- Game Client Public Functions ---
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_init_and_connect_multiple_inputs(server_config):
|
|
# Public function: __init__()
|
|
# Input 1: Custom port & default username
|
|
client1 = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
assert client1.host == server_config["host"]
|
|
assert client1.port == server_config["game_port"]
|
|
assert client1.username == "GeminiUser"
|
|
|
|
# Input 2: Custom port & custom username
|
|
client2 = OpenTTDClient(host=server_config["host"], port=server_config["game_port"], username="E2E_Player_Custom")
|
|
assert client2.username == "E2E_Player_Custom"
|
|
|
|
# Public function: connect()
|
|
# Input 1: Correct server password
|
|
await client1.connect(server_password=server_config["password"])
|
|
assert client1._transport is not None
|
|
await client1.quit()
|
|
|
|
# Input 2: Incorrect server password
|
|
await client2.connect(server_password="wrong_password")
|
|
await asyncio.sleep(0.5)
|
|
assert client2.shutdown_event.is_set()
|
|
await client2.quit()
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_join_company_multiple_inputs(server_config):
|
|
# Public function: join_company()
|
|
# Input 1: Join as spectator (company_id=255)
|
|
client1 = OpenTTDClient(host=server_config["host"], port=server_config["game_port"], username="E2E_Spectator")
|
|
await client1.connect(server_password=server_config["password"])
|
|
await client1.join_company(company_id=255, company_password="")
|
|
await asyncio.wait_for(client1.joined.wait(), timeout=10.0)
|
|
assert client1.joined.is_set()
|
|
await client1.quit()
|
|
|
|
# Input 2: Join specific company ID (company_id=1)
|
|
client2 = OpenTTDClient(host=server_config["host"], port=server_config["game_port"], username="E2E_Player_Join_1")
|
|
await client2.connect(server_password=server_config["password"])
|
|
await client2.join_company(company_id=1, company_password="comp_password")
|
|
await asyncio.sleep(0.5)
|
|
await client2.quit()
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_quit_and_disconnect_multiple_inputs(server_config):
|
|
# Public function: quit() and disconnect()
|
|
client = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
await client.connect(server_password=server_config["password"])
|
|
|
|
# Public function: disconnect()
|
|
# Input 1: disconnect callback with None
|
|
client.disconnect(None)
|
|
assert client.shutdown_event.is_set()
|
|
|
|
# Input 2: disconnect callback with custom string
|
|
client.disconnect("network_lost")
|
|
|
|
# Public function: quit()
|
|
client2 = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
await client2.connect(server_password=server_config["password"])
|
|
# Input 1: quit active connection
|
|
await client2.quit()
|
|
assert client2.shutdown_event.is_set()
|
|
|
|
# Input 2: quit already inactive client
|
|
await client2.quit()
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_change_timetable_wait_time(connected_owner_client):
|
|
# Public function: change_timetable()
|
|
# Input 1: set wait time
|
|
await connected_owner_client.change_timetable(TIMETABLE_VEHICLE_ID, TIMETABLE_ORDER_POSITION, ModifyTimetableFlags.WaitTime, 42)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
|
|
assert entry["orders"][TIMETABLE_ORDER_POSITION]["wait_time"] == 42
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_change_timetable_travel_time(connected_owner_client):
|
|
# Public function: change_timetable()
|
|
# Input 2: set travel time
|
|
await connected_owner_client.change_timetable(TIMETABLE_VEHICLE_ID, TIMETABLE_ORDER_POSITION, ModifyTimetableFlags.TravelTime, 99)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
|
|
assert entry["orders"][TIMETABLE_ORDER_POSITION]["travel_time"] == 99
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_autofill_timetable_enable(connected_owner_client):
|
|
# Public function: autofill_timetable()
|
|
# Input 1: enable autofill
|
|
await connected_owner_client.autofill_timetable(TIMETABLE_VEHICLE_ID, autofill=True, preserve_wait_time=False)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)["autofill"] is True
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_autofill_timetable_disable(connected_owner_client):
|
|
# Public function: autofill_timetable()
|
|
# Input 2: disable autofill, preserve wait time
|
|
await connected_owner_client.autofill_timetable(TIMETABLE_VEHICLE_ID, autofill=False, preserve_wait_time=True)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
|
|
assert entry["autofill"] is False
|
|
assert entry["autofill_preserve_wait_time"] is True
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_set_timetable_start_single_vehicle(connected_owner_client):
|
|
# Public function: set_timetable_start()
|
|
# Input 1: this vehicle only
|
|
await connected_owner_client.set_timetable_start(TIMETABLE_VEHICLE_ID, False, 500000)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
|
|
assert entry["timetable_start"] == 500000
|
|
assert entry["timetable_all"] is False
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_set_timetable_start_all_shared(connected_owner_client):
|
|
# Public function: set_timetable_start()
|
|
# Input 2: all vehicles sharing this order list
|
|
await connected_owner_client.set_timetable_start(TIMETABLE_VEHICLE_ID, True, 600000)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
entry = connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)
|
|
assert entry["timetable_start"] == 600000
|
|
assert entry["timetable_all"] is True
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_set_vehicle_on_time_single_vehicle(connected_owner_client):
|
|
# Public function: set_vehicle_on_time()
|
|
# Input 1: reset lateness for this vehicle only
|
|
await connected_owner_client.set_vehicle_on_time(TIMETABLE_VEHICLE_ID, apply_to_group=False)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)["on_time_apply_to_group"] is False
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_set_vehicle_on_time_apply_to_group(connected_owner_client):
|
|
# Public function: set_vehicle_on_time()
|
|
# Input 2: reset lateness for every vehicle sharing these orders
|
|
await connected_owner_client.set_vehicle_on_time(TIMETABLE_VEHICLE_ID, apply_to_group=True)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID)["on_time_apply_to_group"] is True
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_get_vehicle_timetable_after_change(connected_owner_client):
|
|
# Public function: get_vehicle_timetable()
|
|
# Input 1: a vehicle with observed state
|
|
await connected_owner_client.change_timetable(TIMETABLE_VEHICLE_ID, TIMETABLE_ORDER_POSITION, ModifyTimetableFlags.WaitTime, 15)
|
|
await asyncio.sleep(1.0)
|
|
assert connected_owner_client.get_vehicle_timetable(TIMETABLE_VEHICLE_ID) is not None
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_get_vehicle_timetable_unknown_vehicle(connected_owner_client):
|
|
# Public function: get_vehicle_timetable()
|
|
# Input 2: a vehicle id with no observed state
|
|
assert connected_owner_client.get_vehicle_timetable(999999) is None
|
|
|
|
|
|
# --- Admin Client Public Functions ---
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_init_and_connect_multiple_inputs(server_config):
|
|
# Public function: __init__()
|
|
# Input 1: Custom admin name
|
|
admin1 = OpenTTDAdminClient(host=server_config["host"], port=server_config["admin_port"], admin_name="E2E_Admin_1")
|
|
assert admin1.admin_name == "E2E_Admin_1"
|
|
|
|
# Input 2: Alternative admin name
|
|
admin2 = OpenTTDAdminClient(host=server_config["host"], port=server_config["admin_port"], admin_name="E2E_Admin_2")
|
|
assert admin2.admin_name == "E2E_Admin_2"
|
|
|
|
# Public function: connect()
|
|
# Input 1: secure=True (PAKE auth)
|
|
await admin1.connect(admin_password=server_config["password"], secure=True)
|
|
await asyncio.wait_for(admin1.joined.wait(), timeout=10.0)
|
|
assert admin1.joined.is_set()
|
|
await admin1.quit()
|
|
|
|
# Input 2: secure=False (plaintext auth, rejected by server)
|
|
await admin2.connect(admin_password=server_config["password"], secure=False)
|
|
await asyncio.sleep(0.5)
|
|
assert admin2.shutdown_event.is_set()
|
|
await admin2.quit()
|
|
|
|
# Input 3: incorrect password
|
|
admin3 = OpenTTDAdminClient(host=server_config["host"], port=server_config["admin_port"])
|
|
await admin3.connect(admin_password="wrong_password", secure=True)
|
|
await asyncio.sleep(0.5)
|
|
assert admin3.shutdown_event.is_set()
|
|
await admin3.quit()
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_quit_and_disconnect_multiple_inputs(server_config):
|
|
# Public function: quit() and disconnect()
|
|
admin = OpenTTDAdminClient(host=server_config["host"], port=server_config["admin_port"])
|
|
await admin.connect(admin_password=server_config["password"], secure=True)
|
|
|
|
# Public function: disconnect()
|
|
# Input 1: disconnect callback with None
|
|
admin.disconnect(None)
|
|
assert admin.shutdown_event.is_set()
|
|
|
|
# Input 2: disconnect callback with custom string
|
|
admin.disconnect("admin_shutdown")
|
|
|
|
# Public function: quit()
|
|
admin2 = OpenTTDAdminClient(host=server_config["host"], port=server_config["admin_port"])
|
|
await admin2.connect(admin_password=server_config["password"], secure=True)
|
|
# Input 1: quit active connection
|
|
await admin2.quit()
|
|
assert admin2.shutdown_event.is_set()
|
|
|
|
# Input 2: quit already inactive client
|
|
await admin2.quit()
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_send_rcon_multiple_inputs(connected_admin):
|
|
# Public function: send_rcon()
|
|
# Input 1: command "help"
|
|
await connected_admin.send_rcon("help")
|
|
|
|
# Input 2: command "setting max_clients"
|
|
await connected_admin.send_rcon("setting max_clients")
|
|
await asyncio.sleep(0.5)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_send_chat_multiple_inputs(connected_admin, connected_client):
|
|
# Public function: send_chat()
|
|
# Input 1: ChatBroadcast (action=1, dest_type=0, dest_id=0)
|
|
await connected_admin.send_chat("Hello from E2E Broadcast!", action=1, dest_type=0, dest_id=0)
|
|
|
|
# Input 2: Chat direct to client (action=1, dest_type=1, dest_id=client_id)
|
|
client_id = connected_client.client_id if connected_client.client_id is not None else 1
|
|
await connected_admin.send_chat("Hello private", action=1, dest_type=1, dest_id=client_id)
|
|
|
|
# Input 3: ChatBroadcast action (action=3, dest_type=0)
|
|
await connected_admin.send_chat("wave", action=3, dest_type=0, dest_id=0)
|
|
await asyncio.sleep(0.5)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_update_frequency_multiple_inputs(connected_admin):
|
|
# Public function: update_frequency()
|
|
# Input 1: Chat update to Automatic
|
|
await connected_admin.update_frequency(AdminUpdateType.Chat, AdminUpdateFrequency.Automatic)
|
|
|
|
# Input 2: Console update to Poll
|
|
await connected_admin.update_frequency(AdminUpdateType.Console, AdminUpdateFrequency.Poll)
|
|
await asyncio.sleep(0.5)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_poll_and_helpers_multiple_inputs(connected_admin):
|
|
# Public functions: poll(), poll_clients(), poll_companies(), poll_economy(), poll_stats()
|
|
# Input 1 for poll(): ClientInfo poll with 0xFFFFFFFF
|
|
await connected_admin.poll(AdminUpdateType.ClientInfo, 0xFFFFFFFF)
|
|
# Input 2 for poll(): CompanyInfo poll with 0
|
|
await connected_admin.poll(AdminUpdateType.CompanyInfo, 0)
|
|
|
|
# Input 1 for poll_clients(): 0xFFFFFFFF
|
|
await connected_admin.poll_clients(0xFFFFFFFF)
|
|
# Input 2 for poll_clients(): specific client ID 1
|
|
await connected_admin.poll_clients(1)
|
|
|
|
# Input 1 for poll_companies(): 0xFFFFFFFF
|
|
await connected_admin.poll_companies(0xFFFFFFFF)
|
|
# Input 2 for poll_companies(): specific company ID 0
|
|
await connected_admin.poll_companies(0)
|
|
|
|
# Input 1 for poll_economy(): 0xFFFFFFFF
|
|
await connected_admin.poll_economy(0xFFFFFFFF)
|
|
# Input 2 for poll_economy(): specific company ID 0
|
|
await connected_admin.poll_economy(0)
|
|
|
|
# Input 1 for poll_stats(): 0xFFFFFFFF
|
|
await connected_admin.poll_stats(0xFFFFFFFF)
|
|
# Input 2 for poll_stats(): specific company ID 0
|
|
await connected_admin.poll_stats(0)
|
|
await asyncio.sleep(0.5)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_send_gamescript_multiple_inputs(connected_admin):
|
|
# Public function: send_gamescript()
|
|
# Input 1: healthcheck dict
|
|
await connected_admin.send_gamescript({"command": "healthcheck"})
|
|
|
|
# Input 2: alternative command dict
|
|
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)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_get_timetable_valid_vehicle(connected_admin):
|
|
# Public function: get_timetable()
|
|
# Input 1: a real vehicle id discovered via list_vehicles
|
|
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 len(responses) >= 1 and "vehicles" in responses[-1]
|
|
vehicles = responses[-1]["vehicles"]
|
|
if not vehicles:
|
|
pytest.skip("No vehicles on the test server to query a timetable for.")
|
|
vid = vehicles[0]["id"]
|
|
|
|
data = await connected_admin.get_timetable(vid, timeout=10.0)
|
|
assert data["vehicle_id"] == vid
|
|
for key in ("lateness", "start_tick", "current_order_time", "total_duration", "orders"):
|
|
assert key in data
|
|
assert isinstance(data["orders"], list)
|
|
for order in data["orders"]:
|
|
for key in ("position", "wait_time", "travel_time", "wait_timetabled",
|
|
"travel_timetabled", "wait_fixed", "travel_fixed", "leave_type", "max_speed"):
|
|
assert key in order
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_get_timetable_invalid_vehicle(connected_admin):
|
|
# Public function: get_timetable()
|
|
# Input 2: an id no vehicle can have -> GameScript reports invalid_vehicle
|
|
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 ---
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_protocol_public_functions_multiple_inputs(server_config):
|
|
# Public functions: __init__(), receive_packet(), send_packet()
|
|
client_alt = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
admin_alt = OpenTTDAdminClient(host=server_config["host"], port=server_config["admin_port"])
|
|
|
|
# 1. Test __init__() with multiple inputs (handlers)
|
|
proto_game = OpenTTDProtocol(client_alt)
|
|
proto_admin = OpenTTDAdminProtocol(admin_alt)
|
|
assert proto_game.handler == client_alt
|
|
assert proto_admin.handler == admin_alt
|
|
|
|
client_alt2 = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
admin_alt2 = OpenTTDAdminClient(host=server_config["host"], port=server_config["admin_port"])
|
|
proto_game2 = OpenTTDProtocol(client_alt2)
|
|
proto_admin2 = OpenTTDAdminProtocol(admin_alt2)
|
|
assert proto_game2.handler == client_alt2
|
|
assert proto_admin2.handler == admin_alt2
|
|
|
|
# 2. Test receive_packet() with multiple inputs
|
|
# Input 1: Valid packet structure (length >= 3)
|
|
res_type1, res_data1 = proto_game.receive_packet(None, memoryview(b"\x03\x00\x05")) # type 5 is ServerUnused
|
|
assert res_type1 == PacketGameType.ServerUnused
|
|
|
|
# Input 2: Invalid/short packet structure (length < 3)
|
|
res_type2, res_data2 = proto_game.receive_packet(None, memoryview(b"\x01"))
|
|
assert res_type2 == PacketGameType.ServerUnused # Falls back to ServerUnused on error
|
|
|
|
# 3. Test send_packet() with multiple inputs (using mock transports)
|
|
class FakeTransport:
|
|
def __init__(self):
|
|
self.written = []
|
|
self.closed = False
|
|
def write(self, data):
|
|
self.written.append(data)
|
|
def is_closing(self):
|
|
return self.closed
|
|
|
|
transport = FakeTransport()
|
|
proto_game.transport = transport
|
|
proto_game._can_write.set()
|
|
|
|
# Input 1: send_packet with encryption disabled
|
|
client_alt.encryption_enabled = False
|
|
await proto_game.send_packet(b"\x03\x00\x04") # ClientUnused packet
|
|
assert len(transport.written) == 1
|
|
|
|
# Input 2: send_packet with encryption enabled
|
|
client_alt.encryption_enabled = True
|
|
client_alt._session_key_send = b"\x00" * 32
|
|
client_alt._encryption_nonce = b"\x00" * 24
|
|
await proto_game.send_packet(b"\x03\x00\x05")
|
|
assert len(transport.written) == 2
|