From ecef031551becb15149444c2ed5f8c001f4d9213 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Sat, 29 Aug 2026 15:38:22 +0800 Subject: [PATCH] =?UTF-8?q?approve:=20=E5=A4=8D=E6=B5=8B=E5=B7=B2=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20Bug=EF=BC=88=E5=85=83=E6=99=AF=E9=A1=B9=E7=9B=AE-?= =?UTF-8?q?=E5=88=9D=E5=A7=8B=E8=BF=AD=E4=BB=A3=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/selftest.py | 165 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 scripts/selftest.py diff --git a/scripts/selftest.py b/scripts/selftest.py new file mode 100644 index 0000000..b07f7a1 --- /dev/null +++ b/scripts/selftest.py @@ -0,0 +1,165 @@ +#!/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: [--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()