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