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: the daemon refuses a snapshot document limit of zero

Browse source 
The daemon test ran with --max-snapshot-documents 0 to make every
snapshot a peer error. A limit of zero is now refused at startup, as the
Rust document server refuses it, so the test pins what the daemon does
with it: it exits non-zero naming the rule and never writes its ready
file. The limit's enforcement as a peer error stays pinned in process by
test_cultmesh_local_server_rejects_oversized_snapshot_responses.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 17:48:40 +02:00
commit 835bee3ae0
1 changed files with 11 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -3720,7 +3720,11 @@ class CultMeshTests(unittest.TestCase):
finally:
self._terminate_process(process)
def test_cultmesh_daemon_enforces_snapshot_document_limit_as_peer_error(self) -> None:
def test_cultmesh_daemon_refuses_a_snapshot_document_limit_of_zero_at_startup(self) -> None:
# A limit of zero is a server that could answer no snapshot, which the Rust document
# server's options also refuse; the daemon passes the flag to the server, and exits
# naming the rule instead of serving. The limit's enforcement as a peer error is pinned
# in process by test_cultmesh_local_server_rejects_oversized_snapshot_responses.
with tempfile.TemporaryDirectory() as temp:
ready_path = Path(temp) / "ready.json"
package_src = Path(__file__).resolve().parents[1] / "src"
@ -3731,7 +3735,7 @@ class CultMeshTests(unittest.TestCase):
if not existing_pythonpath
else f"{package_src}{os.pathsep}{existing_pythonpath}"
)
process = subprocess.Popen(
completed = subprocess.run(
[
sys.executable,
"-m",
@ -3746,25 +3750,14 @@ class CultMeshTests(unittest.TestCase):
"--ready-file",
str(ready_path),
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
capture_output=True,
text=True,
env=env,
timeout=30,
)
try:
ready = self._wait_for_ready_file(process, ready_path)
self.assertEqual(ready["snapshotLimits"], {"maxSnapshotBytes": None, "maxSnapshotDocuments": 0})
client = CultMesh.create_client("127.0.0.1", int(ready["port"]), timeout_seconds=2.0)
with self.assertRaisesRegex(CultNetPeerError, "Snapshot document limit exceeded") as raised:
client.fetch_snapshot_response(schema_ids=["cultcache.interop-note"])
self.assertEqual(raised.exception.response["schemaVersion"], "cultnet.error.v0")
self.assertEqual(raised.exception.response["code"], "snapshot_document_limit_exceeded")
self.assertEqual(raised.exception.response["details"]["documentCount"], 1)
self.assertEqual(raised.exception.response["details"]["maxSnapshotDocuments"], 0)
finally:
self._terminate_process(process)
self.assertNotEqual(completed.returncode, 0)
self.assertIn("max_snapshot_documents must be greater than zero", completed.stderr)
self.assertFalse(ready_path.exists())
def test_cultmesh_daemon_serves_opt_in_simulation_observations(self) -> None:
with tempfile.TemporaryDirectory() as temp: