GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultmesh-py: keep the generation test's replies whole; probe the small queue with an empty snapshot

Browse source 
The generation test reads each server packet as one message, so its
server now fragments only above 60000 bytes. The queue test's liveness
probe is an empty snapshot: a hello reply needs more fragments than its
eight-packet queue holds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:51:06 +02:00
commit 3cc081db0f
2 changed files with 7 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

7
packages/cultmesh-py/tests/test_cultmesh.py
View file

@ -4495,7 +4495,10 @@ class CultMeshTests(unittest.TestCase):
finally:
transport.close()
failures_after = server.rudp_send_failures
hello_reply, _ = self._rudp_exchange(server, hello(runtime_id="after-queue").to_wire())
# An empty snapshot fits the small queue; a hello reply would not.
after_reply, _ = self._rudp_exchange(
server, snapshot_request(message_id="after-queue", record_keys=["note:none"]).to_wire()
)
alive = server._rudp_thread is not None and server._rudp_thread.is_alive()
finally:
server.stop()
@ -4518,7 +4521,7 @@ class CultMeshTests(unittest.TestCase):
self.assertEqual(len(served_payload), 800)
self.assertEqual(failures_before, 0)
self.assertEqual(failures_after, 1)
self.assertEqual(hello_reply["schemaVersion"], "cultnet.hello.v0")
self.assertEqual(after_reply["messageId"], "after-queue")
self.assertTrue(alive)
def test_cultmesh_local_server_refuses_a_rudp_put_whose_reply_needs_more_than_65535_fragments(self) -> None:

3
packages/cultmesh-py/tests/test_rudp_server_generation.py
View file

@ -48,7 +48,8 @@ class RudpServerGenerationTests(unittest.TestCase):
def setUp(self) -> None:
node = create_node(runtime_id="rudp-generation")
node.database.register_document(NOTE)
self.server = CultMeshLocalServer(node=node, rudp_connection_id=CONNECTION_ID)
# Replies stay whole datagrams here, so each server packet the pump reads is one message.
self.server = CultMeshLocalServer(node=node, rudp_connection_id=CONNECTION_ID, rudp_max_fragment_bytes=60_000)
self.server.start()
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.bind(("127.0.0.1", 0))