kovagoadiandClaude
25953bea06
Resolve 51 findings from the I/RUF/BLE/TRY002/S110/PLR0402 rule set: - Sort imports and __all__ (I001, RUF022, PLR0402). The sys.path.insert calls in check_public_calls.py and tests/test_e2e.py still precede the openttd imports that depend on them. - Replace unused unpacked values with _ (RUF059) and annotate the two timetable lookup tables as ClassVar (RUF012). - Narrow the best-effort excepts in OpenTTDClient.quit and OpenTTDAdminClient.quit to (OSError, SocketClosed) and log at debug rather than swallowing silently (BLE001, S110). The test doubles now raise an OSError subclass so they still exercise that branch. - Narrow the gamescript JSON fallback to json.JSONDecodeError. The broad catch in receive_packet keeps a noqa: it guards untrusted wire data and must degrade to a no-op packet instead of killing the connection. - Use contextlib.suppress instead of try/except/pass in tests. ruff check . is clean, 102 tests pass, coverage stays at 100%. Co-Authored-By: Claude <[email protected]>
775 lines
32 KiB
Python
775 lines
32 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,
|
|
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
|
|
|
|
|
|
# --- 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, _ = 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
|