55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
import asyncio
|
|
import pytest
|
|
import sys
|
|
import os
|
|
|
|
# Add lib to path
|
|
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'lib'))
|
|
|
|
from openttd import OpenTTDClient
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_connection_and_join():
|
|
# Configuration matches your local server
|
|
SERVER_IP = "127.0.0.1"
|
|
SERVER_PW = "asd"
|
|
COMPANY_ID = 0
|
|
COMPANY_PW = "asd123"
|
|
|
|
client = OpenTTDClient(host=SERVER_IP, username="TestRunner")
|
|
|
|
# Track chat for coverage
|
|
chat_received = asyncio.Event()
|
|
def chat_handler(cid, msg):
|
|
chat_received.set()
|
|
client.on_chat = chat_handler
|
|
|
|
try:
|
|
# 1. Connect
|
|
await client.connect(server_password=SERVER_PW)
|
|
|
|
# 2. Join company
|
|
await client.join_company(company_id=COMPANY_ID, company_password=COMPANY_PW)
|
|
|
|
# 3. Wait for join (timeout after 15s to be safe)
|
|
await asyncio.wait_for(client.joined.wait(), timeout=15.0)
|
|
|
|
assert client.joined.is_set()
|
|
assert client.client_id is not None
|
|
|
|
# 4. Stay briefly to ensure keep-alive/frames work
|
|
await asyncio.sleep(2)
|
|
|
|
# 5. Graceful Quit
|
|
await client.quit()
|
|
|
|
# 6. Wait for shutdown event
|
|
await asyncio.wait_for(client.shutdown_event.wait(), timeout=5.0)
|
|
assert client.shutdown_event.is_set()
|
|
|
|
except Exception as e:
|
|
pytest.fail(f"E2E Test failed: {e}")
|
|
finally:
|
|
if not client.shutdown_event.is_set():
|
|
await client.quit()
|