564 lines
24 KiB
Python
564 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""M11b-2 单事务写入 —— 可执行单元测试（sqlite 文件库真实跑通，非 import 面检查）。
|
||
|
||
覆盖需求点与验收标准：
|
||
- 正常写入：事件 + 状态在同一事务落库，两表各 1 行（TestAtomicWriteSuccess）；
|
||
- 事件插入失败 → 状态更新一并回滚（TestRollbackOnEventFailure）；
|
||
- 状态更新失败/并发冲突 → 整体回滚，事件不留（TestRollbackOnStateFailure）；
|
||
- 并发：多事务写同一实体，不出现脏写/丢失更新（TestConcurrencyNoLostUpdate）；
|
||
- 租户/世界隔离：跨租户同名 entity_id 互不影响（TestTenantWorldIsolation）；
|
||
- P99 ≤ 200ms（TestLatencyP99，percentile 与 m11b-selftest-run.log 同口径）；
|
||
- conn_factory 三级优先级 + fail-closed + 禁读 env/*.json（TestConnFactoryResolution）；
|
||
- 接线面：导出一致性、禁项（无广播/轮询）扫描（TestPackageExports / TestNoForbiddenLogic）。
|
||
|
||
运行：cd apps/scense/pkgs/world_sync && python -m pytest tests/test_m11b2_wiring.py -v
|
||
"""
|
||
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import threading
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from world_sync import pbl_runtime_sql as ps # noqa: E402
|
||
from world_sync.pbl_runtime_errors import ( # noqa: E402
|
||
ConcurrentStateConflict,
|
||
ConnFactoryNotConfigured,
|
||
EventWriteError,
|
||
PblRuntimeError,
|
||
StateWriteError,
|
||
TxConfigError,
|
||
)
|
||
from world_sync.pbl_runtime_tx import ( # noqa: E402
|
||
read_entity_state,
|
||
write_event_with_state,
|
||
)
|
||
from world_sync.pbl_runtime_tx_env import ( # noqa: E402
|
||
build_sqlite_conn_factory,
|
||
resolve_conn_factory,
|
||
)
|
||
|
||
percentile = ps.percentile
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# fixtures / helpers
|
||
# --------------------------------------------------------------------------
|
||
|
||
@pytest.fixture
|
||
def db_path(tmp_path):
|
||
"""文件型 sqlite 库（多线程共享同一库，比 :memory: 更适合并发用例）。"""
|
||
return str(tmp_path / "m11b2.db")
|
||
|
||
|
||
@pytest.fixture
|
||
def factory(db_path):
|
||
"""sqlite 连接工厂，并保证两表 DDL 已建（CREATE TABLE IF NOT EXISTS 幂等）。"""
|
||
f = build_sqlite_conn_factory(db_path)
|
||
c = f()
|
||
for stmt in ps.ddl_for(ps.DIALECT_SQLITE):
|
||
c.execute(stmt)
|
||
c.commit()
|
||
c.close()
|
||
return f
|
||
|
||
|
||
@pytest.fixture
|
||
def conn(factory):
|
||
c = factory()
|
||
yield c
|
||
c.close()
|
||
|
||
|
||
def _event(tenant="t1", world="w1", entity="e1", **kw):
|
||
base = {
|
||
"tenant_id": tenant,
|
||
"world_id": world,
|
||
"entity_id": entity,
|
||
"event_type": "state.updated",
|
||
"payload": {"hp": 100},
|
||
"session_id": "s-1",
|
||
"source": "unit-test",
|
||
}
|
||
base.update(kw)
|
||
return base
|
||
|
||
|
||
def _state(tenant="t1", world="w1", entity="e1", **kw):
|
||
base = {"tenant_id": tenant, "world_id": world, "entity_id": entity,
|
||
"state": {"hp": 100, "x": 1}}
|
||
base.update(kw)
|
||
return base
|
||
|
||
|
||
def _counts(conn):
|
||
ev = conn.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0]
|
||
st = conn.execute("SELECT COUNT(*) FROM pbl_entity_state").fetchone()[0]
|
||
return ev, st
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 1. 正常写入：单事务内事件 + 状态各 1 行
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestAtomicWriteSuccess(object):
|
||
|
||
def test_first_write_inserts_state_and_event(self, conn, factory):
|
||
result = write_event_with_state(_event(), _state(state={"hp": 90}),
|
||
conn_factory=factory)
|
||
assert result["action"] == "inserted"
|
||
assert result["state_version"] == 1
|
||
assert result["dialect"] == "sqlite"
|
||
assert _counts(conn) == (1, 1)
|
||
|
||
row = conn.execute(
|
||
"SELECT event_id, entity_id, state_version, payload, created_at "
|
||
"FROM pbl_runtime_event"
|
||
).fetchone()
|
||
assert row[1] == "e1"
|
||
assert row[2] == 1
|
||
assert '"hp":100' in row[3] # payload 存事件体
|
||
assert row[4].endswith("Z")
|
||
st = conn.execute(
|
||
"SELECT state, state_version, updated_by_event FROM pbl_entity_state"
|
||
).fetchone()
|
||
assert '"hp":90' in st[0] # 状态存新值
|
||
assert st[1] == 1
|
||
assert st[2] == row[0] # 状态行回指生效事件
|
||
|
||
def test_second_write_bumps_version(self, conn, factory):
|
||
write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
|
||
r2 = write_event_with_state(_event(), _state(state={"hp": 70}), conn_factory=factory)
|
||
assert r2["action"] == "updated"
|
||
assert r2["state_version"] == 2
|
||
assert _counts(conn) == (2, 1)
|
||
|
||
def test_event_only_write_keeps_state(self, conn, factory):
|
||
write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
|
||
r = write_event_with_state(_event(event_type="chat"), None, conn_factory=factory)
|
||
assert r["action"] == "event_only"
|
||
assert _counts(conn) == (2, 1)
|
||
|
||
def test_read_entity_state_helper(self, factory):
|
||
write_event_with_state(_event(), _state(state={"hp": 42}), conn_factory=factory)
|
||
got = read_entity_state(conn_factory=factory, tenant_id="t1",
|
||
world_id="w1", entity_id="e1")
|
||
assert got["state_version"] == 1
|
||
assert '"hp":42' in got["state"]
|
||
assert read_entity_state(conn_factory=factory, tenant_id="t1",
|
||
world_id="w1", entity_id="nope") is None
|
||
|
||
def test_missing_tenant_rejected_before_touching_db(self, factory):
|
||
with pytest.raises(TxConfigError) as ei:
|
||
write_event_with_state({"world_id": "w1", "entity_id": "e1"},
|
||
_state(), conn_factory=factory)
|
||
assert ei.value.code == "PBL_TX_CONFIG_ERROR"
|
||
assert ei.value.detail.get("field") == "tenant_id"
|
||
assert _counts(factory()) == (0, 0)
|
||
|
||
def test_conflicting_scope_rejected(self, factory):
|
||
with pytest.raises(TxConfigError):
|
||
write_event_with_state(_event(tenant="t1"),
|
||
_state(tenant="t2"), conn_factory=factory)
|
||
|
||
def test_result_is_serializable(self, factory):
|
||
import json
|
||
r = write_event_with_state(_event(), _state(), conn_factory=factory)
|
||
json.dumps(r) # 不抛即通过
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 2. 失败整体回滚：事件插入失败 → 状态变更消失
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestRollbackOnEventFailure(object):
|
||
|
||
def test_event_failure_rolls_back_state_update(self, conn, factory):
|
||
"""状态先更新成功，事件插入被触发器打坏 → 断言两表零增量、状态仍是旧值。"""
|
||
write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
|
||
conn.execute(
|
||
"CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event "
|
||
"BEGIN SELECT RAISE(ABORT, 'boom: event insert failed'); END"
|
||
)
|
||
|
||
with pytest.raises(EventWriteError) as ei:
|
||
write_event_with_state(_event(), _state(state={"hp": 1}), conn_factory=factory)
|
||
assert ei.value.code == "PBL_RUNTIME_EVENT_WRITE_FAILED"
|
||
assert "boom" in str(ei.value)
|
||
|
||
assert _counts(conn) == (1, 1)
|
||
st = conn.execute("SELECT state, state_version FROM pbl_entity_state").fetchone()
|
||
assert '"hp":90' in st[0]
|
||
assert st[1] == 1
|
||
|
||
def test_duplicate_event_id_rolls_back_state(self, conn, factory):
|
||
"""同 event_id 主键冲突 → 事件失败 → 状态更新回滚。"""
|
||
write_event_with_state(_event(), _state(state={"hp": 90}),
|
||
conn_factory=factory, event_id="dup-1")
|
||
with pytest.raises(EventWriteError):
|
||
write_event_with_state(_event(), _state(state={"hp": 5}),
|
||
conn_factory=factory, event_id="dup-1")
|
||
assert _counts(conn) == (1, 1)
|
||
st = conn.execute("SELECT state_version FROM pbl_entity_state").fetchone()
|
||
assert st[0] == 1
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 3. 失败整体回滚：状态更新失败 / 并发冲突
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestRollbackOnStateFailure(object):
|
||
|
||
def test_expected_version_conflict_rolls_back_both(self, conn, factory):
|
||
write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
|
||
|
||
with pytest.raises(ConcurrentStateConflict) as ei:
|
||
write_event_with_state(
|
||
_event(), _state(state={"hp": 0}, expected_version=99),
|
||
conn_factory=factory)
|
||
assert ei.value.code == "PBL_ENTITY_STATE_CONFLICT"
|
||
assert ei.value.detail["expected_version"] == 99
|
||
assert ei.value.detail["actual_version"] == 1
|
||
assert _counts(conn) == (1, 1) # 冲突时事件一条都不能多留
|
||
|
||
def test_state_trigger_failure_rolls_back_event(self, conn, factory):
|
||
"""状态 UPDATE 失败（触发器报错）→ 本事务的事件也不能落库。"""
|
||
write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
|
||
conn.execute(
|
||
"CREATE TRIGGER fail_state BEFORE UPDATE ON pbl_entity_state "
|
||
"BEGIN SELECT RAISE(ABORT, 'boom: state update failed'); END"
|
||
)
|
||
with pytest.raises(StateWriteError):
|
||
write_event_with_state(_event(), _state(state={"hp": 3}), conn_factory=factory)
|
||
# 事件仍只有首条（失败事务的事件被回滚），状态版本仍为 1
|
||
assert _counts(conn) == (1, 1)
|
||
assert conn.execute(
|
||
"SELECT state_version FROM pbl_entity_state").fetchone()[0] == 1
|
||
|
||
def test_failed_write_leaves_no_open_transaction(self, factory):
|
||
"""失败后连接必须已回滚，后续写入不受影响（不吞错误、不留半截事务）。"""
|
||
c = factory()
|
||
c.execute(
|
||
"CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event "
|
||
"BEGIN SELECT RAISE(ABORT, 'boom'); END"
|
||
)
|
||
with pytest.raises(EventWriteError):
|
||
write_event_with_state(_event(), _state(), conn=c)
|
||
c.execute("DROP TRIGGER fail_event")
|
||
r = write_event_with_state(_event(), _state(), conn=c)
|
||
assert r["state_version"] == 1
|
||
assert c.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0] == 1
|
||
assert c.execute("SELECT COUNT(*) FROM pbl_entity_state").fetchone()[0] == 1
|
||
c.close()
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 4. 并发：不出现脏写 / 丢失更新
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestConcurrencyNoLostUpdate(object):
|
||
|
||
def test_concurrent_writers_no_lost_update(self, factory):
|
||
"""8 线程同时写同一实体：成功版本号连续无空洞，事件数 == 成功数。"""
|
||
threads = 8
|
||
results = []
|
||
errors = []
|
||
lock = threading.Lock()
|
||
barrier = threading.Barrier(threads)
|
||
|
||
def worker(i):
|
||
barrier.wait()
|
||
try:
|
||
r = write_event_with_state(
|
||
_event(payload={"writer": i}),
|
||
_state(state={"hp": 100 + i}),
|
||
conn_factory=factory,
|
||
)
|
||
with lock:
|
||
results.append(r["state_version"])
|
||
except PblRuntimeError as exc:
|
||
with lock:
|
||
errors.append(exc.code)
|
||
except sqlite3.OperationalError as exc:
|
||
# sqlite 文件库锁超时属环境限制，不算脏写
|
||
with lock:
|
||
errors.append("locked:%s" % str(exc)[:40])
|
||
|
||
ts = [threading.Thread(target=worker, args=(i,)) for i in range(threads)]
|
||
for t in ts:
|
||
t.start()
|
||
for t in ts:
|
||
t.join()
|
||
|
||
c = factory()
|
||
try:
|
||
n_events = c.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0]
|
||
st = c.execute("SELECT state_version FROM pbl_entity_state").fetchone()
|
||
finally:
|
||
c.close()
|
||
|
||
assert len(results) >= 1, "all writers failed: %s" % errors
|
||
assert n_events == len(results) # 没有「事件写了但状态没推进」的半截事务
|
||
assert st[0] == len(results) # 版本 == 成功提交数，无丢失更新
|
||
assert sorted(results) == list(range(1, len(results) + 1)) # 连续无空洞
|
||
assert not [e for e in errors if not e.startswith("locked")]
|
||
|
||
def test_optimistic_lock_rejects_stale_writer(self, conn, factory):
|
||
"""显式 expected_version 的写者：版本被别人推进后必须被拒绝（不覆盖）。"""
|
||
write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory)
|
||
write_event_with_state(_event(), _state(state={"hp": 80}), conn_factory=factory)
|
||
with pytest.raises(ConcurrentStateConflict):
|
||
write_event_with_state(_event(), _state(state={"hp": 70}, expected_version=1),
|
||
conn_factory=factory)
|
||
assert _counts(conn) == (2, 1)
|
||
st = conn.execute("SELECT state FROM pbl_entity_state").fetchone()
|
||
assert '"hp":80' in st[0] # 迟到的旧版本写者没有覆盖新状态
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 5. 租户 / 世界隔离
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestTenantWorldIsolation(object):
|
||
|
||
def test_same_entity_id_across_tenants_isolated(self, conn, factory):
|
||
write_event_with_state(_event(tenant="t1"), _state(tenant="t1", state={"hp": 1}),
|
||
conn_factory=factory)
|
||
write_event_with_state(_event(tenant="t2"), _state(tenant="t2", state={"hp": 2}),
|
||
conn_factory=factory)
|
||
assert _counts(conn) == (2, 2)
|
||
r = conn.execute(
|
||
"SELECT tenant_id, state_version FROM pbl_entity_state ORDER BY tenant_id"
|
||
).fetchall()
|
||
assert r == [("t1", 1), ("t2", 1)] # 各自独立版本，互不覆盖
|
||
|
||
def test_same_entity_id_across_worlds_isolated(self, conn, factory):
|
||
write_event_with_state(_event(world="w1"), _state(world="w1"), conn_factory=factory)
|
||
write_event_with_state(_event(world="w2"), _state(world="w2"), conn_factory=factory)
|
||
assert _counts(conn) == (2, 2)
|
||
|
||
def test_event_query_scoped_by_tenant(self, conn, factory):
|
||
write_event_with_state(_event(tenant="t1"), _state(tenant="t1"), conn_factory=factory)
|
||
write_event_with_state(_event(tenant="t2"), _state(tenant="t2"), conn_factory=factory)
|
||
n = conn.execute(
|
||
"SELECT COUNT(*) FROM pbl_runtime_event WHERE tenant_id = ? AND world_id = ?",
|
||
("t1", "w1")).fetchone()[0]
|
||
assert n == 1
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 6. 时延：P99 ≤ 200ms（不含广播）
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestLatencyP99(object):
|
||
|
||
def test_percentile_linear_interpolation(self):
|
||
assert percentile([], 99) == 0.0
|
||
assert percentile([5], 99) == 5.0
|
||
assert percentile([1, 2, 3, 4], 50) == 2.5
|
||
assert abs(percentile(list(range(1, 101)), 99) - 99.01) < 1e-9
|
||
assert percentile([10, 20, 30], 0) == 10
|
||
assert percentile([10, 20, 30], 100) == 30
|
||
|
||
def test_single_tx_p99_under_200ms(self, factory):
|
||
"""100 次单事务写入（事件+状态），P99 ≤ 200ms（sqlite 本地口径）。"""
|
||
samples = []
|
||
for i in range(100):
|
||
r = write_event_with_state(
|
||
_event(entity="perf-%d" % (i % 10), payload={"i": i}),
|
||
_state(entity="perf-%d" % (i % 10), state={"i": i}),
|
||
conn_factory=factory,
|
||
)
|
||
samples.append(r["elapsed_ms"])
|
||
p99 = percentile(samples, 99)
|
||
assert p99 <= 200.0, "P99=%.3fms exceeds 200ms budget" % p99
|
||
assert percentile(samples, 50) <= 200.0
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 7. conn_factory 三级优先级 + fail-closed + 禁读 env/*.json
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestConnFactoryResolution(object):
|
||
|
||
def test_explicit_factory_wins(self, tmp_path, monkeypatch):
|
||
path = str(tmp_path / "explicit.db")
|
||
called = []
|
||
real = build_sqlite_conn_factory(path)
|
||
|
||
def counting():
|
||
called.append(1)
|
||
return real()
|
||
|
||
monkeypatch.setenv("PBL_RUNTIME_DB_URL", "sqlite:///%s" % (tmp_path / "env.db"))
|
||
f = resolve_conn_factory(counting)
|
||
c = f()
|
||
c.execute("SELECT 1").fetchone()
|
||
c.close()
|
||
assert called == [1]
|
||
|
||
def test_env_url_used_when_no_injection(self, tmp_path, monkeypatch):
|
||
path = str(tmp_path / "from-env.db")
|
||
monkeypatch.setenv("PBL_RUNTIME_DB_URL", "sqlite:///%s" % path)
|
||
f = resolve_conn_factory(None)
|
||
c = f()
|
||
assert c is not None
|
||
c.close()
|
||
assert os.path.isfile(path)
|
||
|
||
def test_db_url_string_injection(self, tmp_path, monkeypatch):
|
||
path = str(tmp_path / "by-string.db")
|
||
f = resolve_conn_factory("sqlite:///%s" % path)
|
||
c = f()
|
||
c.close()
|
||
assert os.path.isfile(path)
|
||
|
||
def test_fail_closed_when_nothing_configured(self, monkeypatch):
|
||
import world_sync.pbl_runtime_tx_env as txenv
|
||
monkeypatch.delenv("PBL_RUNTIME_DB_URL", raising=False)
|
||
monkeypatch.delenv("WORLD_SYNC_DB_URL", raising=False)
|
||
# 屏蔽模块自有 conf/db.json，模拟「三级全空」
|
||
monkeypatch.setattr(txenv, "load_db_config", lambda conf_dir=None: None)
|
||
with pytest.raises(ConnFactoryNotConfigured) as ei:
|
||
resolve_conn_factory(None)
|
||
assert ei.value.code == "PBL_TX_CONN_FACTORY_MISSING"
|
||
assert "forbidden" in str(ei.value)
|
||
assert ei.value.detail["tried"] == [
|
||
"explicit", "server_env", "env_vars", "module_conf/db.json"]
|
||
|
||
def test_plaintext_password_in_module_conf_rejected(self, monkeypatch):
|
||
import world_sync.pbl_runtime_tx_env as txenv
|
||
monkeypatch.setattr(
|
||
txenv, "load_db_config",
|
||
lambda conf_dir=None: {"dialect": "mysql", "host": "h",
|
||
"user": "u", "database": "d",
|
||
"password": "s3cr3t"})
|
||
with pytest.raises(TxConfigError):
|
||
txenv._from_module_conf()
|
||
|
||
def test_unsupported_url_scheme_rejected(self):
|
||
import world_sync.pbl_runtime_tx_env as txenv
|
||
with pytest.raises(TxConfigError):
|
||
txenv._parse_db_url("oracle://x/y")
|
||
|
||
def test_no_env_test_json_candidate_in_source(self):
|
||
"""QC #5 回归守卫：源码里不得再出现 env/test.json 之类的凭据文件候选路径。"""
|
||
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
pkg = os.path.join(root, "world_sync")
|
||
banned = ('"env/test.json"', '"env/prod.json"', "'env/test.json'",
|
||
"'env/prod.json'", "projects/pbls/env")
|
||
for name in sorted(os.listdir(pkg)):
|
||
if not name.endswith(".py"):
|
||
continue
|
||
with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh:
|
||
text = fh.read()
|
||
for token in banned:
|
||
assert token not in text, "%s still references %s" % (name, token)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 8. 方言层单测
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestDialectLayer(object):
|
||
|
||
def test_detect_sqlite(self, conn):
|
||
assert ps.detect_dialect(conn) == ps.DIALECT_SQLITE
|
||
|
||
def test_placeholders_per_dialect(self):
|
||
assert ps.placeholder(ps.DIALECT_SQLITE, 0) == "?"
|
||
assert ps.placeholder(ps.DIALECT_MYSQL, 0) == "%s"
|
||
assert ps.placeholder(ps.DIALECT_POSTGRES, 2) == "$3"
|
||
assert ps.render_placeholders(ps.DIALECT_MYSQL, 3) == "%s, %s, %s"
|
||
|
||
def test_for_update_only_on_row_lock_dbs(self):
|
||
assert ps.for_update_clause(ps.DIALECT_SQLITE) == ""
|
||
assert ps.for_update_clause(ps.DIALECT_MYSQL).strip() == "FOR UPDATE"
|
||
assert ps.for_update_clause(ps.DIALECT_POSTGRES).strip() == "FOR UPDATE"
|
||
|
||
def test_read_committed_stmt(self):
|
||
assert "READ COMMITTED" in ps.read_committed_stmt(ps.DIALECT_MYSQL)
|
||
assert "READ COMMITTED" in ps.read_committed_stmt(ps.DIALECT_POSTGRES)
|
||
assert ps.read_committed_stmt(ps.DIALECT_SQLITE) is None
|
||
|
||
def test_begin_stmt(self):
|
||
assert ps.begin_stmt(ps.DIALECT_SQLITE) == "BEGIN IMMEDIATE"
|
||
assert ps.begin_stmt(ps.DIALECT_MYSQL) is None
|
||
|
||
def test_unknown_dialect_raises(self):
|
||
class Weird(object):
|
||
pass
|
||
with pytest.raises(ValueError):
|
||
ps.detect_dialect(Weird())
|
||
|
||
def test_ddl_covers_both_tables(self):
|
||
text = " ".join(ps.ddl_for(ps.DIALECT_SQLITE))
|
||
assert ps.EVENT_TABLE in text and ps.STATE_TABLE in text
|
||
assert "PRIMARY KEY (tenant_id, world_id, entity_id)" in text
|
||
assert "ENGINE=InnoDB" in " ".join(ps.ddl_for(ps.DIALECT_MYSQL))
|
||
|
||
def test_no_ddl_for_unknown_dialect(self):
|
||
with pytest.raises(ValueError):
|
||
ps.ddl_for("oracle")
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 9. 接线面 / 禁项扫描
|
||
# --------------------------------------------------------------------------
|
||
|
||
class TestPackageExports(object):
|
||
|
||
def test_all_is_importable(self):
|
||
import world_sync as ws
|
||
for name in ws.__all__:
|
||
assert hasattr(ws, name), "missing export: %s" % name
|
||
|
||
def test_public_api_surface(self):
|
||
import world_sync as ws
|
||
assert callable(ws.write_event_with_state)
|
||
assert callable(ws.read_entity_state)
|
||
assert "ConcurrentStateConflict" in ws.__all__
|
||
assert "EventWriteError" in ws.__all__
|
||
|
||
def test_sqlor_layer_delegates_to_tx_core(self):
|
||
from world_sync import pbl_runtime_tx, pbl_runtime_tx_sqlor
|
||
assert callable(pbl_runtime_tx_sqlor.write_event_with_state)
|
||
assert pbl_runtime_tx_sqlor._write_event_with_state is \
|
||
pbl_runtime_tx.write_event_with_state
|
||
|
||
|
||
class TestNoForbiddenLogic(object):
|
||
"""验收标准「不引入广播逻辑、不引入轮询逻辑」的机械守卫。"""
|
||
|
||
FORBIDDEN = ("broadcast", "publish_to", "subscribe", "poll(", "polling",
|
||
"while True:", "time.sleep(", "websocket", "event_source")
|
||
|
||
TARGETS = ("pbl_runtime_tx.py", "pbl_runtime_tx_sqlor.py",
|
||
"pbl_runtime_sql.py", "pbl_runtime_errors.py",
|
||
"pbl_runtime_tx_env.py")
|
||
|
||
def test_no_broadcast_or_polling_in_m11b2_modules(self):
|
||
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
pkg = os.path.join(root, "world_sync")
|
||
for name in self.TARGETS:
|
||
with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh:
|
||
lines = fh.readlines()
|
||
for idx, ln in enumerate(lines, 1):
|
||
low = ln.lower()
|
||
for token in self.FORBIDDEN:
|
||
if token in low and not low.lstrip().startswith("#"):
|
||
pytest.fail("%s:%d contains forbidden %r" % (name, idx, token))
|
||
|
||
def test_no_network_imports(self):
|
||
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
pkg = os.path.join(root, "world_sync")
|
||
for name in self.TARGETS:
|
||
with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh:
|
||
text = fh.read()
|
||
for mod in ("import asyncio", "import socket", "import requests",
|
||
"import aiohttp"):
|
||
assert mod not in text, "%s %s" % (name, mod)
|