import asyncio import pytest import pytest_asyncio import sys import os import random from unittest.mock import MagicMock, AsyncMock # 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, PacketAdminType ) # --- 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() # ============================================================================== # --- 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() # --- 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) # --- 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