yumoqing/world_sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(M11b-2d-A3): 移除一次性调试探针 .m11b2d_probe.py（核心检查已升级为 tests/test_m11b2d_fixture_contract.py 守卫用例）

Browse Source
This commit is contained in:
agent.develop 2026-09-22 21:39:50 +08:00
parent 2dc774a406
commit 81fdb86563
1 changed files with 0 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
.m11b2d_probe.py
View File

@ -1,66 +0,0 @@
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tests"))
import _m11b2_compat as H # noqa: E402
e = H.RuntimeEngine()
H.install(e)
try:
r1 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="unit.move",
payload={"dx": 1}, state_updates=[{"entity_id": "e1", "state": {"hp": 2}}],
client_event_id="ev-1", actor_id="u1")
print("R1", sorted(r1.keys()))
print("R1 vals", r1["ok"], r1["dedup"], r1["event_id"], r1["seq"], r1["state_rows"], r1["stmt_count"], r1["channel"])
print("states", r1["states"])
print("trace", H.trace_kinds(e))
print("counts", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE))
print("state row", H.read(tenant_id=7, session_id=11))
# dedup
r2 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="unit.move",
payload={"dx": 1}, state_updates=[{"entity_id": "e1", "state": {"hp": 9}}],
client_event_id="ev-1")
print("R2", r2["dedup"], r2["event_id"], r2["seq"], r2["state_rows"], sorted(r2.keys()))
print("counts after dedup", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE))
# conflict
try:
H.apply(tenant_id=7, world_id=1, session_id=11, event_type="x",
payload={}, state_updates=[{"entity_id": "e1", "state": {"hp": 0}}],
client_event_id="ev-3", base_version=999)
print("NO CONFLICT !!")
except H.RtxError as exc:
print("conflict code", exc.code, "trace", H.trace_kinds(e)[-3:])
print("counts after conflict", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE))
# snapshot
r3 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="snap",
payload={}, state_updates=[{"entity_id": "e1", "state": {"hp": 5}}],
client_event_id="ev-4", snapshot={"reason": "join", "hp": 5})
print("snapshot rows", e.count(H.SNAPSHOT_TABLE), "broadcasts", len(e.broadcasts))
print("broadcast payload keys", sorted(r3.keys()))
print("UPDATE stmts", H.update_statements(e))
print("errors:")
for kw in ({"tenant_id": None}, {"world_id": None}, {"session_id": None}, {"event_type": None}):
base = dict(tenant_id=7, world_id=1, session_id=11, event_type="x", payload={})
base.update(kw)
try:
H.apply(**base)
print(" no raise for", kw)
except H.RtxError as exc:
print(" ", kw, "->", exc.code)
# bad state_updates
for su in ([{"state": {"a": 1}}], [{"entity_id": "e1"}], "notalist"):
try:
H.apply(tenant_id=7, world_id=1, session_id=11, event_type="x", payload={},
state_updates=su, client_event_id="ev-b%d" % id(su) % 100)
print(" no raise for", su)
except H.RtxError as exc:
print(" su", su, "->", exc.code)
# missing tenant
try:
H.apply(tenant_id=None, world_id=1, session_id=11, event_type="x", payload={})
except H.RtxError as exc:
print("tenant->", exc.code)
print("final counts", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE), e.count(H.SNAPSHOT_TABLE))
print("trace tail", H.trace_kinds(e)[-6:])
finally:
H.uninstall()