diff --git a/server/tests/test_gns_gameplay_adapter.py b/server/tests/test_gns_gameplay_adapter.py index 4dc672f..eaa9164 100644 --- a/server/tests/test_gns_gameplay_adapter.py +++ b/server/tests/test_gns_gameplay_adapter.py @@ -3,7 +3,9 @@ from __future__ import annotations import json from collections import deque -from gns_gameplay_adapter import GnsGameplayAdapter +import pytest + +from gns_gameplay_adapter import GnsConnectionAdapter, GnsGameplayAdapter from gns_transport import EventType, GnsEvent, RemoteEndpoint, SendResult from packet_codec import encode_packet from server_core import FalloutTogetherServer, PROTOCOL_VERSION @@ -15,6 +17,7 @@ class FakeGnsTransport: self.events = deque() self.endpoints = {} self.sent = [] + self.send_results = deque() self.disconnected = [] self.closed = False @@ -26,7 +29,7 @@ class FakeGnsTransport: def send_encoded(self, connection_id, encoded): self.sent.append((connection_id, encoded)) - return SendResult.SENT + return self.send_results.popleft() if self.send_results else SendResult.SENT def send_packet(self, connection_id, packet): return self.send_encoded(connection_id, encode_packet(packet)) @@ -165,6 +168,25 @@ def test_gns_adapter_preserves_reliable_and_snapshot_delivery_policy(): assert player_state_relays[-1].delivery is Delivery.RELIABLE_ORDERED +def test_connection_adapter_may_drop_snapshots_but_never_reliable_messages_silently(): + transport = FakeGnsTransport() + connection = GnsConnectionAdapter(transport, 250) + + transport.send_results.append(SendResult.DROPPED) + connection.sendall(b'{"type":"transform","x":1}\n') + + transport.send_results.append(SendResult.BACKPRESSURE) + connection.sendall(b'{"type":"npcState","npcs":[]}\n') + + transport.send_results.append(SendResult.DROPPED) + with pytest.raises(OSError): + connection.sendall(b'{"type":"playerState","characterName":"Nomad"}\n') + + transport.send_results.append(SendResult.BACKPRESSURE) + with pytest.raises(OSError): + connection.sendall(b'{"type":"combatHit","sequence":1}\n') + + 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")