kovagoadi
b33869334a
Timetables have no GameScript API surface, so this implements real DoCommands over the game port (ClientCommand/ServerCommand) instead of the Admin GameScript relay used for list_vehicles(): change_timetable(), autofill_timetable(), set_timetable_start(), and set_vehicle_on_time() send commands, while get_vehicle_timetable() reconstructs state purely by observing ServerCommand broadcasts, since no query command exists. Includes the custom varuint wire codec these commands require, a full usage guide (docs/TIMETABLES.md), and a worked demo in main.py. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
239 lines
8.4 KiB
Python
239 lines
8.4 KiB
Python
import pytest
|
|
import asyncio
|
|
import hashlib
|
|
from openttd.protocol import PacketGameType
|
|
from openttd.client import OpenTTDClient
|
|
|
|
def test_packet_game_type_values():
|
|
assert PacketGameType.ServerFull == 0
|
|
assert PacketGameType.ClientJoin == 2
|
|
assert PacketGameType.ServerWelcome == 21
|
|
assert PacketGameType.ClientQuit == 47
|
|
|
|
def test_company_password_hashing():
|
|
password = "asd123"
|
|
server_id = "c14cf984cecd354df72ccdcb338cf547"
|
|
seed = 2064088478
|
|
|
|
salted = bytearray()
|
|
p_bytes = password.encode('utf-8')
|
|
s_bytes = server_id.encode('utf-8')
|
|
for i in range(32):
|
|
p_char = p_bytes[i] if i < len(p_bytes) else 0
|
|
s_char = s_bytes[i] if i < len(s_bytes) else 0
|
|
seed_char = (seed >> (i % 32)) & 0xFF
|
|
salted.append(p_char ^ s_char ^ seed_char)
|
|
expected_hash = hashlib.md5(salted, usedforsecurity=False).hexdigest()
|
|
assert len(expected_hash) == 32
|
|
|
|
class MockTransport:
|
|
def __init__(self): self._closing = False
|
|
def is_closing(self): return self._closing
|
|
def close(self): self._closing = True
|
|
def write(self, data): return len(data)
|
|
|
|
class MockProtocol:
|
|
def __init__(self): self.sent = []
|
|
async def send_packet(self, data):
|
|
self.sent.append(data)
|
|
return len(data)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_connect_success(monkeypatch):
|
|
# Coverage for client.py:48-50
|
|
client = OpenTTDClient(host="127.0.0.1")
|
|
|
|
class FakeProto:
|
|
def __init__(self): self.sent = []
|
|
async def send_packet(self, data): self.sent.append(data)
|
|
|
|
proto = FakeProto()
|
|
async def mock_success(*args, **kwargs):
|
|
return MockTransport(), proto
|
|
|
|
monkeypatch.setattr(asyncio.get_running_loop(), "create_connection", mock_success)
|
|
|
|
await client.connect()
|
|
assert len(proto.sent) == 1 # ClientGameInfo sent
|
|
assert client._protocol == proto
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_connect_failure(monkeypatch):
|
|
client = OpenTTDClient(host="127.0.0.1")
|
|
async def mock_fail(*args, **kwargs):
|
|
raise Exception("Async Failure")
|
|
monkeypatch.setattr(asyncio.get_running_loop(), "create_connection", mock_fail)
|
|
with pytest.raises(Exception, match="Async Failure"):
|
|
await client.connect()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_error_handling():
|
|
client = OpenTTDClient(host="127.0.0.1")
|
|
client._transport = MockTransport()
|
|
client._protocol = MockProtocol()
|
|
|
|
await client.receive_ServerError(None, 8) # WrongRevision
|
|
assert client.shutdown_event.is_set()
|
|
|
|
client.shutdown_event.clear()
|
|
await client.receive_ServerError(None, 10) # WrongPassword
|
|
await client.receive_ServerError(None, 9) # NameInUse
|
|
await client.receive_ServerError(None, 11) # CompanyMismatch
|
|
await client.receive_ServerError(None, 17) # TimeoutComputer
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_server_full_banned():
|
|
client = OpenTTDClient(host="127.0.0.1")
|
|
client._transport = MockTransport()
|
|
client._protocol = MockProtocol()
|
|
|
|
await client.receive_ServerFull(None)
|
|
await client.receive_ServerBanned(None)
|
|
client.connected(None)
|
|
client.disconnect(None)
|
|
assert client.shutdown_event.is_set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chat_callback():
|
|
client = OpenTTDClient(host="127.0.0.1")
|
|
received = []
|
|
def on_chat(cid, msg):
|
|
received.append((cid, msg))
|
|
client.on_chat = on_chat
|
|
|
|
await client.receive_ServerChat(None, 42, "Hello World")
|
|
assert received == [(42, "Hello World")]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_handlers():
|
|
client = OpenTTDClient(host="127.0.0.1")
|
|
client.log.setLevel(100)
|
|
|
|
await client.receive_ServerUnused(None)
|
|
await client.receive_ServerSync(None)
|
|
await client.receive_ServerClientJoined(None)
|
|
await client.receive_ServerMapBegin(None)
|
|
await client.receive_ServerMapSize(None, size=100)
|
|
await client.receive_ServerMapData(None, data=b"data")
|
|
await client.receive_ServerConfigurationUpdate(None)
|
|
await client.receive_ServerClientInfo(None)
|
|
await client.receive_ServerExternalChat(None)
|
|
await client.receive_ClientAck(None)
|
|
await client.receive_ClientIdentify(None)
|
|
await client.receive_ServerCompanyUpdate(None)
|
|
|
|
client.joined.set()
|
|
await client.join_company(0)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_full_handshake_flow():
|
|
client = OpenTTDClient(host="127.0.0.1", username="TestUser")
|
|
client._protocol = MockProtocol()
|
|
client._transport = MockTransport()
|
|
|
|
await client.receive_ServerGameInfo(None, name="TestSrv", openttd_version="14.0")
|
|
assert len(client._protocol.sent) == 1
|
|
|
|
pake_data = b"S" * 32 + b"N" * 24
|
|
await client.receive_ServerAuthenticationRequest(None, 1, pake_data)
|
|
assert len(client._protocol.sent) == 2
|
|
|
|
await client.receive_ServerEnableEncryption(None, b"E" * 24)
|
|
assert client.encryption_enabled
|
|
assert len(client._protocol.sent) == 3
|
|
|
|
await client.receive_ServerCheckNewGRFs(None)
|
|
assert len(client._protocol.sent) == 4
|
|
|
|
await client.join_company(0, "comp_pw")
|
|
await client.receive_ServerNeedCompanyPassword(None, 1234, "srv_id")
|
|
assert len(client._protocol.sent) == 5
|
|
|
|
await client.receive_ServerWelcome(None, client_id=42)
|
|
assert client.client_id == 42
|
|
assert len(client._protocol.sent) == 6
|
|
|
|
await client.receive_ServerMapDone(None)
|
|
assert len(client._protocol.sent) == 7
|
|
assert client.joined.is_set()
|
|
|
|
await client.receive_ServerFrame(None, 100, 7)
|
|
assert len(client._protocol.sent) == 8
|
|
|
|
await client.quit()
|
|
assert len(client._protocol.sent) == 9
|
|
|
|
client._transport.close()
|
|
await client.quit()
|
|
|
|
|
|
# --- Game Client Edge Cases & Callback Unit Tests ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unit_client_connection_failure(server_config):
|
|
client = OpenTTDClient(host=server_config["host"], port=9999)
|
|
with pytest.raises(OSError):
|
|
await client.connect()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unit_client_already_joined_warning(server_config):
|
|
client = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
client.joined.set()
|
|
await client.join_company(255)
|
|
assert client.joined.is_set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unit_client_disconnect(server_config):
|
|
client = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
client.disconnect(None)
|
|
assert client.shutdown_event.is_set()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unit_client_receive_chat_no_callback(server_config):
|
|
client = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
client.on_chat = None
|
|
await client.receive_ServerChat(None, 1, "hello")
|
|
assert client.on_chat is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unit_client_receive_chat_with_callback(server_config):
|
|
client = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
chats = []
|
|
client.on_chat = lambda cid, msg: chats.append((cid, msg))
|
|
await client.receive_ServerChat(None, 2, "world")
|
|
assert chats == [(2, "world")]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unit_client_receive_auth_request_invalid_type(server_config):
|
|
client = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
await client.receive_ServerAuthenticationRequest(None, 2, b"")
|
|
assert not client.encryption_enabled
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unit_client_noop_callbacks(server_config):
|
|
client = OpenTTDClient(host=server_config["host"], port=server_config["game_port"])
|
|
client.connected(None)
|
|
await client.receive_ServerUnused(None)
|
|
await client.receive_ServerCompanyUpdate(None)
|
|
await client.receive_ServerClientInfo(None)
|
|
await client.receive_ServerSync(None)
|
|
await client.receive_ServerClientJoined(None)
|
|
await client.receive_ServerMapBegin(None)
|
|
await client.receive_ServerMapSize(None)
|
|
await client.receive_ServerMapData(None)
|
|
await client.receive_ServerConfigurationUpdate(None)
|
|
await client.receive_ServerExternalChat(None)
|
|
await client.receive_ServerFull(None)
|
|
await client.receive_ServerBanned(None)
|
|
await client.receive_ClientAck(None)
|
|
await client.receive_ClientIdentify(None)
|
|
assert not client.shutdown_event.is_set()
|
|
assert not client.joined.is_set()
|
|
|
|
|
|
def test_unit_exclude_call_check_decorator():
|
|
from openttd import exclude_call_check
|
|
@exclude_call_check
|
|
def dummy(): pass
|
|
assert dummy.__exclude_call_check__ is True
|