kovagoadiandClaude
f7ca395a4f
The last commit noted in passing that the server-side half of the admin
GameScript channel "is not in this repo -- docker/config is gitignored --
so it has to be updated separately for any of this to work." That was true
of all nine features the README documents: list_vehicles, list_stations,
list_cargo, get_timetable, get_station, get_station_cargo, get_dispatch and
the event stream all answer from 705 lines of Squirrel that no clone could
reproduce, no reviewer could see, and CI never touched.
The bridge now lives in gamescript/AdminBridge/ with its own README, the
same arrangement docker/patches/ uses for the local JGRPP patches, and
docker-compose.yml bind-mounts it read-only over the container's
game/AdminBridge. docker/config stays ignored -- it also holds savegames,
downloaded content and generated config -- so the copy under it is now
shadowed and can be deleted. main.nut is byte-identical to what was running,
apart from the version work below.
Adds a version handshake, because the channel gives no way to tell a stale
bridge from a hung one: a bridge that does not recognise a command drops it
silently, so a client ahead of the server sees nothing but timeouts. The
bridge now answers get_version with its protocol version plus its command
and event catalogues, and get_bridge_version() raises when that is below
GS_BRIDGE_VERSION. It is opt-in rather than checked on connect: GameScripts
do not tick while the game is paused, so an automatic check would refuse to
connect to a paused server. A bridge older than 4 predates get_version
itself and can only fail by timing out, so the E2E test catches that and
reports it by name instead.
tests/test_gamescript.py gives CI a foothold on the GameScript without a
Squirrel toolchain: it parses the .nut files and pins the protocol version
across info.nut, main.nut and protocol.py, the event catalogue against
GameEventType, and the command table against the commands client.py sends.
The version has to be declared three times because a GameScript cannot read
its own info.nut at runtime -- GSController.GetVersion() returns the OpenTTD
version, not the script's.
info.nut also gains MinVersionToLoad() { return 1; }. The engine defaults it
to GetVersion(), so without it this bump would orphan every savegame pinned
to version 3: the scanner finds no compatible script and falls back with a
warning. The bridge keeps no savegame state, so any version can take over.
HandleCommand now dispatches through the same table get_version reports,
rather than an if/else chain, so the catalogue a client feature-detects
against cannot drift from what is implemented.
Co-Authored-By: Claude <[email protected]>
901 lines
38 KiB
Python
901 lines
38 KiB
Python
import asyncio
|
|
import os
|
|
import random
|
|
import sys
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
# Add lib to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'lib'))
|
|
|
|
from openttd import OpenTTDAdminClient, OpenTTDClient
|
|
from openttd.protocol import (
|
|
GS_BRIDGE_VERSION,
|
|
AdminUpdateFrequency,
|
|
AdminUpdateType,
|
|
GameEventType,
|
|
ModifyTimetableFlags,
|
|
OpenTTDAdminProtocol,
|
|
OpenTTDProtocol,
|
|
PacketGameType,
|
|
)
|
|
|
|
# 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
|
|
# A station TIMETABLE_VEHICLE_ID can legally serve, used for add_order/remove_order tests.
|
|
ORDER_STATION_ID = 6
|
|
# A second vehicle of the same company, dedicated to the scheduled-dispatch test, which owns and
|
|
# overwrites this vehicle's dispatch state. It must have its own order list -- dispatch schedules
|
|
# live on the order list, so pointing this at a vehicle that *shares* orders with another would
|
|
# silently rewrite that other vehicle's schedules too. (Cloning a vehicle without sharing orders
|
|
# gives an independent list, but copies the source's schedules along with it.)
|
|
DISPATCH_VEHICLE_ID = 14
|
|
|
|
|
|
# --- 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
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_add_and_remove_order(connected_owner_client, connected_admin):
|
|
# Public functions: add_order(), remove_order()
|
|
# Verified authoritatively via the admin get_timetable() order count. The test appends and
|
|
# inserts an order, then removes both, leaving the vehicle's order list as it started.
|
|
async def order_count():
|
|
data = await connected_admin.get_timetable(TIMETABLE_VEHICLE_ID, timeout=10.0)
|
|
return len(data["orders"])
|
|
|
|
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
|
|
before = await order_count()
|
|
|
|
# add_order input 1: append a goto-station order to the end of the list.
|
|
await connected_owner_client.add_order(TIMETABLE_VEHICLE_ID, ORDER_STATION_ID)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
assert await order_count() == before + 1
|
|
|
|
# add_order input 2: insert another before position 0.
|
|
await connected_owner_client.add_order(TIMETABLE_VEHICLE_ID, ORDER_STATION_ID, before_position=0)
|
|
await asyncio.sleep(1.0)
|
|
assert await order_count() == before + 2
|
|
|
|
# remove_order input 1: delete the one just inserted at the front.
|
|
await connected_owner_client.remove_order(TIMETABLE_VEHICLE_ID, 0)
|
|
await asyncio.sleep(1.0)
|
|
assert await order_count() == before + 1
|
|
|
|
# remove_order input 2: delete the appended order (now the last one) to restore the list.
|
|
await connected_owner_client.remove_order(TIMETABLE_VEHICLE_ID, before)
|
|
await asyncio.sleep(1.0)
|
|
assert not connected_owner_client.shutdown_event.is_set()
|
|
assert await order_count() == before
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_client_scheduled_dispatch_edit_and_view(connected_owner_client, connected_admin):
|
|
# Public functions: set_scheduled_dispatch(), add_dispatch_schedule(), remove_dispatch_schedule(),
|
|
# add_dispatch_slot(), remove_dispatch_slot(), clear_dispatch_schedule(), set_dispatch_duration(),
|
|
# set_dispatch_start_date(), and get_dispatch(). Edits go over the game port and are read back
|
|
# authoritatively via the admin get_dispatch(). The test leaves the vehicle with no schedules.
|
|
veh = DISPATCH_VEHICLE_ID
|
|
owner = connected_owner_client
|
|
await connected_admin.update_frequency(AdminUpdateType.Gamescript, AdminUpdateFrequency.Automatic)
|
|
|
|
async def dispatch():
|
|
return await connected_admin.get_dispatch(veh, timeout=10.0)
|
|
|
|
# The assertions below address schedules by absolute index, so DISPATCH_VEHICLE_ID must start
|
|
# with none of its own; the test restores that state on the way out.
|
|
start = await dispatch() # get_dispatch input 1: a valid vehicle
|
|
assert "schedules" in start and isinstance(start["schedules"], list)
|
|
|
|
# add_dispatch_schedule: two schedules (indices 0 and 1) with different start ticks/durations.
|
|
await owner.add_dispatch_schedule(veh, 0, 3000)
|
|
await asyncio.sleep(0.5)
|
|
await owner.add_dispatch_schedule(veh, 1000, 2000)
|
|
await asyncio.sleep(0.5)
|
|
assert not owner.shutdown_event.is_set()
|
|
data = await dispatch()
|
|
assert len(data["schedules"]) == 2
|
|
assert data["schedules"][0]["duration"] == 3000
|
|
assert data["schedules"][1]["duration"] == 2000
|
|
|
|
# set_dispatch_duration / set_dispatch_start_date: two inputs each (schedule 0 and 1).
|
|
await owner.set_dispatch_duration(veh, 0, 4000)
|
|
await owner.set_dispatch_duration(veh, 1, 2500)
|
|
await owner.set_dispatch_start_date(veh, 0, 1_000_000)
|
|
await owner.set_dispatch_start_date(veh, 1, 2_000_000)
|
|
await asyncio.sleep(0.5)
|
|
|
|
# add_dispatch_slot: two departure slots in schedule 0.
|
|
await owner.add_dispatch_slot(veh, 0, 500)
|
|
await owner.add_dispatch_slot(veh, 0, 1500)
|
|
await asyncio.sleep(0.5)
|
|
data = await dispatch()
|
|
sched0 = data["schedules"][0]
|
|
assert sched0["duration"] == 4000
|
|
# The engine normalises the start tick relative to current game time (advancing it by whole
|
|
# durations to sit near "now"), so it won't equal the requested value verbatim; just confirm
|
|
# a start date was accepted and is reported as an integer.
|
|
assert isinstance(sched0["start_tick"], int)
|
|
assert {s["offset"] for s in sched0["slots"]} == {500, 1500}
|
|
|
|
# remove_dispatch_slot: two inputs (both slots of schedule 0).
|
|
await owner.remove_dispatch_slot(veh, 0, 1500)
|
|
await owner.remove_dispatch_slot(veh, 0, 500)
|
|
await asyncio.sleep(0.5)
|
|
assert (await dispatch())["schedules"][0]["slots"] == []
|
|
|
|
# set_scheduled_dispatch: enable then disable (two inputs), reading the flag back in between.
|
|
await owner.set_scheduled_dispatch(veh, True)
|
|
await asyncio.sleep(0.5)
|
|
assert (await dispatch())["enabled"] == 1
|
|
await owner.set_scheduled_dispatch(veh, False)
|
|
await asyncio.sleep(0.5)
|
|
assert (await dispatch())["enabled"] == 0
|
|
|
|
# clear_dispatch_schedule: two inputs (schedule 0 and 1).
|
|
await owner.clear_dispatch_schedule(veh, 0)
|
|
await owner.clear_dispatch_schedule(veh, 1)
|
|
await asyncio.sleep(0.5)
|
|
|
|
# remove_dispatch_schedule: remove both (higher index first) to restore the vehicle.
|
|
await owner.remove_dispatch_schedule(veh, 1)
|
|
await asyncio.sleep(0.5)
|
|
await owner.remove_dispatch_schedule(veh, 0)
|
|
await asyncio.sleep(0.5)
|
|
assert not owner.shutdown_event.is_set()
|
|
assert (await dispatch())["schedules"] == [] # get_dispatch input 1 (restored state)
|
|
|
|
|
|
# --- 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_get_bridge_version_supported(connected_admin):
|
|
# Public function: get_bridge_version()
|
|
# Input 1: the default minimum, i.e. the version this client is written against.
|
|
#
|
|
# Every GameScript test below this one fails as an unexplained timeout if the server runs a
|
|
# stale AdminBridge (or none at all), because a bridge that does not know a command simply
|
|
# drops it. This test is the one that says so out loud, so run it first when the GS tests
|
|
# start hanging.
|
|
try:
|
|
data = await connected_admin.get_bridge_version(timeout=10.0)
|
|
except asyncio.TimeoutError:
|
|
pytest.fail(
|
|
"The server's AdminBridge GameScript did not answer get_version. It is older than "
|
|
f"version {GS_BRIDGE_VERSION} (which introduced the command), not loaded at all, or "
|
|
"the game is paused. See gamescript/AdminBridge/README.md.")
|
|
|
|
assert data["version"] >= GS_BRIDGE_VERSION
|
|
# The catalogues let a client feature-detect one command rather than compare versions.
|
|
assert "get_version" in data["commands"]
|
|
assert set(data["events"]) == {e.value for e in GameEventType} - {GameEventType.EventsDropped}
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_get_bridge_version_minimum_zero(connected_admin):
|
|
# Public function: get_bridge_version()
|
|
# Input 2: minimum=0 -> read the version without requiring anything of it.
|
|
data = await connected_admin.get_bridge_version(minimum=0, timeout=10.0)
|
|
assert isinstance(data["version"], int)
|
|
assert "get_timetable" in data["commands"]
|
|
|
|
@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_get_dispatch_invalid_vehicle(connected_admin):
|
|
# Public function: get_dispatch()
|
|
# Input 2: an id no vehicle can have -> GameScript reports invalid_vehicle
|
|
with pytest.raises(ValueError, match="invalid_vehicle"):
|
|
await connected_admin.get_dispatch(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)
|
|
|
|
|
|
# --- Game Events ---
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_subscribe_events_all_kinds(connected_admin):
|
|
# Public functions: subscribe_events(), unsubscribe_events()
|
|
# Input 1: no arguments -> every event kind, default interval
|
|
data = await connected_admin.subscribe_events(timeout=10.0)
|
|
assert isinstance(data["events"], list)
|
|
assert "vehicle_arrive" in data["events"] and "cargo_waiting" in data["events"]
|
|
assert data["interval"] == 10
|
|
|
|
stopped = await connected_admin.unsubscribe_events()
|
|
assert stopped["events"] == []
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_subscribe_events_filtered(connected_admin):
|
|
# Public functions: subscribe_events(), unsubscribe_events()
|
|
# Input 2: a narrowed subscription -> only the requested kinds come back
|
|
data = await connected_admin.subscribe_events(
|
|
events=[GameEventType.VehicleArrive, GameEventType.CargoWaiting],
|
|
interval=5, company_id=0, min_cargo_delta=2, include_cargo=False, timeout=10.0)
|
|
assert sorted(data["events"]) == ["cargo_waiting", "vehicle_arrive"]
|
|
assert data["interval"] == 5
|
|
|
|
await connected_admin.unsubscribe_events(timeout=10.0)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_subscribe_events_unknown_kind(connected_admin):
|
|
# Public function: subscribe_events()
|
|
# A kind the GameScript does not know -> it reports unknown_event
|
|
with pytest.raises(ValueError, match="unknown_event"):
|
|
await connected_admin.subscribe_events(events=["definitely_not_an_event"], timeout=10.0)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_wait_for_event_vehicle_reaches_a_stop(connected_admin):
|
|
# Public function: wait_for_event()
|
|
# Input 1: filtered by kind. Needs traffic on the server, so a quiet map skips.
|
|
await connected_admin.subscribe_events(
|
|
events=[GameEventType.VehicleArrive, GameEventType.VehicleDepart],
|
|
interval=2, timeout=10.0)
|
|
try:
|
|
event = await connected_admin.wait_for_event(
|
|
{GameEventType.VehicleArrive, GameEventType.VehicleDepart}, timeout=60.0)
|
|
except asyncio.TimeoutError:
|
|
pytest.skip("No vehicle reached or left a stop on the test server within the timeout.")
|
|
finally:
|
|
await connected_admin.unsubscribe_events()
|
|
|
|
assert event["event"] in ("vehicle_arrive", "vehicle_depart")
|
|
for key in ("tick", "vehicle_id", "station_id", "owner", "vehicle_type", "order_position"):
|
|
assert key in event
|
|
if event["event"] == "vehicle_depart":
|
|
assert event["dwell"] >= 0
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_wait_for_event_any_kind(connected_admin):
|
|
# Public function: wait_for_event()
|
|
# Input 2: no kind filter -> whatever the game produces first
|
|
await connected_admin.subscribe_events(interval=2, timeout=10.0)
|
|
try:
|
|
event = await connected_admin.wait_for_event(timeout=60.0)
|
|
except asyncio.TimeoutError:
|
|
pytest.skip("Nothing happened on the test server within the timeout.")
|
|
finally:
|
|
await connected_admin.unsubscribe_events()
|
|
|
|
assert "event" in event and "tick" in event
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_unsubscribe_events_stops_the_stream(connected_admin):
|
|
# Public function: unsubscribe_events()
|
|
# After unsubscribing the bridge must go quiet, so a fresh wait times out.
|
|
await connected_admin.subscribe_events(interval=2, timeout=10.0)
|
|
await connected_admin.unsubscribe_events(timeout=10.0)
|
|
|
|
connected_admin._event_buffer.clear() # drop anything delivered before we unsubscribed
|
|
with pytest.raises(asyncio.TimeoutError):
|
|
await connected_admin.wait_for_event(timeout=5.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, _ = 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, _ = 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
|