Commonwealth-Online-Server/server/tests/test_gns_gameplay_adapter.py

from __future__ import annotations

import json
from collections import deque

from gns_gameplay_adapter import GnsGameplayAdapter
from gns_transport import EventType, GnsEvent, RemoteEndpoint, SendResult
from packet_codec import encode_packet
from server_core import FalloutTogetherServer, PROTOCOL_VERSION
from transport_policy import Delivery


class FakeGnsTransport:
    def __init__(self):
        self.events = deque()
        self.endpoints = {}
        self.sent = []
        self.disconnected = []
        self.closed = False

    def poll(self):
        return self.events.popleft() if self.events else None

    def remote_endpoint(self, connection_id):
        return self.endpoints.get(connection_id)

    def send_encoded(self, connection_id, encoded):
        self.sent.append((connection_id, encoded))
        return SendResult.SENT

    def send_packet(self, connection_id, packet):
        return self.send_encoded(connection_id, encode_packet(packet))

    def disconnect(self, connection_id, *, reason=0, debug=""):
        self.disconnected.append((connection_id, reason, debug))
        return True

    def close(self):
        self.closed = True


def packet_types_sent(transport: FakeGnsTransport, connection_id: int) -> list[str]:
    return [
        json.loads(encoded.payload)["type"]
        for target, encoded in transport.sent
        if target == connection_id
    ]


def queue_connect(transport: FakeGnsTransport, connection_id: int, port: int = 50000):
    transport.endpoints[connection_id] = RemoteEndpoint("127.0.0.1", port)
    transport.events.append(GnsEvent(EventType.CONNECTED, connection_id))


def queue_packet(transport: FakeGnsTransport, connection_id: int, packet: dict):
    transport.events.append(
        GnsEvent(
            EventType.MESSAGE,
            connection_id,
            json.dumps(packet, separators=(",", ":")).encode(),
        )
    )


def complete_handshake(adapter: GnsGameplayAdapter, transport: FakeGnsTransport, connection_id: int):
    queue_connect(transport, connection_id, 50000 + connection_id)
    assert adapter.pump_once() == 1
    assert "welcome" in packet_types_sent(transport, connection_id)
    queue_packet(transport, connection_id, {"type": "hello", "protocolVersion": PROTOCOL_VERSION})
    assert adapter.pump_once() == 1
    assert "sessionReady" in packet_types_sent(transport, connection_id)


def test_gns_adapter_reuses_v2_admission_and_server_owned_identity():
    server = FalloutTogetherServer(host="127.0.0.1", port=0)
    transport = FakeGnsTransport()
    adapter = GnsGameplayAdapter(server, transport)

    complete_handshake(adapter, transport, 101)
    client = adapter._client_for_connection(101)
    assert client is not None
    assert client.gameplay_active
    assert client.protocol_version == PROTOCOL_VERSION

    queue_packet(
        transport,
        101,
        {
            "type": "transform",
            "playerId": 999999,
            "x": 1.0,
            "y": 2.0,
            "z": 3.0,
            "angleZ": 4.0,
            "cellId": "00000010",
            "worldspaceId": "0000003C",
            "movementType": "normal",
        },
    )
    adapter.pump_once()
    assert client.last_transform is not None
    assert client.last_transform["playerId"] == client.player_id
    assert client.last_transform["playerId"] != 999999


def test_gns_adapter_preserves_reliable_and_snapshot_delivery_policy():
    server = FalloutTogetherServer(host="127.0.0.1", port=0)
    transport = FakeGnsTransport()
    adapter = GnsGameplayAdapter(server, transport)
    complete_handshake(adapter, transport, 201)
    complete_handshake(adapter, transport, 202)

    transport.sent.clear()
    queue_packet(
        transport,
        202,
        {
            "type": "transform",
            "x": 10.0,
            "y": 0.0,
            "z": 0.0,
            "angleZ": 0.0,
            "cellId": "00000010",
            "worldspaceId": "0000003C",
            "movementType": "normal",
        },
    )
    adapter.pump_once()
    queue_packet(
        transport,
        201,
        {
            "type": "transform",
            "x": 0.0,
            "y": 0.0,
            "z": 0.0,
            "angleZ": 0.0,
            "cellId": "00000010",
            "worldspaceId": "0000003C",
            "movementType": "normal",
        },
    )
    adapter.pump_once()

    transform_relays = [
        encoded
        for target, encoded in transport.sent
        if target == 202 and json.loads(encoded.payload).get("type") == "transform"
    ]
    assert transform_relays
    assert transform_relays[-1].delivery is Delivery.UNRELIABLE_SEQUENCED

    queue_packet(
        transport,
        201,
        {"type": "playerState", "characterName": "Nomad", "actionEvents": []},
    )
    adapter.pump_once()
    player_state_relays = [
        encoded
        for target, encoded in transport.sent
        if target == 202 and json.loads(encoded.payload).get("type") == "playerState"
    ]
    assert player_state_relays
    assert player_state_relays[-1].delivery is Delivery.RELIABLE_ORDERED


def test_gns_adapter_enforces_existing_ban_policy(tmp_path):
    server = FalloutTogetherServer(host="127.0.0.1", port=0, bans_path=str(tmp_path / "bans.json"))
    server.ban_ip("127.0.0.1", "test ban")
    transport = FakeGnsTransport()
    adapter = GnsGameplayAdapter(server, transport)

    queue_connect(transport, 301)
    adapter.pump_once()
    assert adapter._client_for_connection(301) is None
    assert transport.disconnected
    session_ended = [
        json.loads(encoded.payload)
        for target, encoded in transport.sent
        if target == 301 and json.loads(encoded.payload).get("type") == "sessionEnded"
    ]
    assert session_ended
    assert session_ended[-1]["code"] == "banned"
    assert server.get_stats()["bannedConnectionsRejected"] == 1


def test_gns_adapter_ends_session_on_native_oversize_event():
    server = FalloutTogetherServer(host="127.0.0.1", port=0)
    transport = FakeGnsTransport()
    adapter = GnsGameplayAdapter(server, transport)
    complete_handshake(adapter, transport, 401)

    transport.events.append(
        GnsEvent(EventType.OVERSIZE_MESSAGE, 401, reason=0, debug="too large")
    )
    adapter.pump_once()
    assert adapter._client_for_connection(401) is None
    assert transport.disconnected
    ended_packets = [
        json.loads(encoded.payload)
        for target, encoded in transport.sent
        if target == 401 and json.loads(encoded.payload).get("type") == "sessionEnded"
    ]
    assert ended_packets
    assert ended_packets[-1]["code"] == "packet_too_large"