from __future__ import annotations import json import socket import time import pytest from server_core import PROTOCOL_VERSION, FalloutTogetherServer, states_share_interest _RECV_BUFFERS: dict[socket.socket, bytes] = {} def recv_packet(sock: socket.socket, timeout: float = 2.0) -> dict: sock.settimeout(timeout) data = _RECV_BUFFERS.get(sock, b"") while b"\n" not in data: chunk = sock.recv(4096) if not chunk: raise ConnectionError("socket closed before a complete packet was received") data += chunk line, remainder = data.split(b"\n", 1) _RECV_BUFFERS[sock] = remainder return json.loads(line.decode("utf-8")) def recv_until(sock: socket.socket, packet_type: str, timeout: float = 2.0) -> dict: deadline = time.monotonic() + timeout while time.monotonic() < deadline: remaining = max(0.01, deadline - time.monotonic()) try: packet = recv_packet(sock, timeout=remaining) except socket.timeout: break if packet.get("type") == packet_type: return packet raise AssertionError(f"did not receive packet type {packet_type}") def assert_no_packet_type(sock: socket.socket, packet_type: str, timeout: float = 0.2) -> None: with pytest.raises(AssertionError): recv_until(sock, packet_type, timeout=timeout) def send_packet(sock: socket.socket, packet: dict) -> None: sock.sendall(json.dumps(packet, separators=(",", ":")).encode() + b"\n") def connect(server: FalloutTogetherServer) -> tuple[socket.socket, dict]: sock = socket.create_connection(("127.0.0.1", server.port), timeout=2.0) return sock, recv_packet(sock) def hello(sock: socket.socket) -> dict: send_packet(sock, {"type": "hello", "protocolVersion": PROTOCOL_VERSION}) return recv_until(sock, "sessionReady") def transform( cell: str, x: float, y: float, world: str = "0000003C", movement_type: str = "normal", ) -> dict: return { "type": "transform", "x": x, "y": y, "z": 0.0, "angleZ": 0.0, "cellId": cell, "worldspaceId": world, "movementType": movement_type, } def start_server(max_players: int = 16) -> FalloutTogetherServer: server = FalloutTogetherServer(host="127.0.0.1", port=0, max_players=max_players) server.start() deadline = time.time() + 2 while not server.is_running() and time.time() < deadline: time.sleep(0.01) return server def test_interest_exact_cell_and_exterior_radius(): a = transform("00000001", 0, 0) same = transform("00000001", 50000, 50000) near = transform("00000002", 1000, 1000) far = transform("00000002", 20000, 20000) other_world = transform("00000002", 1000, 1000, world="0000003D") assert states_share_interest(a, same) assert states_share_interest(a, near) assert not states_share_interest(a, far) assert not states_share_interest(a, other_world) def test_recv_packet_preserves_coalesced_lines(): reader, writer = socket.socketpair() try: writer.sendall(b'{"type":"first"}\n{"type":"second"}\n') assert recv_packet(reader)["type"] == "first" assert recv_packet(reader)["type"] == "second" finally: _RECV_BUFFERS.pop(reader, None) reader.close() writer.close() def test_idle_transport_does_not_take_world_authority(): server = start_server() idle, _ = connect(server) active, active_welcome = connect(server) ready = hello(active) assert ready["playerId"] == active_welcome["playerId"] stats = server.get_stats() assert stats["connectedClients"] == 1 assert stats["pendingConnections"] == 1 assert server._world_state_host_player_id == active_welcome["playerId"] idle.close() active.close() server.stop() def test_max_players_applies_to_activated_sessions_not_probes(): server = start_server(max_players=1) first, _ = connect(server) hello(first) second, _ = connect(server) send_packet(second, {"type": "hello", "protocolVersion": PROTOCOL_VERSION}) ended = recv_until(second, "sessionEnded") assert ended["code"] == "server_full" first.close() second.close() server.stop() def test_protocol_mismatch_is_rejected(): server = start_server() sock, _ = connect(server) send_packet(sock, {"type": "hello", "protocolVersion": PROTOCOL_VERSION + 1}) ended = recv_until(sock, "sessionEnded") assert ended["code"] == "protocol_mismatch" sock.close() server.stop() def test_nan_transform_is_rejected_and_not_cached(): server = start_server() sock, welcome = connect(server) hello(sock) sock.sendall( b'{"type":"transform","x":NaN,"y":0,"z":0,"angleZ":0,"cellId":"00000001"}\n' ) time.sleep(0.05) client = server._find_client_by_player_id(welcome["playerId"]) assert client is not None assert client.last_transform is None assert server.get_stats()["packetsRejected"] >= 1 sock.close() server.stop() def test_transform_interest_filters_distant_peer(): server = start_server() a, _ = connect(server) hello(a) b, _ = connect(server) hello(b) send_packet(b, transform("00000020", 25000, 25000)) time.sleep(0.05) send_packet(a, transform("00000010", 0, 0)) assert_no_packet_type(b, "transform") assert server.get_stats()["transformPacketsInterestFiltered"] >= 1 a.close() b.close() server.stop() def test_rejected_normal_teleport_is_corrected_and_never_relayed(): server = start_server() a, a_welcome = connect(server) hello(a) b, _ = connect(server) hello(b) send_packet(b, transform("00000010", 64.0, 0.0)) recv_until(a, "transform") send_packet(a, transform("00000010", 0.0, 0.0)) first_relay = recv_until(b, "transform") assert first_relay["x"] == 0.0 send_packet(a, transform("00000010", 100000.0, 0.0)) correction = recv_until(a, "positionCorrection") assert correction["x"] == 0.0 assert correction["cellId"] == "00000010" assert_no_packet_type(b, "transform") client = server._find_client_by_player_id(a_welcome["playerId"]) assert client is not None assert client.last_transform is not None assert client.last_transform["x"] == 0.0 stats = server.get_stats() assert stats["movementPacketsRejected"] >= 1 assert stats["movementCorrectionsSent"] >= 1 a.close() b.close() server.stop() def test_scope_change_requires_explicit_transition(): server = start_server() sock, welcome = connect(server) hello(sock) send_packet(sock, transform("00000010", 0.0, 0.0)) time.sleep(0.05) send_packet(sock, transform("00000011", 10.0, 0.0)) correction = recv_until(sock, "positionCorrection") assert correction["cellId"] == "00000010" client = server._find_client_by_player_id(welcome["playerId"]) assert client is not None assert client.last_transform is not None assert client.last_transform["cellId"] == "00000010" send_packet(sock, transform("00000011", 10.0, 0.0, movement_type="cell_change")) deadline = time.monotonic() + 1.0 while time.monotonic() < deadline: if client.last_transform is not None and client.last_transform.get("cellId") == "00000011": break time.sleep(0.01) assert client.last_transform is not None assert client.last_transform["cellId"] == "00000011" sock.close() server.stop() def test_zero_elapsed_burst_cannot_bypass_movement_envelope(): server = start_server() sock, welcome = connect(server) hello(sock) send_packet(sock, transform("00000010", 0.0, 0.0)) send_packet(sock, transform("00000010", 50000.0, 0.0)) correction = recv_until(sock, "positionCorrection") assert correction["x"] == 0.0 client = server._find_client_by_player_id(welcome["playerId"]) assert client is not None assert client.last_transform is not None assert client.last_transform["x"] == 0.0 sock.close() server.stop() def test_oversized_unterminated_packet_closes_session(): server = start_server() sock, _ = connect(server) sock.sendall(b"x" * (64 * 1024 + 1)) ended = recv_until(sock, "sessionEnded") assert ended["code"] == "packet_too_large" sock.close() server.stop()