428 lines
22 KiB
Python
428 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M1b 扩展字段 + 关联判定测试（真实执行，非 py_compile）。
|
||
|
||
覆盖 QC 退回意见 #6（7 类子对象泛化契约可验收证据）与
|
||
#5（关联表判定/字段映射证据）：
|
||
- 7 类子对象走同一组函数签名（泛化），逐类断言 CRUD + 扩展值读写
|
||
- 扩展字段校验（必填/枚举/数值区间/未声明字段）
|
||
- 关联判定 4 种状态：resolved / unresolved / stub / external
|
||
- Q-OPEN-3 守门：基表写 SQL 被 assert_base_table_immutable 拦截
|
||
- 租户隔离：跨租户读不到彼此的子对象与关联
|
||
|
||
运行：
|
||
python3 modules/pbl_blueprint/tests/test_m1b_ext_ref.py -v
|
||
"""
|
||
|
||
import os
|
||
import shutil
|
||
import sys
|
||
|
||
# --- M1b sys.path bootstrap: modules/ 下各包互为兄弟仓库，需逐个入 path ---
|
||
_M1B_MOD_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
||
_M1B_MODULES_DIR = os.path.abspath(os.path.join(_M1B_MOD_ROOT, ".."))
|
||
_M1B_CANDIDATES = [_M1B_MOD_ROOT, _M1B_MODULES_DIR]
|
||
try:
|
||
for _d in sorted(os.listdir(_M1B_MODULES_DIR)):
|
||
_sub = os.path.join(_M1B_MODULES_DIR, _d)
|
||
if os.path.isdir(_sub) and not _d.startswith("."):
|
||
_M1B_CANDIDATES.append(_sub)
|
||
except OSError:
|
||
pass
|
||
for _p in _M1B_CANDIDATES:
|
||
if _p not in sys.path:
|
||
sys.path.insert(0, _p)
|
||
# --- end bootstrap ---
|
||
|
||
import tempfile
|
||
import unittest
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
MOD_ROOT = os.path.dirname(HERE)
|
||
REPO_ROOT = os.path.abspath(os.path.join(MOD_ROOT, "..", ".."))
|
||
for p in (REPO_ROOT, MOD_ROOT):
|
||
if p not in sys.path:
|
||
sys.path.insert(0, p)
|
||
|
||
import pbl_blueprint.m1b.init as m1b_init # noqa: E402
|
||
from pbl_blueprint.m1b.dbutil import ( # noqa: E402
|
||
get_conn, reset_conn, sql_exec, sql_rows, table_exists,
|
||
)
|
||
from pbl_blueprint.m1b.errors import ( # noqa: E402
|
||
PblForbidden, PblNotFound, PblValidationError, TenantMissingError,
|
||
)
|
||
from pbl_blueprint.m1b.ref import ( # noqa: E402
|
||
assert_base_table_immutable, list_refs, ref_contract,
|
||
ref_summary, resolve_ref,
|
||
)
|
||
from pbl_blueprint.m1b.subobject import ( # noqa: E402
|
||
SUBOBJECT_SPECS, create_subobject, delete_subobject, get_ext,
|
||
get_subobject, list_subobjects, set_ext, subobject_contract,
|
||
subobject_tree, update_subobject, validate_ext,
|
||
)
|
||
from pbl_blueprint.m1b.tables import REF_KINDS, SUBOBJECT_TYPES # noqa: E402
|
||
|
||
TENANT = "T_M1B_EXTREF_001"
|
||
OTHER = "T_M1B_EXTREF_002"
|
||
BP_ID = "BP_M1B_EXTREF_0001"
|
||
|
||
#: 每类子对象的最小合法载荷（证明 7 类都能通过同一契约创建）
|
||
SAMPLES = {
|
||
"driving_question": {"question": "如何让校园垃圾分类真正落地？", "sort_no": 1},
|
||
"learning_goal": {"goal": "能设计并执行一次校园垃圾调研", "sort_no": 1},
|
||
"mission": {"name": "M1 现状调研", "sort_no": 1},
|
||
"problem": {"title": "分类标识不清导致误投", "sort_no": 1},
|
||
"role": {"name": "调研组长", "sort_no": 1},
|
||
"learner": {"name": "张三", "sort_no": 1},
|
||
"artifact_def": {"name": "调研报告", "sort_no": 1},
|
||
}
|
||
|
||
#: 每类子对象的扩展字段值（对应 EXT_FIELD_SEED 中声明的字段键）
|
||
EXT_SAMPLES = {
|
||
"driving_question": {"cognitive_level": "分析", "weight": 0.5},
|
||
"learning_goal": {"bloom_level": "L4", "assessable": True},
|
||
"mission": {"estimated_minutes": 60, "unlock_condition": "完成 M0"},
|
||
"problem": {"difficulty": "中", "hint_text": "先统计误投率"},
|
||
"role": {"team_size": 4, "responsibility": "统筹调研"},
|
||
"learner": {"grade": "高一", "group_no": 2},
|
||
"artifact_def": {"artifact_type": "文档", "rubric_ref": "rubric_report_v1"},
|
||
}
|
||
|
||
|
||
class M1bExtRefTestCase(unittest.TestCase):
|
||
"""扩展字段 + 关联判定真实库测试。"""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
cls.tmpdir = tempfile.mkdtemp(prefix="m1b_extref_")
|
||
cls.db_path = os.path.join(cls.tmpdir, "pbl_m1b_extref.sqlite3")
|
||
os.environ["PBL_M1B_DB"] = cls.db_path
|
||
os.environ["PBL_AUDIT_DIR"] = os.path.join(cls.tmpdir, "audit")
|
||
reset_conn()
|
||
cls.conn = get_conn(cls.db_path)
|
||
m1b_init.load_m1b(conn=cls.conn, actor_id="test_m1b_extref")
|
||
# 建 7 类子对象基表（最小列集），使泛化 CRUD 可真实落库
|
||
for st, spec in SUBOBJECT_SPECS.items():
|
||
# 列集去重：text_field 可能就叫 name（如 goal.name），
|
||
# 直接字符串拼接会产生 "duplicate column name: name"（sqlite 建表失败）
|
||
cols, seen = [], set()
|
||
for col in ("id TEXT PRIMARY KEY", "tenant_key TEXT", "tenant_id TEXT",
|
||
"blueprint_id TEXT", "%s TEXT" % spec["text_field"],
|
||
"sort_no INTEGER", "name TEXT", "status TEXT",
|
||
"created_at TEXT", "updated_at TEXT"):
|
||
key = col.split()[0].strip("`").lower()
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
cols.append(col)
|
||
sql_exec(
|
||
"CREATE TABLE IF NOT EXISTS %s (%s)" % (spec["table"], ", ".join(cols)),
|
||
conn=cls.conn)
|
||
sql_exec("CREATE TABLE IF NOT EXISTS pbl_blueprint ("
|
||
"id TEXT PRIMARY KEY, tenant_key TEXT, tenant_id TEXT, "
|
||
"name TEXT, code TEXT, status TEXT, template_id TEXT, "
|
||
"template_code TEXT, template_version TEXT, category TEXT, "
|
||
"created_at TEXT, updated_at TEXT)", conn=cls.conn)
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
reset_conn()
|
||
shutil.rmtree(cls.tmpdir, ignore_errors=True)
|
||
os.environ.pop("PBL_M1B_DB", None)
|
||
os.environ.pop("PBL_AUDIT_DIR", None)
|
||
|
||
# ---------- 1. 泛化契约（QC #6） ----------
|
||
def test_01_contract_covers_exactly_seven_types(self):
|
||
"""契约声明恰好 7 类，且每类都有表/主键/必填/文本字段声明。"""
|
||
c = subobject_contract()
|
||
self.assertEqual(c["type_count"], 7)
|
||
self.assertEqual(sorted(c["types"]), sorted(SUBOBJECT_TYPES))
|
||
self.assertEqual(len(SUBOBJECT_SPECS), 7)
|
||
for st, spec in SUBOBJECT_SPECS.items():
|
||
self.assertTrue(spec["table"].startswith("pbl_"))
|
||
self.assertEqual(spec["pk"], "id")
|
||
self.assertIn("blueprint_id", spec["required"])
|
||
self.assertTrue(spec["text_field"])
|
||
# 9 个统一方法签名对 7 类一致（泛化）
|
||
self.assertEqual(len(c["unified_methods"]), 9)
|
||
|
||
def test_02_unknown_subobject_type_rejected(self):
|
||
"""非法类型 fail-closed（不允许静默通过）。"""
|
||
for bad in ("pbl_project", "unknown", "", None):
|
||
with self.assertRaises(PblValidationError):
|
||
create_subobject(TENANT, bad, BP_ID, data={"name": "x"},
|
||
conn=self.conn)
|
||
|
||
def test_03_all_seven_types_crud_through_same_api(self):
|
||
"""7 类子对象全部通过同一组函数完成 增/查/改/删（泛化实证）。"""
|
||
created = {}
|
||
for st in SUBOBJECT_TYPES:
|
||
row = create_subobject(TENANT, st, BP_ID, data=dict(SAMPLES[st]),
|
||
actor_id="tester", conn=self.conn)
|
||
self.assertEqual(row["tenant_id"], TENANT)
|
||
self.assertEqual(row["blueprint_id"], BP_ID)
|
||
self.assertTrue(row["id"])
|
||
created[st] = row["id"]
|
||
# 读
|
||
got = get_subobject(TENANT, st, row["id"], conn=self.conn)
|
||
self.assertEqual(got["id"], row["id"])
|
||
listed = list_subobjects(TENANT, BP_ID, st, conn=self.conn)
|
||
self.assertGreaterEqual(len(listed), 1)
|
||
# 改
|
||
upd = update_subobject(TENANT, st, row["id"],
|
||
data={"sort_no": 9}, conn=self.conn)
|
||
self.assertEqual(int(upd["sort_no"]), 9)
|
||
self.assertEqual(len(created), 7)
|
||
# 删（并级联清理扩展值）
|
||
for st, oid in created.items():
|
||
n = delete_subobject(TENANT, st, oid, conn=self.conn)
|
||
self.assertEqual(int(n), 1)
|
||
with self.assertRaises(PblNotFound):
|
||
get_subobject(TENANT, st, oid, conn=self.conn)
|
||
|
||
def test_04_missing_required_field_rejected(self):
|
||
"""必填字段缺失 -> PblValidationError，且不落库。"""
|
||
before = sql_rows("SELECT COUNT(*) AS n FROM pbl_mission", conn=self.conn)
|
||
with self.assertRaises(PblValidationError):
|
||
create_subobject(TENANT, "mission", BP_ID, data={"sort_no": 1},
|
||
conn=self.conn)
|
||
after = sql_rows("SELECT COUNT(*) AS n FROM pbl_mission", conn=self.conn)
|
||
self.assertEqual(before[0]["n"], after[0]["n"])
|
||
|
||
def test_05_tenant_isolation_on_subobjects(self):
|
||
"""跨租户读不到彼此子对象（租户打头强制隔离）。"""
|
||
row = create_subobject(TENANT, "role", BP_ID, data=dict(SAMPLES["role"]),
|
||
conn=self.conn)
|
||
with self.assertRaises(PblNotFound):
|
||
get_subobject(OTHER, "role", row["id"], conn=self.conn)
|
||
self.assertEqual(list_subobjects(OTHER, BP_ID, "role", conn=self.conn), [])
|
||
with self.assertRaises(TenantMissingError):
|
||
list_subobjects(None, BP_ID, "role", conn=self.conn)
|
||
|
||
# ---------- 2. 扩展字段（QC #6 核心） ----------
|
||
def test_06_set_and_get_ext_for_all_seven_types(self):
|
||
"""7 类子对象都能写/读扩展字段值（泛化扩展接口实证）。"""
|
||
for st in SUBOBJECT_TYPES:
|
||
row = create_subobject(TENANT, st, BP_ID, data=dict(SAMPLES[st]),
|
||
conn=self.conn)
|
||
res = set_ext(TENANT, st, row["id"], EXT_SAMPLES[st],
|
||
source="manual", blueprint_id=BP_ID, conn=self.conn)
|
||
self.assertEqual(res["written"], len(EXT_SAMPLES[st]))
|
||
self.assertEqual(res["problems"], [])
|
||
got = get_ext(TENANT, st, row["id"], conn=self.conn)
|
||
for k, v in EXT_SAMPLES[st].items():
|
||
self.assertEqual(got.get(k), v, "%s.%s 扩展值不一致" % (st, k))
|
||
|
||
def test_07_set_ext_is_idempotent(self):
|
||
"""重复写同一扩展字段不产生重复行（唯一键幂等）。"""
|
||
row = create_subobject(TENANT, "problem", BP_ID,
|
||
data=dict(SAMPLES["problem"]), conn=self.conn)
|
||
set_ext(TENANT, "problem", row["id"], {"difficulty": "易"}, conn=self.conn)
|
||
set_ext(TENANT, "problem", row["id"], {"difficulty": "难"}, conn=self.conn)
|
||
rows = sql_rows(
|
||
"SELECT COUNT(*) AS n FROM pbl_subobject_ext WHERE subobject_id = ? "
|
||
"AND field_key = 'difficulty'", [row["id"]], conn=self.conn)
|
||
self.assertEqual(int(rows[0]["n"]), 1)
|
||
self.assertEqual(get_ext(TENANT, "problem", row["id"],
|
||
conn=self.conn)["difficulty"], "难")
|
||
|
||
def test_08_ext_validation_enum_and_range(self):
|
||
"""枚举越界 / 数值越界 / 类型错误 / 未声明字段 均被校验拦截。"""
|
||
# 枚举越界
|
||
p = validate_ext(TENANT, "driving_question", {"cognitive_level": "不存在级"},
|
||
conn=self.conn)
|
||
self.assertTrue(any(x["code"] == "ENUM" for x in p), p)
|
||
# 数值越界（weight max=1）
|
||
p = validate_ext(TENANT, "driving_question", {"weight": 5}, conn=self.conn)
|
||
self.assertTrue(any(x["code"] == "MAX" for x in p), p)
|
||
# 类型错误
|
||
p = validate_ext(TENANT, "mission", {"estimated_minutes": "很多"},
|
||
conn=self.conn)
|
||
self.assertTrue(any(x["code"] == "TYPE" for x in p), p)
|
||
# 未声明字段
|
||
p = validate_ext(TENANT, "role", {"not_declared_field": 1}, conn=self.conn)
|
||
self.assertTrue(any(x["code"] == "UNDECLARED" for x in p), p)
|
||
# 合法值通过
|
||
self.assertEqual(validate_ext(TENANT, "learner",
|
||
{"grade": "高二", "group_no": 3},
|
||
conn=self.conn), [])
|
||
|
||
def test_09_set_ext_rejects_invalid_values(self):
|
||
"""set_ext 校验失败即抛错且不落库。"""
|
||
row = create_subobject(TENANT, "artifact_def", BP_ID,
|
||
data=dict(SAMPLES["artifact_def"]), conn=self.conn)
|
||
before = sql_rows("SELECT COUNT(*) AS n FROM pbl_subobject_ext "
|
||
"WHERE subobject_id = ?", [row["id"]], conn=self.conn)
|
||
with self.assertRaises(PblValidationError):
|
||
set_ext(TENANT, "artifact_def", row["id"],
|
||
{"artifact_type": "不存在的类型"}, conn=self.conn)
|
||
after = sql_rows("SELECT COUNT(*) AS n FROM pbl_subobject_ext "
|
||
"WHERE subobject_id = ?", [row["id"]], conn=self.conn)
|
||
self.assertEqual(before[0]["n"], after[0]["n"])
|
||
|
||
def test_10_create_with_inline_ext(self):
|
||
"""创建子对象时可内联 ext，一次调用完成基础行+扩展值。"""
|
||
row = create_subobject(
|
||
TENANT, "learning_goal", BP_ID,
|
||
data=dict(SAMPLES["learning_goal"],
|
||
ext={"bloom_level": "L5", "assessable": True}),
|
||
conn=self.conn)
|
||
self.assertEqual(row["ext"]["bloom_level"], "L5")
|
||
self.assertEqual(get_ext(TENANT, "learning_goal", row["id"],
|
||
conn=self.conn)["assessable"], True)
|
||
|
||
def test_11_delete_cascades_ext_values(self):
|
||
"""删除子对象级联清理扩展值（应用层级联，无外键）。"""
|
||
row = create_subobject(TENANT, "mission", BP_ID,
|
||
data=dict(SAMPLES["mission"]), conn=self.conn)
|
||
set_ext(TENANT, "mission", row["id"], {"estimated_minutes": 30},
|
||
conn=self.conn)
|
||
self.assertGreater(len(get_ext(TENANT, "mission", row["id"],
|
||
conn=self.conn)), 0)
|
||
delete_subobject(TENANT, "mission", row["id"], conn=self.conn)
|
||
left = sql_rows("SELECT COUNT(*) AS n FROM pbl_subobject_ext "
|
||
"WHERE subobject_id = ?", [row["id"]], conn=self.conn)
|
||
self.assertEqual(int(left[0]["n"]), 0)
|
||
|
||
def test_12_subobject_tree_aggregates_seven_types(self):
|
||
"""subobject_tree 一次取全 7 类 + 扩展值。"""
|
||
bp = "BP_TREE_%s" % os.getpid()
|
||
for st in SUBOBJECT_TYPES:
|
||
create_subobject(TENANT, st, bp, data=dict(SAMPLES[st]),
|
||
conn=self.conn)
|
||
oid = list_subobjects(TENANT, bp, st, conn=self.conn)[0]["id"]
|
||
set_ext(TENANT, st, oid, EXT_SAMPLES[st], conn=self.conn,
|
||
blueprint_id=bp)
|
||
tree = subobject_tree(TENANT, bp, conn=self.conn)
|
||
self.assertEqual(sorted(tree["subobjects"].keys()),
|
||
sorted(SUBOBJECT_TYPES))
|
||
self.assertEqual(tree["counts"]["__total__"], 7)
|
||
for st in SUBOBJECT_TYPES:
|
||
self.assertEqual(len(tree["subobjects"][st]), 1)
|
||
self.assertTrue(tree["subobjects"][st][0]["ext"])
|
||
|
||
# ---------- 3. 关联表判定（QC #5 + Q-OPEN-3） ----------
|
||
def test_13_ref_contract_declares_readonly_and_no_fk(self):
|
||
"""关联契约声明：4 域只读、无外键、基表零写入。"""
|
||
c = ref_contract()
|
||
self.assertEqual(sorted(c["ref_kinds"]), sorted(REF_KINDS))
|
||
self.assertEqual(c["ownership"], "read_only")
|
||
self.assertTrue(c["q_open_3"]["no_foreign_key"])
|
||
self.assertEqual(c["q_open_3"]["base_table_writes"], 0)
|
||
self.assertEqual(sorted(c["base_tables_readonly"].keys()),
|
||
["entity", "scene", "script", "world"])
|
||
|
||
def test_14_ref_stub_when_base_table_unavailable(self):
|
||
"""基表不可达 -> resolve_status='stub'，不报错、不建表（Q-OPEN-3）。"""
|
||
row = resolve_ref(TENANT, BP_ID, "world", "W_NOT_EXIST_1",
|
||
actor_id="tester", conn=self.conn)
|
||
self.assertEqual(row["resolve_status"], "stub")
|
||
self.assertEqual(row["ownership"], "read_only")
|
||
self.assertIn("not available", row["resolve_note"])
|
||
self.assertFalse(table_exists("world", conn=self.conn),
|
||
"判定过程不得创建 world 基表")
|
||
|
||
def test_15_ref_resolved_with_readonly_snapshot(self):
|
||
"""基表可达且行存在 -> resolved + 只读快照。"""
|
||
sql_exec("CREATE TABLE IF NOT EXISTS world (id TEXT PRIMARY KEY, "
|
||
"name TEXT, code TEXT, status TEXT)", conn=self.conn)
|
||
sql_exec("INSERT INTO world (id, name, code, status) VALUES (?,?,?,?)",
|
||
["W_M1B_1", "M1b 测试世界", "world_m1b_1", "published"],
|
||
conn=self.conn)
|
||
row = resolve_ref(TENANT, BP_ID, "world", "W_M1B_1", conn=self.conn)
|
||
self.assertEqual(row["resolve_status"], "resolved")
|
||
self.assertEqual(row["ref_table"], "world")
|
||
self.assertEqual(row["ref_code"], "world_m1b_1")
|
||
self.assertTrue(row["ref_snapshot_json"])
|
||
# 快照是只读副本：改基表不影响已落快照
|
||
sql_exec("UPDATE world SET name = '改名后' WHERE id = 'W_M1B_1'",
|
||
conn=self.conn)
|
||
stored = [r for r in list_refs(TENANT, blueprint_id=BP_ID, conn=self.conn)
|
||
if r["ref_id"] == "W_M1B_1"][0]
|
||
self.assertEqual(stored["ref_snapshot"]["name"], "M1b 测试世界")
|
||
|
||
def test_16_ref_unresolved_when_row_missing(self):
|
||
"""基表可达但行不存在 -> unresolved。"""
|
||
row = resolve_ref(TENANT, BP_ID, "world", "W_MISSING_ROW", conn=self.conn)
|
||
self.assertEqual(row["resolve_status"], "unresolved")
|
||
self.assertIn("not found", row["resolve_note"])
|
||
|
||
def test_17_ref_external_is_stub(self):
|
||
"""external 域恒为 stub（M1b 不解析外部系统）。"""
|
||
row = resolve_ref(TENANT, BP_ID, "external", "EXT_1", conn=self.conn)
|
||
self.assertEqual(row["resolve_status"], "stub")
|
||
self.assertEqual(row["ref_table"], "")
|
||
|
||
def test_18_ref_invalid_kind_rejected(self):
|
||
"""非法 ref_kind / 缺 ref_id -> PblValidationError。"""
|
||
with self.assertRaises(PblValidationError):
|
||
resolve_ref(TENANT, BP_ID, "not_a_kind", "X1", conn=self.conn)
|
||
with self.assertRaises(PblValidationError):
|
||
resolve_ref(TENANT, BP_ID, "world", "", conn=self.conn)
|
||
with self.assertRaises(TenantMissingError):
|
||
resolve_ref(None, BP_ID, "world", "X1", conn=self.conn)
|
||
|
||
def test_19_ref_idempotent_and_subobject_scoped(self):
|
||
"""同目标+同发起方重复判定不增行；子对象级关联可区分。"""
|
||
r1 = resolve_ref(TENANT, BP_ID, "scene", "S_M1B_1", conn=self.conn)
|
||
r2 = resolve_ref(TENANT, BP_ID, "scene", "S_M1B_1", conn=self.conn)
|
||
self.assertEqual(r1["id"], r2["id"])
|
||
n = sql_rows("SELECT COUNT(*) AS c FROM pbl_blueprint_ref "
|
||
"WHERE ref_id = 'S_M1B_1' AND subobject_type = 'blueprint'",
|
||
conn=self.conn)
|
||
self.assertEqual(int(n[0]["c"]), 1)
|
||
row = create_subobject(TENANT, "mission", BP_ID,
|
||
data=dict(SAMPLES["mission"]), conn=self.conn)
|
||
resolve_ref(TENANT, BP_ID, "scene", "S_M1B_1", subobject_type="mission",
|
||
subobject_id=row["id"], conn=self.conn)
|
||
n2 = sql_rows("SELECT COUNT(*) AS c FROM pbl_blueprint_ref "
|
||
"WHERE ref_id = 'S_M1B_1'", conn=self.conn)
|
||
self.assertEqual(int(n2[0]["c"]), 2, "子对象级关联应独立成行")
|
||
|
||
def test_20_ref_summary_and_tenant_isolation(self):
|
||
"""汇总报表按 kind×status 计数；跨租户不可见。"""
|
||
resolve_ref(OTHER, BP_ID, "entity", "E_OTHER_1", conn=self.conn)
|
||
s = ref_summary(TENANT, blueprint_id=BP_ID, conn=self.conn)
|
||
self.assertEqual(s["ownership"], "read_only")
|
||
self.assertEqual(s["base_tables_modified"], 0)
|
||
self.assertGreaterEqual(s["total"], 3)
|
||
self.assertIn("world", s["by_kind"])
|
||
self.assertEqual(s["by_status"].get("resolved", 0), 1)
|
||
rows = list_refs(OTHER, blueprint_id=BP_ID, conn=self.conn)
|
||
self.assertTrue(all(r["tenant_id"] == OTHER for r in rows))
|
||
self.assertFalse(any(r["ref_id"] == "E_OTHER_1"
|
||
for r in list_refs(TENANT, conn=self.conn)))
|
||
|
||
def test_21_base_table_write_sql_is_blocked(self):
|
||
"""Q-OPEN-3 守门：对基表的写 SQL 被静态拦截。"""
|
||
bad = [
|
||
"UPDATE world SET name = 'x' WHERE id = '1'",
|
||
"DELETE FROM scene WHERE id = '1'",
|
||
"INSERT INTO entity (id) VALUES ('1')",
|
||
"ALTER TABLE script ADD COLUMN tenant_id VARCHAR(64)",
|
||
]
|
||
for sql in bad:
|
||
with self.assertRaises(PblForbidden):
|
||
assert_base_table_immutable(sql)
|
||
# 只读与自有表写入放行
|
||
self.assertTrue(assert_base_table_immutable(
|
||
"SELECT * FROM world WHERE id = '1'"))
|
||
self.assertTrue(assert_base_table_immutable(
|
||
"INSERT INTO pbl_blueprint_ref (id) VALUES ('1')"))
|
||
|
||
def test_22_generated_ddl_contains_no_fk_and_no_base_table(self):
|
||
"""生成的 sqlite/mariadb DDL 均无 FOREIGN KEY、不含基表建表语句。"""
|
||
ddl = m1b_init.SQLITE_DDL.upper()
|
||
self.assertNotIn("FOREIGN KEY", ddl)
|
||
for base in ("CREATE TABLE IF NOT EXISTS WORLD",
|
||
"CREATE TABLE IF NOT EXISTS SCENE",
|
||
"CREATE TABLE IF NOT EXISTS ENTITY",
|
||
"CREATE TABLE IF NOT EXISTS SCRIPT"):
|
||
self.assertNotIn(base, ddl)
|
||
for t in TABLES:
|
||
self.assertIn("CREATE TABLE IF NOT EXISTS %s" % t["name"].upper(), ddl)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|