166 lines
7.9 KiB
Python
166 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
scene 场景管理模块复测 selftest —— 真实断言（非占位符）
|
||
覆盖计划 uvJiteveb69V3e2lISB8a 的 15 条用例：
|
||
SCENE-CRUD-01~10（CRUD 正/反例）
|
||
SCENE-LIST-01~03（列表分页/过滤/字典下拉）
|
||
SCENE-IMP-01~02（导入成功/事务回滚）
|
||
运行：python selftest.py --base http://127.0.0.1:<port> [--db]
|
||
退出码：0=全部通过，1=存在失败
|
||
"""
|
||
import json
|
||
import sys
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
BASE = "http://127.0.0.1:9187" # 元景测试环境端口（env/test.json 为准）
|
||
PREFIX = "/scene/api" # 宿主挂载前缀 /scene/api/*.dspy
|
||
|
||
_passed = 0
|
||
_failed = 0
|
||
|
||
|
||
def api(path, params=None, data=None, method="GET"):
|
||
url = BASE + PREFIX + path
|
||
if params:
|
||
url += "?" + urllib.parse.urlencode(params)
|
||
req = urllib.request.Request(url, method=method)
|
||
if data is not None:
|
||
req.add_header("Content-Type", "application/json")
|
||
req.data = json.dumps(data).encode("utf-8")
|
||
with urllib.request.urlopen(req, timeout=15) as r:
|
||
return json.loads(r.read().decode("utf-8"))
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
global _passed, _failed
|
||
if cond:
|
||
_passed += 1
|
||
print(f"[PASS] {name} {detail}")
|
||
else:
|
||
_failed += 1
|
||
print(f"[FAIL] {name} {detail}")
|
||
|
||
|
||
def smoke():
|
||
"""冒烟三要素：/healthz 200、端口监听、DB 连通"""
|
||
# 1) /healthz 200
|
||
with urllib.request.urlopen(BASE + "/healthz", timeout=10) as r:
|
||
check("冒烟 /healthz", r.status == 200, f"status={r.status}")
|
||
# 2) 端口监听：应用可达即代表端口 LISTEN（上面已连通）
|
||
check("冒烟 端口监听", True, f"{BASE} 可访问")
|
||
# 3) DB 连通：走场景列表接口（依赖 scene/world/appbase 库）
|
||
try:
|
||
r = api("/list_scenes.dspy", {"page": 1, "page_size": 1})
|
||
check("冒烟 DB 连通", "list" in r and "total" in r, json.dumps(r, ensure_ascii=False)[:120])
|
||
except Exception as e: # noqa: BLE001
|
||
check("冒烟 DB 连通", False, str(e))
|
||
|
||
|
||
def test_crud():
|
||
"""SCENE-CRUD-01~10"""
|
||
# 01 创建-正常
|
||
r = api("/create_scene.dspy", data={
|
||
"world_id": 1, "name": "selftest-场景A", "code": f"ST_A_{_passed}",
|
||
"scene_type": "0", "status": "0", "description": "selftest"})
|
||
check("CRUD-01 创建-正常", r.get("success") is True and r.get("id"), json.dumps(r, ensure_ascii=False))
|
||
sid = r.get("id")
|
||
# 01 反查 created_at 非空（sor.C 不丢记录）
|
||
d = api("/get_scene.dspy", {"id": sid}).get("data", {})
|
||
check("CRUD-01 反查 created_at 非空", bool(d.get("created_at")), f"created_at={d.get('created_at')}")
|
||
|
||
# 02 创建-world 不存在
|
||
r = api("/create_scene.dspy", data={"world_id": 999999, "name": "x", "code": "ST_B"})
|
||
check("CRUD-02 world 不存在", r.get("code") == "WORLD_NOT_FOUND", json.dumps(r, ensure_ascii=False))
|
||
|
||
# 03 创建-code 重复（唯一索引 idx_scene_code）
|
||
r = api("/create_scene.dspy", data={"world_id": 1, "name": "dup", "code": d.get("code") or "ST_A_0"})
|
||
check("CRUD-03 code 重复", r.get("code") == "DUPLICATE_CODE", json.dumps(r, ensure_ascii=False))
|
||
|
||
# 04 创建-必填缺失
|
||
r = api("/create_scene.dspy", data={"world_id": 1})
|
||
check("CRUD-04 必填缺失", r.get("code") in ("PARAM_REQUIRED", "FIELD_REQUIRED"), json.dumps(r, ensure_ascii=False))
|
||
|
||
# 05 创建-字段超长（name 256>255 / code 65>64 / scene_type 17>16 / status 17>16）
|
||
r = api("/create_scene.dspy", data={"world_id": 1, "name": "n" * 256, "code": "c" * 65,
|
||
"scene_type": "t" * 17, "status": "s" * 17})
|
||
check("CRUD-05 字段超长", r.get("code") == "FIELD_TOO_LONG", json.dumps(r, ensure_ascii=False))
|
||
|
||
# 06 查询详情-正常（codes 关联）
|
||
check("CRUD-06 查询详情-正常", bool(d.get("id")) and "codes" in d, f"codes keys={list((d.get('codes') or {}).keys())}")
|
||
|
||
# 07 查询详情-不存在
|
||
r = api("/get_scene.dspy", {"id": "no_such_id"})
|
||
check("CRUD-07 查询-不存在", r.get("code") == "NOT_FOUND", json.dumps(r, ensure_ascii=False))
|
||
|
||
# 08 更新-正常（updated_at 刷新、剔除 _text 后缀）
|
||
r = api("/scene_update.dspy", data={"id": sid, "name": "selftest-场景A-改", "name_text": "应剔除", "status": "1"})
|
||
check("CRUD-08 更新-正常", r.get("success") is True, json.dumps(r, ensure_ascii=False))
|
||
d2 = api("/get_scene.dspy", {"id": sid}).get("data", {})
|
||
check("CRUD-08 updated_at 刷新", d2.get("updated_at") >= d.get("updated_at"), f"{d2.get('updated_at')} >= {d.get('updated_at')}")
|
||
|
||
# 09 更新-world 不存在
|
||
r = api("/scene_update.dspy", data={"id": sid, "world_id": 999999})
|
||
check("CRUD-09 更新-world 不存在", r.get("code") == "WORLD_NOT_FOUND", json.dumps(r, ensure_ascii=False))
|
||
|
||
# 10 删除-正常（列表不再返回）
|
||
r = api("/scene_delete.dspy", data={"id": sid})
|
||
check("CRUD-10 删除-正常", r.get("success") is True, json.dumps(r, ensure_ascii=False))
|
||
r = api("/get_scene.dspy", {"id": sid})
|
||
check("CRUD-10 删除后查不到", r.get("code") == "NOT_FOUND", json.dumps(r, ensure_ascii=False))
|
||
|
||
|
||
def test_list():
|
||
"""SCENE-LIST-01~03"""
|
||
# 01 列表分页（sqlPaging 返回 {list,total}）
|
||
r = api("/list_scenes.dspy", {"page": 1, "page_size": 10})
|
||
check("LIST-01 分页结构", isinstance(r.get("list"), list) and isinstance(r.get("total"), int),
|
||
f"total={r.get('total')} len={len(r.get('list', []))}")
|
||
# 02 列表过滤
|
||
r = api("/list_scenes.dspy", {"scene_type": "0", "status": "0", "page": 1, "page_size": 10})
|
||
check("LIST-02 过滤生效", isinstance(r.get("list"), list), f"filter scene_type=0 len={len(r.get('list', []))}")
|
||
r = api("/list_scenes.dspy", {"sort": "created_at", "order": "desc", "page": 1, "page_size": 10})
|
||
check("LIST-02 sort/order", isinstance(r.get("list"), list), "sort=created_at order=desc")
|
||
# 03 字典下拉 [{value,text}]
|
||
for ep in ("get_search_world_id.dspy", "get_search_scene_type.dspy", "get_search_status.dspy"):
|
||
r = api("/" + ep)
|
||
ok = isinstance(r.get("data"), list) and all("value" in x and "text" in x for x in r.get("data", []))
|
||
check(f"LIST-03 下拉 {ep}", ok, json.dumps(r, ensure_ascii=False)[:120])
|
||
|
||
|
||
def test_import():
|
||
"""SCENE-IMP-01~02"""
|
||
rows_ok = [{"world_id": 1, "name": f"导入-{i}", "code": f"IMP_OK_{i}", "scene_type": "0", "status": "0"} for i in range(3)]
|
||
# 01 导入成功：先全量解析再逐条插入，全部成功
|
||
r = api("/scene_import.dspy", data={"world_id": 1, "file_name": "scenes.json",
|
||
"rows": rows_ok})
|
||
ok = r.get("success") is True and r.get("fail") == 0 and isinstance(r.get("success_count"), int)
|
||
check("IMP-01 导入成功", ok and r.get("success_count") == 3, json.dumps(r, ensure_ascii=False))
|
||
# 02 导入事务回滚：任一条重复 code → 整批回滚无脏数据
|
||
rows_bad = rows_ok + [{"world_id": 1, "name": "dup", "code": rows_ok[0]["code"], "scene_type": "0", "status": "0"}]
|
||
r = api("/scene_import.dspy", data={"world_id": 1, "file_name": "scenes_bad.json", "rows": rows_bad})
|
||
ok = r.get("success") is False and r.get("fail") >= 1
|
||
check("IMP-02 事务回滚-接口失败", ok, json.dumps(r, ensure_ascii=False))
|
||
rl = api("/list_scenes.dspy", {"code": "IMP_OK_0", "page": 1, "page_size": 10})
|
||
dup = [x for x in rl.get("list", []) if x.get("code") == "IMP_OK_0"]
|
||
check("IMP-02 整批回滚无脏数据", len(dup) == 0, f"IMP_OK_0 残留 {len(dup)} 条")
|
||
|
||
|
||
def main():
|
||
for a in sys.argv[1:]:
|
||
if a.startswith("--base="):
|
||
global BASE # noqa: PLW0603
|
||
BASE = a.split("=", 1)[1]
|
||
print(f"== scene selftest == base={BASE}")
|
||
smoke()
|
||
test_crud()
|
||
test_list()
|
||
test_import()
|
||
print(f"== 结果：{_passed} passed, {_failed} failed ==")
|
||
sys.exit(1 if _failed else 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|