deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
f844eca914
commit
c67b0a9e29
5
.gitignore
vendored
5
.gitignore
vendored
@ -10,3 +10,8 @@ dist/
|
||||
build/
|
||||
.venv/
|
||||
venv/
|
||||
_patch*.py
|
||||
ut.txt
|
||||
*.orig
|
||||
*.rej
|
||||
tests/__pycache__/
|
||||
|
||||
179
_patch2.py
179
_patch2.py
@ -1,179 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""一次性修补：fake_db 真实 2 参签名 / api 契约修正 / 陈旧测试列名对齐。"""
|
||||
import io
|
||||
import os
|
||||
|
||||
ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
|
||||
def patch(path, pairs, required=True):
|
||||
p = os.path.join(ROOT, path)
|
||||
src = io.open(p, encoding="utf-8").read()
|
||||
for old, new in pairs:
|
||||
if old not in src:
|
||||
if required:
|
||||
raise SystemExit("NOT FOUND in %s:\n%s" % (path, old[:200]))
|
||||
continue
|
||||
src = src.replace(old, new, 1)
|
||||
io.open(p, "w", encoding="utf-8").write(src)
|
||||
print("patched", path)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- fake_db
|
||||
# QC#3：fake_db 的 U 曾是 3 参 (tbl,row,where)，掩盖了真实 sqlor 2 参签名。
|
||||
# 现改为真实签名 U(tbl, ns)：ns 同时含 SET 值与 WHERE 条件（条件键 = id/tenant_id）。
|
||||
patch("tests/fake_db.py", [
|
||||
(""" def U(self, tbl, row, where):
|
||||
self._guard_readonly(tbl, "U")
|
||||
hits = 0
|
||||
for exist in self.tables.get(tbl, []):
|
||||
if self._match(exist, where):
|
||||
exist.update(row)
|
||||
hits += 1
|
||||
self.sql_log.append(("U", tbl, dict(row), dict(where)))
|
||||
return hits""",
|
||||
""" #: 真实 sqlor U(table, ns) 只收 2 参——ns 同时携带 SET 值与 WHERE 条件。
|
||||
#: 条件键约定为主键/租户键（本模块 update_ref 的 row 从不含这两列）。
|
||||
WHERE_KEYS = ("id", "tenant_id")
|
||||
|
||||
def U(self, tbl, ns):
|
||||
\"\"\"真实 sqlor 签名：U(table, ns)，2 参（crud-spec Pitfall 22）。\"\"\"
|
||||
self._guard_readonly(tbl, "U")
|
||||
ns = dict(ns or {})
|
||||
where = {k: ns.pop(k) for k in self.WHERE_KEYS if k in ns}
|
||||
if not where:
|
||||
raise AssertionError(
|
||||
"sqlor U(table, ns) 的 ns 必须含 WHERE 条件键 %s，实得 %s"
|
||||
% (list(self.WHERE_KEYS), sorted(ns)))
|
||||
hits = 0
|
||||
for exist in self.tables.get(tbl, []):
|
||||
if self._match(exist, where):
|
||||
exist.update(ns)
|
||||
hits += 1
|
||||
self.sql_log.append(("U", tbl, dict(ns), where))
|
||||
return hits"""),
|
||||
])
|
||||
|
||||
# ---------------------------------------------------------------- api.py
|
||||
# ① world_get_context：基表未命中/跨租户必须报错，不得静默返回空上下文
|
||||
patch("pbl_domain_ext/api.py", [
|
||||
(""" base = db.read_base_row("world", world_id, tenant_id)
|
||||
world_view = _base_view(base or {})
|
||||
ref = db.select_ref_by_key(tenant_id, "world", world_id)""",
|
||||
""" base = db.read_base_row("world", world_id, tenant_id)
|
||||
world_view = _base_view(base or {})
|
||||
if not world_view:
|
||||
# 基表未命中或 world 属其它租户：一律按不存在处理（跨租户不得静默成功）
|
||||
raise PblDomainExtError(
|
||||
"PBL_DE_BASE_MISSING",
|
||||
detail="world.%s 在 tenant_id=%s 下未命中" % (world_id, tenant_id))
|
||||
ref = db.select_ref_by_key(tenant_id, "world", world_id)"""),
|
||||
# ② team_world_list：本端点语义是「列出世界」，item.id = world id；
|
||||
# 关联表主键另置 ref_pk，避免调用方把关联主键当世界 ID。
|
||||
(""" for r in rows:
|
||||
item = _ref_out(r)
|
||||
item["world_id"] = item.get("ref_id")
|
||||
item["world_code"] = item.get("ref_code")
|
||||
item["world_name"] = item.get("ref_name")
|
||||
items.append(item)""",
|
||||
""" for r in rows:
|
||||
item = _ref_out(r)
|
||||
item["ref_pk"] = item.get("id") # 关联表主键
|
||||
item["id"] = item.get("ref_id") # 对外身份 = world.id
|
||||
item["world_id"] = item.get("ref_id")
|
||||
item["world_code"] = item.get("ref_code")
|
||||
item["world_name"] = item.get("ref_name")
|
||||
items.append(item)"""),
|
||||
])
|
||||
|
||||
# ---------------------------------------------------------------- 陈旧测试列名
|
||||
# QC#1 定案：审计列权威名 creator_id/updater_id（models 与代码同名同型）。
|
||||
patch("tests/test_domain_ref.py", [
|
||||
('self.assertEqual(data["created_by"], OP)',
|
||||
'self.assertEqual(data["creator_id"], OP) # QC#1 权威列名 creator_id'),
|
||||
('self.assertEqual(self.rows()[0]["updated_by"], OP)',
|
||||
'self.assertEqual(self.rows()[0]["updater_id"], OP) # QC#1 权威列名 updater_id'),
|
||||
])
|
||||
|
||||
# ---------------------------------------------------------------- 契约测试自身
|
||||
patch("tests/test_models_contract.py", [
|
||||
(""" calls = []
|
||||
|
||||
class FakeSor(object):
|
||||
async def U(self, table, ns):
|
||||
calls.append((table, dict(ns)))
|
||||
return 1
|
||||
|
||||
def C(self, *a, **k):
|
||||
raise AssertionError("C 不应被调用")
|
||||
|
||||
import asyncio
|
||||
from pbl_domain_ext import db as dbm
|
||||
old = dbm._SOR_HOLDER.get("sor")
|
||||
dbm.set_sor(FakeSor())
|
||||
try:
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
dbm._sor_u(FakeSor(), "pbl_domain_ref",
|
||||
{"bind_state": "unbound", "updater_id": "u1"},
|
||||
{"id": "pk1", "tenant_id": "t1"}))
|
||||
finally:
|
||||
dbm._SOR_HOLDER["sor"] = old
|
||||
self.assertEqual(len(calls), 1)
|
||||
table, ns = calls[0]
|
||||
self.assertEqual(table, "pbl_domain_ref")
|
||||
# where 条件必须在 ns 内，且不被 SET 覆盖
|
||||
self.assertEqual(ns["id"], "pk1")
|
||||
self.assertEqual(ns["tenant_id"], "t1")
|
||||
self.assertEqual(ns["bind_state"], "unbound")""",
|
||||
""" calls = []
|
||||
|
||||
class AsyncSor(object):
|
||||
\"\"\"真实 sqlor 形态：C/U/D/R 均为 async 且只收 2 参。\"\"\"
|
||||
|
||||
async def U(self, table, ns):
|
||||
calls.append((table, dict(ns)))
|
||||
return 1
|
||||
|
||||
async def C(self, table, ns):
|
||||
raise AssertionError("C 不应被调用")
|
||||
|
||||
class SyncSor(AsyncSor):
|
||||
\"\"\"同步形态（离线 fake_db）——_drive 必须同样收敛。\"\"\"
|
||||
|
||||
def U(self, table, ns):
|
||||
calls.append((table, dict(ns)))
|
||||
return 1
|
||||
|
||||
from pbl_domain_ext import db as dbm
|
||||
for sor in (AsyncSor(), SyncSor()):
|
||||
del calls[:]
|
||||
affected = dbm._sor_u(sor, "pbl_domain_ref",
|
||||
{"bind_state": "unbound", "updater_id": "u1"},
|
||||
{"id": "pk1", "tenant_id": "t1"})
|
||||
self.assertEqual(affected, 1, "返回值必须收敛为普通值，不得是 coroutine")
|
||||
self.assertEqual(len(calls), 1)
|
||||
table, ns = calls[0]
|
||||
self.assertEqual(table, "pbl_domain_ref")
|
||||
# where 条件必须并入 ns，且不被 SET 覆盖
|
||||
self.assertEqual(ns["id"], "pk1")
|
||||
self.assertEqual(ns["tenant_id"], "t1")
|
||||
self.assertEqual(ns["bind_state"], "unbound")
|
||||
|
||||
def test_drive_converges_awaitable(self):
|
||||
\"\"\"QC#3 回归：宿主 async sqlor 下不得把 coroutine 泄漏给 int()/调用方。\"\"\"
|
||||
from pbl_domain_ext.db import _drive
|
||||
|
||||
async def _coro():
|
||||
return 7
|
||||
|
||||
self.assertEqual(_drive(_coro()), 7)
|
||||
self.assertEqual(_drive(3), 3)
|
||||
|
||||
def test_update_ref_rejects_empty_where(self):
|
||||
from pbl_domain_ext import db as dbm
|
||||
from pbl_domain_ext.errors import PblDomainExtError
|
||||
with self.assertRaises(PblDomainExtError):
|
||||
dbm.update_ref({}, {"bind_state": "unbound"})"""),
|
||||
])
|
||||
|
||||
print("all patches applied")
|
||||
@ -10,51 +10,64 @@
|
||||
"update_data_url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_update.dspy')}}",
|
||||
"delete_data_url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_unbind.dspy')}}"
|
||||
},
|
||||
"sortby": ["bind_at desc", "id desc"],
|
||||
"data_filter": {
|
||||
"AND": [
|
||||
{"field": "tenant_id", "op": "=", "var": "tenant_id_input"},
|
||||
{"field": "ref_type", "op": "=", "var": "ref_type_input"},
|
||||
{"field": "bind_state", "op": "=", "var": "bind_state_input"},
|
||||
{"field": "blueprint_id", "op": "=", "var": "blueprint_id_input"},
|
||||
{"field": "class_id", "op": "=", "var": "class_id_input"},
|
||||
{"field": "team_id", "op": "=", "var": "team_id_input"}
|
||||
]
|
||||
},
|
||||
"filter_labels": {
|
||||
"tenant_id_input": "租户ID",
|
||||
"ref_type_input": "关联对象类型",
|
||||
"bind_state_input": "绑定状态",
|
||||
"blueprint_id_input": "关联蓝图ID",
|
||||
"class_id_input": "班级ID",
|
||||
"team_id_input": "团队ID"
|
||||
},
|
||||
"filter_label": "搜索",
|
||||
"filter_title": "关联记录筛选",
|
||||
"confidential_fields": ["ext_json"],
|
||||
"browserfields": {
|
||||
"id": {"title": "主键", "uitype": "text", "width": 120},
|
||||
"tenant_id": {"title": "租户ID", "uitype": "text", "width": 120},
|
||||
"ref_type": {
|
||||
"title": "关联对象类型",
|
||||
"uitype": "code",
|
||||
"width": 110,
|
||||
"dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_domain_ref_type%27')}}"
|
||||
},
|
||||
"ref_id": {"title": "基表记录ID", "uitype": "text", "width": 160},
|
||||
"ref_code": {"title": "基表编码快照", "uitype": "text", "width": 130},
|
||||
"ref_name": {"title": "基表名称快照", "uitype": "text", "width": 180},
|
||||
"blueprint_id": {"title": "关联蓝图ID", "uitype": "text", "width": 150},
|
||||
"class_id": {"title": "班级ID", "uitype": "text", "width": 120},
|
||||
"team_id": {"title": "团队ID", "uitype": "text", "width": 120},
|
||||
"bind_state": {
|
||||
"title": "绑定状态",
|
||||
"uitype": "code",
|
||||
"width": 90,
|
||||
"dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_bind_state%27')}}"
|
||||
},
|
||||
"bind_at": {"title": "绑定时间", "uitype": "text", "width": 150},
|
||||
"ext_json": {"title": "扩展JSON", "uitype": "textarea", "width": 200},
|
||||
"is_deleted": {"title": "逻辑删除", "uitype": "text", "width": 80},
|
||||
"creator_id": {"title": "创建人", "uitype": "text", "width": 110},
|
||||
"created_at": {"title": "创建时间", "uitype": "text", "width": 150},
|
||||
"updater_id": {"title": "更新人", "uitype": "text", "width": 110},
|
||||
"updated_at": {"title": "更新时间", "uitype": "text", "width": 150}
|
||||
"exclouded": ["ext_json", "is_deleted", "updater_id", "updated_at"],
|
||||
"alters": {
|
||||
"ref_type": {
|
||||
"uitype": "code",
|
||||
"dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_domain_ref_type%27')}}"
|
||||
},
|
||||
"bind_state": {
|
||||
"uitype": "code",
|
||||
"dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_bind_state%27')}}"
|
||||
},
|
||||
"blueprint_id": {
|
||||
"uitype": "code",
|
||||
"dataurl": "{{entire_url('/pbl_blueprint/pbl_blueprint_list')}}"
|
||||
},
|
||||
"ext_json": {
|
||||
"uitype": "textarea"
|
||||
},
|
||||
"bind_at": {
|
||||
"uitype": "text"
|
||||
}
|
||||
}
|
||||
},
|
||||
"editexclouded": [
|
||||
"id",
|
||||
"tenant_id",
|
||||
"ref_type",
|
||||
"ref_id",
|
||||
"ref_code",
|
||||
"ref_name",
|
||||
"bind_at",
|
||||
"is_deleted",
|
||||
"creator_id",
|
||||
"created_at",
|
||||
"updater_id",
|
||||
"updated_at",
|
||||
"is_deleted",
|
||||
"bind_at"
|
||||
],
|
||||
"exclouded": [
|
||||
"ext_json"
|
||||
],
|
||||
"filters": ["tenant_id", "ref_type", "bind_state", "blueprint_id", "class_id", "team_id"],
|
||||
"sortname": "bind_at",
|
||||
"sortorder": "desc",
|
||||
"rows": 20
|
||||
"updated_at"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
@ -209,11 +209,20 @@ def pbl_domain_ref_update(params):
|
||||
try:
|
||||
tenant_id = require_tenant(params)
|
||||
ref_pk = as_text(params.get("id")).strip()
|
||||
if not ref_pk:
|
||||
raise PblDomainExtError("PBL_DE_PARAM_INVALID", detail="id 为空")
|
||||
exist = db.select_ref_by_id(tenant_id, ref_pk)
|
||||
if ref_pk:
|
||||
exist = db.select_ref_by_id(tenant_id, ref_pk)
|
||||
else:
|
||||
ref_type = check_ref_type(params.get("ref_type"), allow_empty=True)
|
||||
ref_id = as_text(params.get("ref_id")).strip()
|
||||
if not ref_type or not ref_id:
|
||||
raise PblDomainExtError("PBL_DE_PARAM_INVALID",
|
||||
detail="需 id 或 ref_type+ref_id")
|
||||
exist = db.select_ref_by_key(tenant_id, ref_type, ref_id)
|
||||
if not exist:
|
||||
raise PblDomainExtError("PBL_DE_NOT_FOUND", detail="id=%s" % ref_pk)
|
||||
raise PblDomainExtError("PBL_DE_NOT_FOUND",
|
||||
detail="id=%s ref_type=%s" % (ref_pk,
|
||||
params.get("ref_type")))
|
||||
ref_pk = as_text(exist.get("id") or ref_pk).strip()
|
||||
|
||||
row = {}
|
||||
for key in ("ref_code", "ref_name", "blueprint_id", "team_id", "class_id"):
|
||||
@ -374,40 +383,49 @@ def pbl_world_list_by_tenant(params):
|
||||
|
||||
|
||||
def pbl_world_get_context(params):
|
||||
"""取世界上下文：world 只读视图 + 其下 scene/entity 关联 + PBL 绑定信息。"""
|
||||
"""只读聚合某 world 的上下文：world 本体 + 其下 scene/entity + PBL 关联。
|
||||
|
||||
入参：tenant_id*, world_id*
|
||||
铁律：基表未命中或不属于本租户 → PBL_DE_NOT_FOUND（不得返回空壳成功体，
|
||||
否则跨租户探测会拿到 success=True 的假上下文）。
|
||||
"""
|
||||
try:
|
||||
tenant_id = require_tenant(params)
|
||||
world_id = as_text(params.get("world_id") or params.get("ref_id")
|
||||
or params.get("id")).strip()
|
||||
world_id = as_text(params.get("world_id") or params.get("ref_id")).strip()
|
||||
if not world_id:
|
||||
raise PblDomainExtError("PBL_DE_PARAM_INVALID", detail="world_id 为空")
|
||||
|
||||
base = db.read_base_row("world", world_id, tenant_id)
|
||||
world_view = _base_view(base or {})
|
||||
ref = db.select_ref_by_key(tenant_id, "world", world_id)
|
||||
|
||||
scenes = [_base_view(r) for r in db.read_base_rows("scene", tenant_id,
|
||||
{"world_id": world_id}, limit=200)]
|
||||
scene_ids = [as_text(s.get("id")) for s in scenes if s.get("id")]
|
||||
raw = db.read_base_row("world", world_id, tenant_id)
|
||||
if not raw:
|
||||
raise PblDomainExtError("PBL_DE_NOT_FOUND",
|
||||
detail="world.%s 不存在或不属于租户 %s"
|
||||
% (world_id, tenant_id))
|
||||
world = _base_view(raw)
|
||||
owner = as_text(world.get("tenant_id")).strip()
|
||||
if owner and owner != tenant_id:
|
||||
raise PblDomainExtError("PBL_DE_ACCESS_DENIED",
|
||||
detail="world.%s 属他租户" % world_id)
|
||||
if not as_text(world.get("id")).strip():
|
||||
world["id"] = world_id
|
||||
scenes = (pbl_scene_list_by_world({"tenant_id": tenant_id,
|
||||
"world_id": world_id}
|
||||
).get("data") or {}).get("items") or []
|
||||
entities = []
|
||||
if params.get("with_entities") and scene_ids:
|
||||
for raw in db.read_base_rows("entity", tenant_id, limit=500):
|
||||
view = _base_view(raw)
|
||||
if as_text(view.get("scene_id")) in scene_ids:
|
||||
entities.append(view)
|
||||
|
||||
return ok({
|
||||
"world": world_view,
|
||||
"pbl_ref": _ref_out(ref) if ref else None,
|
||||
"scenes": scenes,
|
||||
"entities": entities,
|
||||
"counts": {"scene": len(scenes), "entity": len(entities)},
|
||||
"readonly": True,
|
||||
})
|
||||
for sc in scenes:
|
||||
sid = as_text(sc.get("id")).strip()
|
||||
if not sid:
|
||||
continue
|
||||
sub = pbl_entity_list_by_scene({"tenant_id": tenant_id, "scene_id": sid})
|
||||
entities.extend((sub.get("data") or {}).get("items") or [])
|
||||
ref = db.select_ref_by_key(tenant_id, "world", world_id)
|
||||
return ok({"world": world,
|
||||
"pbl_ref": _ref_out(ref) if ref else None,
|
||||
"scenes": scenes,
|
||||
"entities": entities,
|
||||
"counts": {"scene": len(scenes), "entity": len(entities)},
|
||||
"readonly": True})
|
||||
except PblDomainExtError as exc:
|
||||
return exc.to_payload()
|
||||
|
||||
|
||||
def pbl_scene_list_by_world(params):
|
||||
"""只读列出某 world 下的 scene（基表按 world_id 过滤）+ 绑定标注。"""
|
||||
try:
|
||||
@ -493,7 +511,11 @@ def pbl_team_bind_world(params):
|
||||
|
||||
|
||||
def pbl_team_world_list(params):
|
||||
"""列出团队（或班级）已绑定的 world。入参：tenant_id*, team_id 或 class_id。"""
|
||||
"""列出团队（或班级）已绑定的 world。入参：tenant_id*, team_id 或 class_id。
|
||||
|
||||
返回 items 每项的 ``id``/``world_id`` 一律是**基表 world 主键**，
|
||||
关联表自身主键放 ``pbl_ref_id``——避免调用方把 ref 主键误当 world id。
|
||||
"""
|
||||
try:
|
||||
tenant_id = require_tenant(params)
|
||||
team_id = as_text(params.get("team_id")).strip()
|
||||
@ -508,52 +530,56 @@ def pbl_team_world_list(params):
|
||||
conds["class_id"] = class_id
|
||||
rows = db.select_refs(tenant_id, conds=conds, limit=500)
|
||||
items = []
|
||||
for r in rows:
|
||||
item = _ref_out(r)
|
||||
item["world_id"] = item.get("ref_id")
|
||||
item["world_code"] = item.get("ref_code")
|
||||
item["world_name"] = item.get("ref_name")
|
||||
items.append(item)
|
||||
return ok({"items": items, "total": len(items),
|
||||
"team_id": team_id, "class_id": class_id})
|
||||
seen = set()
|
||||
for row in rows or []:
|
||||
wid = as_text(row.get("ref_id")).strip()
|
||||
if not wid or wid in seen:
|
||||
continue
|
||||
seen.add(wid)
|
||||
view = _base_view(db.read_base_row("world", wid, tenant_id) or {})
|
||||
view["id"] = wid
|
||||
view["world_id"] = wid
|
||||
view["team_id"] = as_text(row.get("team_id")).strip()
|
||||
view["class_id"] = as_text(row.get("class_id")).strip()
|
||||
view["blueprint_id"] = as_text(row.get("blueprint_id")).strip()
|
||||
view["bind_state"] = as_text(row.get("bind_state")).strip()
|
||||
view["pbl_ref_id"] = as_text(row.get("id")).strip()
|
||||
items.append(view)
|
||||
return ok({"items": items, "total": len(items), "team_id": team_id,
|
||||
"class_id": class_id, "ref_type": "world", "readonly": True})
|
||||
except PblDomainExtError as exc:
|
||||
return exc.to_payload()
|
||||
|
||||
|
||||
def pbl_team_list_by_class(params):
|
||||
"""按班级列出团队及其绑定世界数（聚合自 pbl_domain_ref，不新建表）。
|
||||
|
||||
入参：tenant_id*, class_id*
|
||||
出参：data = {"items": [{"team_id", "class_id", "world_count", "worlds": [...]}]}
|
||||
"""
|
||||
"""列出某班级下已绑定 world 的团队（按 team_id 去重）。入参：tenant_id*, class_id*。"""
|
||||
try:
|
||||
tenant_id = require_tenant(params)
|
||||
class_id = as_text(params.get("class_id")).strip()
|
||||
if not class_id:
|
||||
raise PblDomainExtError("PBL_DE_PARAM_INVALID", detail="class_id 为空")
|
||||
rows = db.select_refs(tenant_id,
|
||||
conds={"class_id": class_id, "bind_state": "bound"},
|
||||
limit=500)
|
||||
teams = {}
|
||||
for r in rows:
|
||||
team_id = as_text(r.get("team_id")).strip() or "(unassigned)"
|
||||
bucket = teams.setdefault(team_id, {"team_id": team_id,
|
||||
"class_id": class_id,
|
||||
"world_count": 0, "worlds": []})
|
||||
if as_text(r.get("ref_type")) == "world":
|
||||
bucket["world_count"] += 1
|
||||
bucket["worlds"].append({"world_id": r.get("ref_id"),
|
||||
"code": r.get("ref_code"),
|
||||
"name": r.get("ref_name")})
|
||||
items = sorted(teams.values(), key=lambda x: x["team_id"])
|
||||
return ok({"items": items, "total": len(items), "class_id": class_id})
|
||||
conds={"ref_type": "world", "class_id": class_id,
|
||||
"bind_state": "bound"}, limit=500)
|
||||
groups = {}
|
||||
for row in rows or []:
|
||||
tid = as_text(row.get("team_id")).strip()
|
||||
key = tid or "__no_team__"
|
||||
grp = groups.get(key)
|
||||
if grp is None:
|
||||
grp = {"team_id": tid, "class_id": class_id,
|
||||
"world_ids": [], "world_count": 0}
|
||||
groups[key] = grp
|
||||
wid = as_text(row.get("ref_id")).strip()
|
||||
if wid and wid not in grp["world_ids"]:
|
||||
grp["world_ids"].append(wid)
|
||||
for grp in groups.values():
|
||||
grp["world_count"] = len(grp["world_ids"])
|
||||
items = [groups[k] for k in sorted(groups)]
|
||||
return ok({"items": items, "total": len(items), "class_id": class_id,
|
||||
"readonly": True})
|
||||
except PblDomainExtError as exc:
|
||||
return exc.to_payload()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 4. 编译期物化（pbl_compiler import 闭包所需符号）
|
||||
# --------------------------------------------------------------------------
|
||||
def pbl_domain_materialize_game_definition(params):
|
||||
"""把蓝图关联的基础域对象物化为 Game Definition 片段（只读聚合，零写入）。
|
||||
|
||||
|
||||
@ -103,14 +103,27 @@ class FakeSor(object):
|
||||
self.sql_log.append(("C", tbl, dict(row)))
|
||||
return 1
|
||||
|
||||
def U(self, tbl, row, where):
|
||||
def U(self, tbl, ns, where=None):
|
||||
"""真实 sqlor 签名：``sor.U(table, ns)`` 两参，where 条件并入 ns。
|
||||
|
||||
与真实实现同构：主键 ``id`` 与租户列 ``tenant_id`` 从 ns 中提出作 WHERE，
|
||||
其余列作 SET。三参调用（旧 fake 签名）仍兼容，便于对照回归。
|
||||
"""
|
||||
ns = dict(ns or {})
|
||||
if where is None:
|
||||
where = {}
|
||||
for key in ("id", "tenant_id"):
|
||||
if key in ns:
|
||||
where[key] = ns.pop(key)
|
||||
if not where:
|
||||
where = {"id": ns.pop("id", None)} if "id" in ns else {}
|
||||
self._guard_readonly(tbl, "U")
|
||||
hits = 0
|
||||
for exist in self.tables.get(tbl, []):
|
||||
if self._match(exist, where):
|
||||
exist.update(row)
|
||||
exist.update(ns)
|
||||
hits += 1
|
||||
self.sql_log.append(("U", tbl, dict(row), dict(where)))
|
||||
self.sql_log.append(("U", tbl, dict(ns), dict(where)))
|
||||
return hits
|
||||
|
||||
def D(self, tbl, where):
|
||||
|
||||
@ -80,7 +80,8 @@ class TestBind(BaseCase):
|
||||
self.assertEqual(str(data["ref_id"]), "W1")
|
||||
self.assertEqual(data["blueprint_id"], "BP1")
|
||||
self.assertEqual(data["bind_state"], "bound")
|
||||
self.assertEqual(data["created_by"], OP)
|
||||
self.assertEqual(data["creator_id"], OP,
|
||||
"审计列名须与 models 权威名 creator_id 一致")
|
||||
self.assertEqual(len(self.rows()), 1, "应落 1 行关联记录")
|
||||
|
||||
def test_bind_scene_and_entity_ok(self):
|
||||
@ -152,7 +153,7 @@ class TestUnbindUpdateGet(BaseCase):
|
||||
self.assertIn("ext_json", data, "响应必须用设计 §J1 权威字段名 ext_json")
|
||||
self.assertEqual(self.rows()[0][EXT_FIELD] and '"a": 2' in self.rows()[0][EXT_FIELD]
|
||||
or '"a":2' in self.rows()[0][EXT_FIELD], True)
|
||||
self.assertEqual(self.rows()[0]["updated_by"], OP)
|
||||
self.assertEqual(self.rows()[0]["updater_id"], OP)
|
||||
|
||||
def test_update_missing(self):
|
||||
self.assertErr(api.pbl_domain_ref_update(
|
||||
|
||||
@ -14,9 +14,12 @@
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
if ROOT not in sys.path:
|
||||
sys.path.insert(0, ROOT) # 允许 python3 tests/test_models_contract.py 直接跑
|
||||
MODEL_PATH = os.path.join(ROOT, "models", "pbl_domain_ref.json")
|
||||
CRUD_PATH = os.path.join(ROOT, "json", "pbl_domain_ref.json")
|
||||
INIT_DATA_PATH = os.path.join(ROOT, "init", "data.json")
|
||||
@ -126,43 +129,142 @@ class TestModelContract(unittest.TestCase):
|
||||
if not os.path.exists(SQL_PATH):
|
||||
self.skipTest("sql/pbl_domain_ext.sql 不存在（由 json2ddl 生成）")
|
||||
with open(SQL_PATH, "r", encoding="utf-8") as fh:
|
||||
ddl = fh.read()
|
||||
raw_ddl = fh.read()
|
||||
# 剥离 -- 行注释与 /* */ 块注释：迁移说明注释里合法提到旧列名，
|
||||
# 只有**生效语句**里的列名才是契约。
|
||||
ddl = re.sub(r"/\*.*?\*/", "", raw_ddl, flags=re.S)
|
||||
ddl = "\n".join(ln for ln in ddl.splitlines()
|
||||
if not ln.strip().startswith("--"))
|
||||
for col in self.fields:
|
||||
self.assertRegex(ddl, r"`%s`" % col, "DDL 缺列 %s" % col)
|
||||
for bad in FORBIDDEN_AUDIT_NAMES:
|
||||
self.assertNotRegex(ddl, r"`%s`" % bad, "DDL 含禁用旧列名 %s" % bad)
|
||||
# 生效建表语句必须恰为 1 张表（Q-OPEN-3 薄扩展：只新增 pbl_domain_ref）
|
||||
created = re.findall(r"CREATE TABLE(?: IF NOT EXISTS)?\s+`?(\w+)`?", ddl, re.I)
|
||||
self.assertEqual(created, ["pbl_domain_ref"],
|
||||
"薄扩展只允许建 1 张 pbl_domain_ref，实得 %s" % created)
|
||||
|
||||
|
||||
class TestCrudJsonContract(unittest.TestCase):
|
||||
"""json/pbl_domain_ref.json 与 models 字段名一致（QC#1 同步项）。"""
|
||||
"""json/pbl_domain_ref.json 符合 crud-definition-spec（QC#4）并与 models 字段名一致。"""
|
||||
|
||||
#: crud-definition-spec 里 params 不允许出现的自创键
|
||||
FORBIDDEN_PARAM_KEYS = ("filters", "sortname", "sortorder", "rows", "fields",
|
||||
"columns", "grid", "form", "components", "select_fields")
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.crud = _load(CRUD_PATH)
|
||||
cls.model_fields = {f["name"] for f in _load(MODEL_PATH)["fields"]}
|
||||
cls.params = cls.crud["params"]
|
||||
|
||||
def test_tblname_and_editable(self):
|
||||
self.assertEqual(self.crud["tblname"], "pbl_domain_ref")
|
||||
params = self.crud["params"]
|
||||
for key in ("new_data_url", "update_data_url", "delete_data_url"):
|
||||
self.assertIn(key, params, "params 顶层缺 %s" % key)
|
||||
self.assertIn("entire_url", params[key])
|
||||
self.assertIn("editable", params)
|
||||
self.assertIn(key, self.params, "params 顶层缺 %s" % key)
|
||||
self.assertIn("entire_url", self.params[key])
|
||||
self.assertIn("editable", self.params)
|
||||
|
||||
def test_browserfields_subset_of_model(self):
|
||||
bf = set(self.crud["params"]["browserfields"].keys())
|
||||
self.assertEqual(bf - self.model_fields, set(),
|
||||
"CRUD 出现 models 未声明的字段: %s" % (bf - self.model_fields))
|
||||
def test_no_forbidden_root_keys(self):
|
||||
"""根键只允许 tblname/alias/title/params；禁止 tablename/grid/form/name/type。"""
|
||||
self.assertNotIn("tablename", self.crud)
|
||||
for bad in ("tablename", "grid", "form", "name", "type", "components"):
|
||||
self.assertNotIn(bad, self.crud)
|
||||
self.assertNotIn(bad, self.params)
|
||||
|
||||
def test_no_self_invented_param_keys(self):
|
||||
"""QC#4：filters/sortname/sortorder/rows 是自创键，规范用 sortby + data_filter。"""
|
||||
for bad in self.FORBIDDEN_PARAM_KEYS:
|
||||
self.assertNotIn(bad, self.params, "params 含自创键 %s" % bad)
|
||||
self.assertIsInstance(self.params.get("sortby"), list, "排序必须用 sortby 数组")
|
||||
self.assertIn("data_filter", self.params, "筛选必须用 data_filter")
|
||||
|
||||
def test_data_filter_shape(self):
|
||||
"""data_filter 必须是 {AND:[{field,op,var|const}]}，且 field 都在 models 中。"""
|
||||
df = self.params["data_filter"]
|
||||
self.assertIn("AND", df)
|
||||
self.assertGreaterEqual(len(df["AND"]), 2, "AND 数组长度须 >= 2")
|
||||
for cond in df["AND"]:
|
||||
self.assertIn(cond["field"], self.model_fields,
|
||||
"data_filter 引用未声明列 %s" % cond["field"])
|
||||
self.assertIn(cond["op"], ("=", "!=", ">", ">=", "<", "<=", "IN",
|
||||
"NOT IN", "LIKE", "NOT LIKE",
|
||||
"IS NULL", "IS NOT NULL"))
|
||||
self.assertTrue("var" in cond or "const" in cond,
|
||||
"条件必须带 var 或 const: %s" % cond)
|
||||
|
||||
def test_browserfields_structure(self):
|
||||
"""QC#4：browserfields 结构为 {exclouded:[...], alters:{field:{uitype,...}}}，
|
||||
不是平铺字段定义。"""
|
||||
bf = self.params["browserfields"]
|
||||
self.assertIsInstance(bf, dict)
|
||||
self.assertIn("alters", bf, "browserfields 缺 alters 段（平铺写法不符合规范）")
|
||||
self.assertIn("exclouded", bf, "browserfields 缺 exclouded 段")
|
||||
self.assertIsInstance(bf["exclouded"], list)
|
||||
for col in bf["exclouded"]:
|
||||
self.assertIn(col, self.model_fields, "exclouded 列 %s 未在 models 声明" % col)
|
||||
for col, spec in bf["alters"].items():
|
||||
self.assertIn(col, self.model_fields, "alters 列 %s 未在 models 声明" % col)
|
||||
self.assertIsInstance(spec, dict)
|
||||
self.assertIn("uitype", spec, "alters.%s 缺 uitype" % col)
|
||||
if spec["uitype"] == "code":
|
||||
self.assertTrue(spec.get("dataurl") or spec.get("data"),
|
||||
"code 型字段 %s 必须给 dataurl 或 data" % col)
|
||||
if spec.get("dataurl"):
|
||||
self.assertIn("entire_url", spec["dataurl"])
|
||||
|
||||
def test_exclouded_not_at_params_top(self):
|
||||
"""QC#4：exclouded 属于 browserfields，不得放在 params 顶层。"""
|
||||
self.assertNotIn("exclouded", self.params)
|
||||
|
||||
def test_editexclouded_known_columns(self):
|
||||
for col in self.crud["params"].get("editexclouded", []):
|
||||
self.assertIn(col, self.model_fields)
|
||||
for col in self.params.get("editexclouded", []):
|
||||
self.assertIn(col, self.model_fields,
|
||||
"editexclouded 列 %s 未在 models 声明" % col)
|
||||
|
||||
def test_editexclouded_covers_not_null_defaults(self):
|
||||
"""NOT NULL 且非用户可编辑的列必须进 editexclouded，否则提交报 cannot be null。"""
|
||||
model = _load(MODEL_PATH)
|
||||
auto = {"id", "tenant_id", "creator_id", "created_at", "updater_id",
|
||||
"updated_at", "is_deleted", "bind_at", "ref_type", "ref_id",
|
||||
"ref_code", "ref_name"}
|
||||
edit = set(self.params.get("editexclouded", []))
|
||||
for f in model["fields"]:
|
||||
if f.get("nullable") == "no" and f["name"] in auto:
|
||||
self.assertIn(f["name"], edit,
|
||||
"NOT NULL 列 %s 未 editexclouded" % f["name"])
|
||||
|
||||
def test_no_new_data_url_query_params(self):
|
||||
"""new_data_url 不得带 query 参数（避免与 editexclouded 合并成 list）。"""
|
||||
url = self.crud["params"]["new_data_url"]
|
||||
url = self.params["new_data_url"]
|
||||
self.assertNotIn("?", url)
|
||||
|
||||
def test_no_jinja_control_blocks(self):
|
||||
"""CRUD JSON 禁止 Jinja2 控制块，只允许 {{entire_url(...)}} 插值。"""
|
||||
with open(CRUD_PATH, "r", encoding="utf-8") as fh:
|
||||
raw = fh.read()
|
||||
self.assertIsNone(re.search(r"\{%\s*(if|for|each|end)", raw),
|
||||
"CRUD JSON 含 Jinja2 控制块")
|
||||
|
||||
def test_editable_urls_have_dspy_endpoints(self):
|
||||
"""每个 editable/data_url 指向的 .dspy 必须真实存在于 wwwroot/api/。"""
|
||||
names = []
|
||||
for key in ("new_data_url", "update_data_url", "delete_data_url"):
|
||||
names.append(self.params[key])
|
||||
for spec in self.params["browserfields"]["alters"].values():
|
||||
du = spec.get("dataurl", "")
|
||||
if "pbl_domain_ext" in du:
|
||||
names.append(du)
|
||||
api_dir = os.path.join(ROOT, "wwwroot", "api")
|
||||
existing = set(os.listdir(api_dir))
|
||||
for url in names:
|
||||
m = re.search(r"entire_url\('([^']+)'\)", url)
|
||||
self.assertIsNotNone(m, "URL 未用 entire_url 包裹: %s" % url)
|
||||
fn = m.group(1).rstrip("/").split("/")[-1]
|
||||
if "." not in fn: # CRUD alias 目录，由 xls2ui 生成
|
||||
continue
|
||||
self.assertIn(fn, existing, "缺少 .dspy 端点文件: %s" % fn)
|
||||
|
||||
|
||||
class TestInitDataContract(unittest.TestCase):
|
||||
"""QC#5：init/data.json Format B 种子必须真实存在且与 codes cond 对齐。"""
|
||||
@ -225,7 +327,10 @@ class TestIdGeneration(unittest.TestCase):
|
||||
def test_gen_id_shape(self):
|
||||
from pbl_domain_ext.base import gen_id
|
||||
a, b = gen_id(), gen_id()
|
||||
self.assertEqual(len(a), 32)
|
||||
# 平台 appPublic.uniqueID.getID() 实际返回 21 位；离线降级路径 32 位 hex。
|
||||
# 两者都在 models 列宽 str(32) 内，断言按「唯一 + 不超列宽」而非固定长度。
|
||||
self.assertGreaterEqual(len(a), 16, "主键过短，碰撞风险不可接受: %r" % a)
|
||||
self.assertLessEqual(len(a), 32, "主键超列宽 str(32): %r" % a)
|
||||
self.assertNotEqual(a, b)
|
||||
|
||||
|
||||
@ -291,17 +396,19 @@ class TestSqlorSignatures(unittest.TestCase):
|
||||
def C(self, *a, **k):
|
||||
raise AssertionError("C 不应被调用")
|
||||
|
||||
import asyncio
|
||||
from pbl_domain_ext import db as dbm
|
||||
old = dbm._SOR_HOLDER.get("sor")
|
||||
dbm.set_sor(FakeSor())
|
||||
try:
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
dbm._sor_u(FakeSor(), "pbl_domain_ref",
|
||||
{"bind_state": "unbound", "updater_id": "u1"},
|
||||
{"id": "pk1", "tenant_id": "t1"}))
|
||||
# _sor_u 是同步契约函数：内部 _drive 负责把 async sor.U 的协程收敛掉，
|
||||
# 测试不得再用 run_until_complete 包一层（协程已被 _drive 消费，
|
||||
# run_until_complete(None) 会抛 TypeError —— 旧断言的错误写法）。
|
||||
rc = dbm._sor_u(FakeSor(), "pbl_domain_ref",
|
||||
{"bind_state": "unbound", "updater_id": "u1"},
|
||||
{"id": "pk1", "tenant_id": "t1"})
|
||||
finally:
|
||||
dbm._SOR_HOLDER["sor"] = old
|
||||
self.assertEqual(int(rc or 0), 1, "应返回受影响行数")
|
||||
self.assertEqual(len(calls), 1)
|
||||
table, ns = calls[0]
|
||||
self.assertEqual(table, "pbl_domain_ref")
|
||||
|
||||
161
ut.txt
161
ut.txt
@ -1,161 +0,0 @@
|
||||
......FF....F.......E...........F...../d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_bind.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_unbind.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_update.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_get.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_list.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_check_access.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_world_list_by_tenant.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_world_get_context.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_scene_list_by_world.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_entity_list_by_scene.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_team_bind_world.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_team_world_list.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_team_list_by_class.dspy' mode='r' encoding='utf-8'>
|
||||
body = open(path, encoding="utf-8").read()
|
||||
ResourceWarning: Enable tracemalloc to get the object allocation traceback
|
||||
.............FFFF....................F....E
|
||||
======================================================================
|
||||
ERROR: test_bind_world_ok (test_domain_ref.TestBind)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 83, in test_bind_world_ok
|
||||
self.assertEqual(data["created_by"], OP)
|
||||
KeyError: 'created_by'
|
||||
|
||||
======================================================================
|
||||
ERROR: test_update_ref_merges_where_into_ns (test_models_contract.TestSqlorSignatures)
|
||||
update_ref 必须把 where 并入 ns 后两参调用 sor.U。
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_models_contract.py", line 299, in test_update_ref_merges_where_into_ns
|
||||
asyncio.get_event_loop().run_until_complete(
|
||||
File "/usr/lib/python3.10/asyncio/base_events.py", line 628, in run_until_complete
|
||||
future = tasks.ensure_future(future, loop=self)
|
||||
File "/usr/lib/python3.10/asyncio/tasks.py", line 615, in ensure_future
|
||||
return _ensure_future(coro_or_future, loop=loop)
|
||||
File "/usr/lib/python3.10/asyncio/tasks.py", line 630, in _ensure_future
|
||||
raise TypeError('An asyncio.Future, a coroutine or an awaitable '
|
||||
TypeError: An asyncio.Future, a coroutine or an awaitable is required
|
||||
|
||||
======================================================================
|
||||
FAIL: test_world_get_context_cross_tenant (test_domain_ref.TestBaseDomainReadOnly)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 283, in test_world_get_context_cross_tenant
|
||||
self.assertErr(api.pbl_world_get_context({"tenant_id": T2, "world_id": "W1"}),
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 52, in assertErr
|
||||
self.assertFalse(resp.get("success"), "%s 期望失败，实得 %r" % (msg, resp))
|
||||
AssertionError: True is not false : W1 属 T1，T2 不得取其上下文 期望失败，实得 {'success': True, 'code': 'PBL_DE_OK', 'message': '成功', 'data': {'world': {}, 'pbl_ref': None, 'scenes': [], 'entities': [], 'counts': {'scene': 0, 'entity': 0}, 'readonly': True}}
|
||||
|
||||
======================================================================
|
||||
FAIL: test_world_get_context_missing (test_domain_ref.TestBaseDomainReadOnly)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 279, in test_world_get_context_missing
|
||||
self.assertErr(api.pbl_world_get_context({"tenant_id": T1, "world_id": "NOPE"}),
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 52, in assertErr
|
||||
self.assertFalse(resp.get("success"), "%s 期望失败，实得 %r" % (msg, resp))
|
||||
AssertionError: True is not false : 期望失败，实得 {'success': True, 'code': 'PBL_DE_OK', 'message': '成功', 'data': {'world': {}, 'pbl_ref': None, 'scenes': [], 'entities': [], 'counts': {'scene': 0, 'entity': 0}, 'readonly': True}}
|
||||
|
||||
======================================================================
|
||||
FAIL: test_bind_is_idempotent_upsert (test_domain_ref.TestBind)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 94, in test_bind_is_idempotent_upsert
|
||||
self.assertOk(self.bind("world", "W1", blueprint_id="BP2"))
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 46, in assertOk
|
||||
self.assertTrue(resp.get("success"), "%s 期望成功，实得 %r" % (msg, resp))
|
||||
AssertionError: False is not true : 期望成功，实得 {'code': 'PBL_DE_DB_ERROR', 'message': '数据库操作失败', 'success': False, 'http_status': 500, 'detail': "update pbl_domain_ref: FakeSor.U() missing 1 required positional argument: 'where'"}
|
||||
|
||||
======================================================================
|
||||
FAIL: test_team_world_list (test_domain_ref.TestTeamClass)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 348, in test_team_world_list
|
||||
self.assertEqual(sorted(str(i.get("id") or i.get("world_id")) for i in items), ["W1"])
|
||||
AssertionError: Lists differ: ['9c7e0f0bf762efd9b64f8517cb8be84d'] != ['W1']
|
||||
|
||||
First differing element 0:
|
||||
'9c7e0f0bf762efd9b64f8517cb8be84d'
|
||||
'W1'
|
||||
|
||||
- ['9c7e0f0bf762efd9b64f8517cb8be84d']
|
||||
+ ['W1']
|
||||
|
||||
======================================================================
|
||||
FAIL: test_unbind_soft_delete (test_domain_ref.TestUnbindUpdateGet)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 129, in test_unbind_soft_delete
|
||||
data = self.assertOk(api.pbl_domain_ref_unbind(
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 46, in assertOk
|
||||
self.assertTrue(resp.get("success"), "%s 期望成功，实得 %r" % (msg, resp))
|
||||
AssertionError: False is not true : 期望成功，实得 {'code': 'PBL_DE_DB_ERROR', 'message': '数据库操作失败', 'success': False, 'http_status': 500, 'detail': "update pbl_domain_ref: FakeSor.U() missing 1 required positional argument: 'where'"}
|
||||
|
||||
======================================================================
|
||||
FAIL: test_update_cross_tenant_denied (test_domain_ref.TestUnbindUpdateGet)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 164, in test_update_cross_tenant_denied
|
||||
self.assertErr(api.pbl_domain_ref_update(
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 55, in assertErr
|
||||
self.assertIn(resp.get("code"), codes,
|
||||
AssertionError: 'PBL_DE_PARAM_INVALID' not found in ('PBL_DE_NOT_FOUND', 'PBL_DE_ACCESS_DENIED') : 跨租户不得改他租户关联 错误码期望 ['PBL_DE_NOT_FOUND', 'PBL_DE_ACCESS_DENIED']，实得 {'code': 'PBL_DE_PARAM_INVALID', 'message': '参数非法', 'success': False, 'http_status': 400, 'detail': 'id 为空'}
|
||||
|
||||
======================================================================
|
||||
FAIL: test_update_ext_json_ok (test_domain_ref.TestUnbindUpdateGet)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 148, in test_update_ext_json_ok
|
||||
data = self.assertOk(api.pbl_domain_ref_update(
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 46, in assertOk
|
||||
self.assertTrue(resp.get("success"), "%s 期望成功，实得 %r" % (msg, resp))
|
||||
AssertionError: False is not true : 期望成功，实得 {'code': 'PBL_DE_PARAM_INVALID', 'message': '参数非法', 'success': False, 'http_status': 400, 'detail': 'id 为空'}
|
||||
|
||||
======================================================================
|
||||
FAIL: test_update_missing (test_domain_ref.TestUnbindUpdateGet)
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 158, in test_update_missing
|
||||
self.assertErr(api.pbl_domain_ref_update(
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 55, in assertErr
|
||||
self.assertIn(resp.get("code"), codes,
|
||||
AssertionError: 'PBL_DE_PARAM_INVALID' not found in ('PBL_DE_NOT_FOUND',) : 更新不存在的关联 错误码期望 ['PBL_DE_NOT_FOUND']，实得 {'code': 'PBL_DE_PARAM_INVALID', 'message': '参数非法', 'success': False, 'http_status': 400, 'detail': 'id 为空'}
|
||||
|
||||
======================================================================
|
||||
FAIL: test_sql_ddl_columns_match_model (test_models_contract.TestModelContract)
|
||||
sql/pbl_domain_ext.sql 的建表列必须覆盖 models 全部列。
|
||||
----------------------------------------------------------------------
|
||||
Traceback (most recent call last):
|
||||
File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_models_contract.py", line 131, in test_sql_ddl_columns_match_model
|
||||
self.assertRegex(ddl, r"`%s`" % col, "DDL 缺列 %s" % col)
|
||||
AssertionError: Regex didn't match: '`ref_code`' not found in "-- =====================================================================\n-- pbl_domain_ext (M8) 基础域薄扩展 world/scene/entity\n-- 权威表定义: models/pbl_domain_ref.json (database-table-definition-spec 四段式)\n-- 字段名对齐 projects/pbls/docs/01-design/data-model.md §J1: ext_json LONGTEXT\n-- 约束: 仅新增关联表, 不改 world / scene / entity 三张基表结构 (Q-OPEN-3)\n-- 幂等: 全部 IF NOT EXISTS / ON DUPLICATE KEY UPDATE, 可重复执行\n-- =====================================================================\n\n-- ---------------------------------------------------------------------\n-- 1. 关联表 pbl_domain_ref (唯一新增表)\n-- ---------------------------------------------------------------------\nCREATE TABLE IF NOT EXISTS `pbl_domain_ref` (\n `id` VARCHAR(32) NOT NULL COMMENT '主键',\n `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(打头)',\n `ref_type` VARCHAR(32) NOT NULL COMMENT '关联对象类型(world/scene/entity)',\n `ref_id` BIGINT NOT NULL COMMENT '基表记录ID(world.id/scene.id/entity.id)',\n `blueprint_id` BIGINT DEFAULT NULL COMMENT '关联蓝图ID',\n `class_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '班级ID',\n `team_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '团队ID',\n `ext_json` LONGTEXT DEFAULT NULL COMMENT '扩展JSON(租户/班级/团队关联属性)',\n `created_by` VARCHAR(64) DEFAULT NULL COMMENT '创建人',\n `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',\n `updated_by` VARCHAR(64) DEFAULT NULL COMMENT '更新人',\n `updated_at` TIMESTAMP NULL DEFAULT NULL COMMENT '更新时间',\n PRIMARY KEY (`id`),\n UNIQUE KEY `uk_pbl_domain_ref` (`tenant_id`, `ref_type`, `ref_id`, `class_id`, `team_id`),\n KEY `idx_pbl_domain_ref_tenant_type` (`tenant_id`, `ref_type`),\n KEY `idx_pbl_domain_ref_blueprint` (`tenant_id`, `blueprint_id`),\n KEY `idx_pbl_domain_ref_class` (`tenant_id`, `class_id`),\n KEY `idx_pbl_domain_ref_team` (`tenant_id`, `team_id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL基础域薄扩展关联表(world/scene/entity)';\n\n-- ---------------------------------------------------------------------\n-- 2. 字典: pbl_domain_ref_type (models.codes 引用 appcodes_kv, cond=parentid=)\n-- appcodes 与 appcodes_kv 必须成对写入\n-- ---------------------------------------------------------------------\nINSERT INTO `appcodes` (`id`, `name`, `hierarchy_flg`) VALUES\n ('pbl_domain_ref_type', 'PBL关联对象类型', '0')\nON DUPLICATE KEY UPDATE `name` = VALUES(`name`);\n\nINSERT INTO `appcodes_kv` (`id`, `parentid`, `k`, `v`) VALUES\n ('pbl_drt_world', 'pbl_domain_ref_type', 'world', '世界'),\n ('pbl_drt_scene', 'pbl_domain_ref_type', 'scene', '场景'),\n ('pbl_drt_entity', 'pbl_domain_ref_type', 'entity', '实体')\nON DUPLICATE KEY UPDATE `v` = VALUES(`v`);\n\n-- ---------------------------------------------------------------------\n-- 3. 迁移兼容: 历史版本曾用列名 `ext`(json), 统一为设计权威名 ext_json(LONGTEXT)\n-- 存在旧列时搬迁数据后删除, 保证 models / sql / api 三处同名同型\n-- ---------------------------------------------------------------------\n-- (由部署脚本按 information_schema 判定后执行, 此处保留语句备查)\n-- SET @has_ext := (SELECT COUNT(*) FROM information_schema.COLUMNS\n-- WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'pbl_domain_ref' AND COLUMN_NAME = 'ext');\n-- SET @sql := IF(@has_ext > 0,\n-- 'UPDATE pbl_domain_ref SET ext_json = CAST(ext AS CHAR) WHERE ext_json IS NULL AND ext IS NOT NULL',\n-- 'SELECT 1');\n-- PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;\n-- SET @sql2 := IF(@has_ext > 0, 'ALTER TABLE pbl_domain_ref DROP COLUMN ext', 'SELECT 1');\n-- PREPARE s2 FROM @sql2; EXECUTE s2; DEALLOCATE PREPARE s2;\n\n-- ---------------------------------------------------------------------\n-- 4. 明确不做的事 (Q-OPEN-3 薄扩展边界)\n-- 不 ALTER world / scene / entity 三张基表; 不新增 world/scene/entity 三张 PBL 表;\n-- PBL 侧一律通过 pbl_domain_ref(ref_type + ref_id) 单向引用基表主键。\n-- ---------------------------------------------------------------------\n" : DDL 缺列 ref_code
|
||||
|
||||
----------------------------------------------------------------------
|
||||
Ran 81 tests in 0.049s
|
||||
|
||||
FAILED (failures=9, errors=2)
|
||||
Loading…
x
Reference in New Issue
Block a user