openttd-client/tests/test_logic.py

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, 11) # NameInUse
    await client.receive_ServerError(None, 17) # Timeout

@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_ServerCommand(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()