Test Protocol V2 synthetic client

server/tests/test_protocol_v2_client.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import socket
+import threading
+import time
+
+from protocol_v2_client import PROTOCOL_VERSION, ProtocolV2Client
+from server_core import FalloutTogetherServer
+
+
+def _start_server(max_players: int = 4) -> tuple[FalloutTogetherServer, threading.Thread]:
+    server = FalloutTogetherServer(host="127.0.0.1", port=0, max_players=max_players)
+    server._prepare_server_socket()
+    thread = threading.Thread(target=server._accept_loop, daemon=True)
+    thread.start()
+    return server, thread
+
+
+def _stop_server(server: FalloutTogetherServer, thread: threading.Thread) -> None:
+    server.stop()
+    thread.join(timeout=3.0)
+
+
+def _recv_until(client: ProtocolV2Client, packet_type: str, timeout: float = 2.0) -> dict:
+    assert client.socket is not None
+    previous_timeout = client.socket.gettimeout()
+    client.socket.settimeout(0.25)
+    deadline = time.monotonic() + timeout
+    try:
+        while time.monotonic() < deadline:
+            try:
+                packet = client.recv_packet()
+            except socket.timeout:
+                continue
+            if packet.get("type") == packet_type:
+                return packet
+    finally:
+        client.socket.settimeout(previous_timeout)
+    raise AssertionError(f"did not receive packet type {packet_type}")
+
+
+def test_synthetic_client_uses_protocol_v2_without_legacy_activation() -> None:
+    server, thread = _start_server()
+    client = ProtocolV2Client("127.0.0.1", server.port, name="pytest-v2")
+    try:
+        session = client.connect()
+        assert session.player_id == 1
+        assert session.server_protocol_version == PROTOCOL_VERSION
+
+        stats = server.get_stats()
+        assert stats["connectedClients"] == 1
+        assert stats["protocolV2Connections"] == 1
+        assert stats["legacyConnections"] == 0
+        assert stats["pendingConnections"] == 0
+    finally:
+        client.close()
+        _stop_server(server, thread)
+
+
+def test_same_cell_v2_clients_receive_each_others_transforms() -> None:
+    server, thread = _start_server()
+    first = ProtocolV2Client("127.0.0.1", server.port, name="first")
+    second = ProtocolV2Client("127.0.0.1", server.port, name="second")
+    try:
+        first.connect()
+        second.connect()
+
+        # The first client receives a host-assignment control packet after sessionReady.
+        _recv_until(first, "worldStateHost")
+
+        second.send_transform(
+            x=100.0,
+            y=200.0,
+            z=300.0,
+            angle_z=0.5,
+            cell_id=0x0000003C,
+            movement_speed=150.0,
+            animation_direction=90.0,
+            is_moving=True,
+        )
+        relayed = _recv_until(first, "transform")
+        assert relayed["playerId"] == second.session.player_id
+        assert relayed["cellId"] == "0000003C"
+        assert relayed["animationDirection"] == 90.0
+    finally:
+        first.close()
+        second.close()
+        _stop_server(server, thread)