kovagoadiandClaude
ba26b59c40
The station queries and the cargo events name a cargo only by a bare numeric id -- get_station()'s and get_station_cargo()'s cargo_id, the cargo_waiting events, the per-cargo load on vehicle events. Those ids index the cargo table the loaded NewGRFs build for the running game, so the same id is coal in one save and grain in another and callers had no way to resolve them. The AdminBridge GameScript has answered a list_cargo command all along; no method on OpenTTDAdminClient sent it. This adds the missing half, so no GameScript change is needed for it to work. It goes through _gs_query() like get_station(), inheriting the request_id correlation and the Gamescript auto-subscribe, with one difference worth knowing: the GS handler defines no error reply for this command, so unlike the other queries it can time out but can never raise ValueError. The reply lists cargo in GSCargoList order rather than by id -- against the dev server the ids come back 10 down to 0 -- so the docstring and PROTOCOL.md both warn to index the list by cargo_id and not by position. Also teaches the main_admin.py demo to resolve the labels before printing a station's cargo, which is what the bare ids in its output were asking for all along. Co-Authored-By: Claude <[email protected]>
906 lines
38 KiB
Python
906 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 (
|
|
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_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)
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_list_cargo_table(connected_admin):
|
|
# Public function: list_cargo()
|
|
# Input 1: an explicit timeout. Every game has a cargo table, so an empty list would be a bug.
|
|
data = await connected_admin.list_cargo(timeout=10.0)
|
|
assert isinstance(data["cargo"], list) and data["cargo"]
|
|
for cargo in data["cargo"]:
|
|
for key in ("cargo_id", "label", "freight"):
|
|
assert key in cargo
|
|
assert cargo["freight"] in (0, 1)
|
|
ids = [cargo["cargo_id"] for cargo in data["cargo"]]
|
|
assert len(ids) == len(set(ids))
|
|
# Any cargo set carries passengers as well as freight, so both kinds must show up.
|
|
assert any(cargo["freight"] == 0 for cargo in data["cargo"])
|
|
assert any(cargo["freight"] == 1 for cargo in data["cargo"])
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_admin_list_cargo_resolves_station_cargo_ids(connected_admin):
|
|
# Public function: list_cargo()
|
|
# Input 2: the default timeout. The point of the call: naming the bare ids get_station() returns.
|
|
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.")
|
|
|
|
labels = {cargo["cargo_id"]: cargo["label"] for cargo in (await connected_admin.list_cargo())["cargo"]}
|
|
detail = await connected_admin.get_station(stations[0]["id"], timeout=10.0)
|
|
for cargo in detail["cargo"]:
|
|
assert cargo["cargo_id"] in labels
|
|
|
|
|
|
# --- 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
|