kovagoadi
2eea541158
Editing (game port, OpenTTDClient): a core of JGRPP's scheduled dispatch DoCommands — set_scheduled_dispatch (enable/disable), add/remove schedule, add/remove/clear slots, and set duration/start date. Adds the command IDs to protocol.py. Viewing (admin, OpenTTDAdminClient.get_dispatch): the GameScript API has no dispatch support, so a new server patch (docker/patches/0002-*) adds read-only GSOrder.GetScheduledDispatch* / IsScheduledDispatchEnabled getters, an AdminBridge GameScript get_dispatch handler exposes them, and get_dispatch() returns the live schedules and slots (mirrors get_timetable). Note: set_dispatch_start_date values are normalised by the engine relative to current game time, so they read back offset from the requested value. Includes unit + e2e tests, a demo in main.py, and protocol/timetable docs. The AdminBridge GameScript and the patched OpenTTD-patches clone live outside this repo; the 0002 patch file is the durable source for the latter. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
773 lines
32 KiB
Python
773 lines
32 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
|
|
# A station TIMETABLE_VEHICLE_ID can legally serve, used for add_order/remove_order tests.
|
|
ORDER_STATION_ID = 6
|
|
|
|
|
|
# --- 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 = TIMETABLE_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)
|
|
|
|
start = await dispatch() # get_dispatch input 1: a valid vehicle
|
|
assert "schedules" in start and isinstance(start["schedules"], list)
|
|
if start["schedules"]:
|
|
pytest.skip("Test vehicle already has dispatch schedules; expected a clean vehicle.")
|
|
|
|
# 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)
|
|
|
|
|
|
# --- 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
|