Commonwealth-Online/Commonwealth-Online-Server
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 2 Wiki Activity

Exercise transport-neutral GNS server path

Browse Source
This commit is contained in:
Nomads_Reach
2026-08-16 00:45:02 -04:00
parent 99b242eee6
commit 2f38c5b633
1 changed files with 16 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/tests/test_gns_gameplay_adapter.py
+16 -11
View File
@@ -9,8 +9,9 @@ from gns_gameplay_adapter import GnsConnectionAdapter, GnsGameplayAdapter
from gns_snapshot_envelope import decode_snapshot, encode_snapshot from gns_snapshot_envelope import decode_snapshot, encode_snapshot
from gns_transport import EventType, GnsEvent, RemoteEndpoint, SendResult from gns_transport import EventType, GnsEvent, RemoteEndpoint, SendResult
from packet_codec import encode_packet from packet_codec import encode_packet
from server_core import FalloutTogetherServer, PROTOCOL_VERSION from server_core import PROTOCOL_VERSION
from transport_policy import Delivery from transport_policy import Delivery
from transport_server import TransportAwareFalloutTogetherServer
class FakeGnsTransport: class FakeGnsTransport:
@@ -105,7 +106,7 @@ def transform_packet(x: float) -> dict:
def test_gns_adapter_reuses_v2_admission_and_server_owned_identity(): def test_gns_adapter_reuses_v2_admission_and_server_owned_identity():
server = FalloutTogetherServer(host="127.0.0.1", port=0) server = TransportAwareFalloutTogetherServer(host="127.0.0.1", port=0)
transport = FakeGnsTransport() transport = FakeGnsTransport()
adapter = GnsGameplayAdapter(server, transport) adapter = GnsGameplayAdapter(server, transport)
@@ -125,7 +126,7 @@ def test_gns_adapter_reuses_v2_admission_and_server_owned_identity():
def test_gns_adapter_preserves_reliable_and_snapshot_delivery_policy(): def test_gns_adapter_preserves_reliable_and_snapshot_delivery_policy():
server = FalloutTogetherServer(host="127.0.0.1", port=0) server = TransportAwareFalloutTogetherServer(host="127.0.0.1", port=0)
transport = FakeGnsTransport() transport = FakeGnsTransport()
adapter = GnsGameplayAdapter(server, transport) adapter = GnsGameplayAdapter(server, transport)
complete_handshake(adapter, transport, 201) complete_handshake(adapter, transport, 201)
@@ -168,7 +169,7 @@ def test_gns_adapter_preserves_reliable_and_snapshot_delivery_policy():
def test_gns_snapshot_reordering_and_duplicates_never_roll_state_backward(): def test_gns_snapshot_reordering_and_duplicates_never_roll_state_backward():
server = FalloutTogetherServer(host="127.0.0.1", port=0) server = TransportAwareFalloutTogetherServer(host="127.0.0.1", port=0)
transport = FakeGnsTransport() transport = FakeGnsTransport()
adapter = GnsGameplayAdapter(server, transport) adapter = GnsGameplayAdapter(server, transport)
complete_handshake(adapter, transport, 501) complete_handshake(adapter, transport, 501)
@@ -197,7 +198,7 @@ def test_gns_snapshot_reordering_and_duplicates_never_roll_state_backward():
def test_gns_raw_unsequenced_snapshot_is_rejected_before_state_mutation(): def test_gns_raw_unsequenced_snapshot_is_rejected_before_state_mutation():
server = FalloutTogetherServer(host="127.0.0.1", port=0) server = TransportAwareFalloutTogetherServer(host="127.0.0.1", port=0)
transport = FakeGnsTransport() transport = FakeGnsTransport()
adapter = GnsGameplayAdapter(server, transport) adapter = GnsGameplayAdapter(server, transport)
complete_handshake(adapter, transport, 601) complete_handshake(adapter, transport, 601)
@@ -217,26 +218,30 @@ def test_connection_adapter_may_drop_snapshots_but_never_reliable_messages_silen
connection = GnsConnectionAdapter(transport, 250) connection = GnsConnectionAdapter(transport, 250)
transport.send_results.append(SendResult.DROPPED) transport.send_results.append(SendResult.DROPPED)
connection.sendall(b'{"type":"transform","x":1}\n') connection.send_encoded(encode_packet({"type": "transform", "x": 1}))
transform_wire = transport.sent[-1][1] transform_wire = transport.sent[-1][1]
assert decode_snapshot(transform_wire.payload) is not None assert decode_snapshot(transform_wire.payload) is not None
transport.send_results.append(SendResult.BACKPRESSURE) transport.send_results.append(SendResult.BACKPRESSURE)
connection.sendall(b'{"type":"npcState","npcs":[]}\n') connection.send_encoded(encode_packet({"type": "npcState", "npcs": []}))
npc_wire = transport.sent[-1][1] npc_wire = transport.sent[-1][1]
assert decode_snapshot(npc_wire.payload) is not None assert decode_snapshot(npc_wire.payload) is not None
transport.send_results.append(SendResult.DROPPED) transport.send_results.append(SendResult.DROPPED)
with pytest.raises(OSError): with pytest.raises(OSError):
connection.sendall(b'{"type":"playerState","characterName":"Nomad"}\n') connection.send_encoded(encode_packet({"type": "playerState", "characterName": "Nomad"}))
transport.send_results.append(SendResult.BACKPRESSURE) transport.send_results.append(SendResult.BACKPRESSURE)
with pytest.raises(OSError): with pytest.raises(OSError):
connection.sendall(b'{"type":"combatHit","sequence":1}\n') connection.send_encoded(encode_packet({"type": "combatHit", "sequence": 1}))
def test_gns_adapter_enforces_existing_ban_policy(tmp_path): 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 = TransportAwareFalloutTogetherServer(
host="127.0.0.1",
port=0,
bans_path=str(tmp_path / "bans.json"),
)
server.ban_ip("127.0.0.1", "test ban") server.ban_ip("127.0.0.1", "test ban")
transport = FakeGnsTransport() transport = FakeGnsTransport()
adapter = GnsGameplayAdapter(server, transport) adapter = GnsGameplayAdapter(server, transport)
@@ -256,7 +261,7 @@ def test_gns_adapter_enforces_existing_ban_policy(tmp_path):
def test_gns_adapter_ends_session_on_native_oversize_event(): def test_gns_adapter_ends_session_on_native_oversize_event():
server = FalloutTogetherServer(host="127.0.0.1", port=0) server = TransportAwareFalloutTogetherServer(host="127.0.0.1", port=0)
transport = FakeGnsTransport() transport = FakeGnsTransport()
adapter = GnsGameplayAdapter(server, transport) adapter = GnsGameplayAdapter(server, transport)
complete_handshake(adapter, transport, 401) complete_handshake(adapter, transport, 401)