Test Python GNS bridge wrapper

server/tests/test_gns_transport.py

@@ -0,0 +1,137 @@
+from __future__ import annotations
+
+import ctypes
+from collections import deque
+
+import pytest
+
+from gns_transport import EventType, GnsServerTransport, GnsTransportError, SendResult
+
+
+class FakeFunction:
+    def __init__(self, callback):
+        self.callback = callback
+        self.argtypes = None
+        self.restype = None
+
+    def __call__(self, *args):
+        return self.callback(*args)
+
+
+class FakeNativeBridge:
+    def __init__(self):
+        self.events = deque()
+        self.sent = []
+        self.disconnects = []
+        self.destroyed = False
+        self.co_gns_server_create = FakeFunction(self._create)
+        self.co_gns_server_destroy = FakeFunction(self._destroy)
+        self.co_gns_server_local_port = FakeFunction(lambda handle: 7777)
+        self.co_gns_server_connection_count = FakeFunction(lambda handle: 2)
+        self.co_gns_server_poll = FakeFunction(self._poll)
+        self.co_gns_server_send = FakeFunction(self._send)
+        self.co_gns_server_disconnect = FakeFunction(self._disconnect)
+
+    def _create(self, host, port, out_handle, error_buffer, error_size):
+        assert host == b"127.0.0.1"
+        assert port == 0
+        ctypes.cast(out_handle, ctypes.POINTER(ctypes.c_void_p))[0] = ctypes.c_void_p(0x1234)
+        error_buffer.value = b""
+        return 1
+
+    def _destroy(self, handle):
+        self.destroyed = True
+
+    def _poll(self, handle, event_pointer, payload_buffer, payload_capacity):
+        if not self.events:
+            return 0
+        event_type, connection_id, payload, reason, debug = self.events.popleft()
+        event = event_pointer._obj
+        event.type = event_type
+        event.connection_id = connection_id
+        event.reason = reason
+        event.payload_size = len(payload)
+        event.debug = debug.encode()
+        if event_type == EventType.MESSAGE and payload:
+            assert len(payload) <= payload_capacity
+            ctypes.memmove(payload_buffer, payload, len(payload))
+        return 1
+
+    def _send(self, handle, connection_id, payload_buffer, payload_size, delivery):
+        payload = ctypes.string_at(payload_buffer, payload_size)
+        self.sent.append((connection_id, payload, delivery))
+        return SendResult.SENT
+
+    def _disconnect(self, handle, connection_id, reason, debug):
+        self.disconnects.append((connection_id, reason, debug))
+        return 1
+
+
+def test_wrapper_polls_message_events_without_line_framing():
+    native = FakeNativeBridge()
+    native.events.append((EventType.CONNECTED, 42, b"", 0, ""))
+    native.events.append((EventType.MESSAGE, 42, b'{"type":"transform","x":1}', 0, ""))
+
+    transport = GnsServerTransport("127.0.0.1", 0, native_library=native)
+    assert transport.local_port == 7777
+    assert transport.connection_count == 2
+
+    connected = transport.poll()
+    assert connected is not None
+    assert connected.type is EventType.CONNECTED
+    assert connected.connection_id == 42
+
+    message = transport.poll()
+    assert message is not None
+    assert message.type is EventType.MESSAGE
+    assert message.payload == b'{"type":"transform","x":1}'
+    assert not message.payload.endswith(b"\n")
+    assert transport.poll() is None
+
+
+def test_wrapper_maps_protocol_delivery_to_native_send_modes():
+    native = FakeNativeBridge()
+    transport = GnsServerTransport("127.0.0.1", 0, native_library=native)
+
+    assert transport.send_packet(42, {"type": "transform", "x": 1}) is SendResult.SENT
+    assert transport.send_packet(42, {"type": "playerState", "characterName": "Nomad"}) is SendResult.SENT
+
+    assert native.sent[0][0] == 42
+    assert native.sent[0][2] == 0
+    assert native.sent[0][1] == b'{"type":"transform","x":1}'
+    assert native.sent[1][2] == 1
+    assert native.sent[1][1] == b'{"type":"playerState","characterName":"Nomad"}'
+
+
+def test_wrapper_preserves_disconnect_and_oversize_events():
+    native = FakeNativeBridge()
+    native.events.append((EventType.DISCONNECTED, 9, b"", 5003, "peer timeout"))
+    native.events.append((EventType.OVERSIZE_MESSAGE, 10, b"x" * (64 * 1024 + 1), 0, "too large"))
+    transport = GnsServerTransport("127.0.0.1", 0, native_library=native)
+
+    disconnected = transport.poll()
+    assert disconnected is not None
+    assert disconnected.type is EventType.DISCONNECTED
+    assert disconnected.connection_id == 9
+    assert disconnected.reason == 5003
+    assert disconnected.debug == "peer timeout"
+
+    oversize = transport.poll()
+    assert oversize is not None
+    assert oversize.type is EventType.OVERSIZE_MESSAGE
+    assert oversize.connection_id == 10
+    assert oversize.payload == b""
+
+
+def test_wrapper_disconnect_and_close_are_idempotent():
+    native = FakeNativeBridge()
+    transport = GnsServerTransport("127.0.0.1", 0, native_library=native)
+    assert transport.disconnect(77, reason=1000, debug="test")
+    assert native.disconnects == [(77, 1000, b"test")]
+
+    transport.close()
+    transport.close()
+    assert native.destroyed
+    assert transport.is_closed
+    with pytest.raises(GnsTransportError):
+        transport.poll()