108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
import sys
|
|
sys.path.insert(0, "/d/pipeline/workspaces/0/sdlc_general/modules/world_sync/tests")
|
|
import _m11b2_compat as c
|
|
|
|
_orig = c._default_clause
|
|
|
|
|
|
def patched(fd):
|
|
dv = fd.get("default")
|
|
if isinstance(dv, str) and dv.strip() == "":
|
|
return " DEFAULT ''"
|
|
return _orig(fd)
|
|
|
|
|
|
c._default_clause = patched
|
|
|
|
e = c.RuntimeEngine()
|
|
c.install(e)
|
|
|
|
|
|
def ts(i):
|
|
return "2026-06-01 00:%02d:%02d" % (i // 60, i % 60)
|
|
|
|
|
|
def show(tag, r):
|
|
print(tag, "keys", sorted(r.keys()))
|
|
print(tag, {k: r.get(k) for k in ("ok", "dedup", "event_id", "event_uid", "seq",
|
|
"state_rows", "stmt_count", "channel", "cursor",
|
|
"sla_breach", "states")})
|
|
|
|
|
|
r1 = c.apply(tenant_id=7, world_id=11, session_id=22, event_type="unit.move",
|
|
payload={"dx": 1}, state_updates=[{"entity_id": "e1", "state": {"hp": 2}}],
|
|
client_event_id="ev-1", actor_id="u1", occurred_at=ts(1))
|
|
show("R1", r1)
|
|
|
|
# dedup: same client_event_id
|
|
before = (e.count(c.EVENT_TABLE), e.count(c.STATE_TABLE))
|
|
r3 = c.apply(tenant_id=7, world_id=11, session_id=22, event_type="unit.move",
|
|
payload={"dx": 2}, state_updates=[{"entity_id": "e1", "state": {"hp": 9}}],
|
|
client_event_id="ev-1", actor_id="u2", occurred_at=ts(2))
|
|
show("R3dedup", r3)
|
|
print("counts", e.count(c.EVENT_TABLE), e.count(c.STATE_TABLE), before,
|
|
"bcast", len(e.broadcasts), "trace", c.trace_kinds(e))
|
|
|
|
# conflict
|
|
try:
|
|
c.apply(tenant_id=7, world_id=11, session_id=22, event_type="unit.move",
|
|
payload={"dx": 9}, state_updates=[{"entity_id": "e1", "state": {"hp": 99}}],
|
|
client_event_id="ev-4", occurred_at=ts(4), base_version=1)
|
|
print("CONFLICT NOT RAISED")
|
|
except Exception as exc:
|
|
print("CONFLICT", type(exc).__name__, getattr(exc, "code", None), str(exc)[:130])
|
|
print("trace tail", c.trace_kinds(e)[-3:], "counts", e.count(c.EVENT_TABLE), e.count(c.STATE_TABLE))
|
|
print("state now", e.query("SELECT state_version, state_json FROM pbl_entity_state"))
|
|
|
|
# snapshot
|
|
r5 = c.apply(tenant_id=7, world_id=11, session_id=22, event_type="snap", payload={},
|
|
state_updates=[], client_event_id="ev-5", occurred_at=ts(5),
|
|
snapshot={"reason": "reconnect", "entities": {"e1": {"hp": 5}}})
|
|
show("R5", r5)
|
|
print("snap", e.count(c.SNAPSHOT_TABLE))
|
|
|
|
# multi-state updates + seed_state
|
|
e.seed_state(7, 22, "e9", '{"hp":1}', state_version=3)
|
|
r6 = c.apply(tenant_id=7, world_id=11, session_id=22, event_type="multi", payload={},
|
|
state_updates=[{"entity_id": "e9", "state": {"hp": 7}},
|
|
{"entity_id": "new1", "state": {"x": 1}}],
|
|
client_event_id="ev-6", occurred_at=ts(6))
|
|
print("R6", r6["states"], r6["state_rows"], "updates", len(c.update_statements(e)))
|
|
print("read e9", c.read(tenant_id=7, session_id=22, entity_ids=["e9"]))
|
|
print("event rows", e.query("SELECT id, event_uid, seq, state_version, source, occurred_at FROM pbl_runtime_event ORDER BY id"))
|
|
|
|
# validation
|
|
for label, kw in (("tenant", dict(world_id=11, session_id=22, event_type="x")),
|
|
("param", dict(tenant_id=7, session_id=22, event_type="x")),
|
|
("etype", dict(tenant_id=7, world_id=11, session_id=22)),
|
|
("su", dict(tenant_id=7, world_id=11, session_id=22, event_type="x", state_updates="no")),
|
|
("eid", dict(tenant_id=7, world_id=11, session_id=22, event_type="x",
|
|
state_updates=[{"state": {"a": 1}}])),
|
|
("val", dict(tenant_id=7, world_id=11, session_id=22, event_type="x",
|
|
state_updates=[{"entity_id": "e1"}]))):
|
|
try:
|
|
c.apply(**kw)
|
|
print(label, "NOT RAISED")
|
|
except Exception as exc:
|
|
print(label, type(exc).__name__, getattr(exc, "code", None), str(exc)[:60])
|
|
print("trace after validation", c.trace_kinds(e)[-2:], "broadcasts", len(e.broadcasts))
|
|
|
|
# guards
|
|
for label, fn in (
|
|
("unknown", lambda: c.build_row(c.EVENT_TABLE, {"no_such_col": 1})),
|
|
("missing", lambda: c.build_row(c.STATE_TABLE, {"tenant_id": "t1"})),
|
|
("immutable", lambda: c.build_row(c.STATE_TABLE, {"state_json": "{}", "created_at": "x"},
|
|
strict=True, full_row=False)),
|
|
("append_evt", lambda: c.assert_append_only("UPDATE", c.EVENT_TABLE)),
|
|
("append_base", lambda: c.assert_append_only("DELETE", "world")),
|
|
("append_ok", lambda: c.assert_append_only("INSERT", c.EVENT_TABLE)),
|
|
):
|
|
try:
|
|
print(label, "OK", fn())
|
|
except Exception as exc:
|
|
print(label, "RAISED", getattr(exc, "code", None), str(exc)[:90])
|
|
c.uninstall()
|
|
|
|
import os
|
|
print("files", os.listdir("/tmp")[:3])
|