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 # --- 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 # --- 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) # --- 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