# -*- coding: utf-8 -*- """第 3 批写接口端到端自测：新增 -> 编辑 -> 删除，逐项校验文案与状态。 覆盖 /store/store.role、/store/store.user、/store/store.address、/store/page。 所有中间数据都会在脚本末尾清理，避免污染库。 """ import json import requests BASE = "http://127.0.0.1:18080" TOKEN = "testsession123" S = requests.Session() S.trust_env = False H = {"Access-Token": TOKEN, "Content-Type": "application/json"} ok_count = 0 fail_count = 0 def call(path, payload=None, method="post", params=None): url = BASE + path if method == "get": r = S.get(url, headers=H, params=params or {}, timeout=60) else: r = S.post(url, headers=H, data=json.dumps(payload or {}), timeout=60) return r.json() def check(name, cond, detail=""): global ok_count, fail_count if cond: ok_count += 1 print(f" [OK] {name}") else: fail_count += 1 print(f" [FAIL] {name} {detail}") def expect_ok(name, resp, message=None): good = resp.get("status") == 200 and (message is None or resp.get("message") == message) check(name, good, f"-> {json.dumps(resp, ensure_ascii=False)[:160]}") def expect_fail(name, resp, message): good = resp.get("status") == 500 and resp.get("message") == message check(name, good, f"-> {json.dumps(resp, ensure_ascii=False)[:160]}") def main(): print("== 角色 CRUD") roles_before = call("/store/store.role/list", method="get")["data"]["list"] menus = [] def collect(nodes): for n in nodes: menus.append(n["menu_id"]) collect(n.get("children") or []) collect(call("/store/store.menu/list", method="get")["data"]["list"]) pick = menus[:5] expect_fail("新增角色-未选菜单", call("/store/store.role/add", {"form": {"role_name": "自测角色", "parent_id": 0, "sort": 99, "menus": []}}), "菜单权限不能为空，请重新选择") expect_ok("新增角色", call("/store/store.role/add", {"form": {"role_name": "自测角色", "parent_id": 0, "sort": 99, "menus": pick}}), "添加成功") roles_after = call("/store/store.role/list", method="get")["data"]["list"] created = [r for r in roles_after if r.get("role_name") == "自测角色"] check("角色已落库", len(created) == 1, f"got {len(created)}") role_id = created[0]["role_id"] if created else None check("角色菜单关联正确", created and sorted(created[0].get("menuIds") or []) == sorted(pick), f"{created[0].get('menuIds') if created else None}") expect_ok("编辑角色", call("/store/store.role/edit", {"roleId": role_id, "form": {"role_name": "自测角色改", "parent_id": 0, "sort": 98, "menus": pick[:3]}}), "更新成功") role_now = [r for r in call("/store/store.role/list", method="get")["data"]["list"] if r["role_id"] == role_id] check("角色编辑生效", role_now and role_now[0]["role_name"] == "自测角色改" and sorted(role_now[0].get("menuIds") or []) == sorted(pick[:3]), json.dumps(role_now, ensure_ascii=False)[:200] if role_now else "missing") print("== 管理员 CRUD") expect_fail("新增管理员-用户名重复", call("/store/store.user/add", {"form": { "user_name": "admin", "real_name": "x", "password": "123456", "password_confirm": "123456", "roles": [role_id]}}), "用户名已存在") if any(u.get("user_name") == "admin" for u in call("/store/store.user/list", method="get", params={"page": 1})["data"]["list"]["data"]) else None expect_fail("新增管理员-两次密码不一致", call("/store/store.user/add", {"form": { "user_name": "autotest_admin", "real_name": "自测", "password": "123456", "password_confirm": "654321", "roles": [role_id]}}), "确认密码不正确") expect_fail("新增管理员-未选角色", call("/store/store.user/add", {"form": { "user_name": "autotest_admin", "real_name": "自测", "password": "123456", "password_confirm": "123456", "roles": []}}), "请选择所属角色") expect_ok("新增管理员", call("/store/store.user/add", {"form": { "user_name": "AutoTest_Admin", "real_name": "自测账号", "password": "123456", "password_confirm": "123456", "roles": [role_id], "sort": 99}}), "添加成功") users = call("/store/store.user/list", method="get", params={"page": 1})["data"]["list"]["data"] created_user = [u for u in users if u["user_name"] == "autotest_admin"] check("用户名已转小写落库", len(created_user) == 1, str([u["user_name"] for u in users])) user_id = created_user[0]["store_user_id"] if created_user else None check("角色关联正确", created_user and created_user[0].get("roleIds") == [role_id], str(created_user[0].get("roleIds")) if created_user else "missing") # 新账号能登录，说明 Java 生成的 $2y$ 哈希与 PHP password_verify 同构 login = S.post(BASE + "/store/passport/login", headers={"Content-Type": "application/json"}, data=json.dumps({"username": "autotest_admin", "password": "123456"}), timeout=60).json() expect_ok("新管理员可登录", login) new_token = (login.get("data") or {}).get("token") expect_ok("编辑管理员", call("/store/store.user/edit", {"userId": user_id, "form": { "user_name": "autotest_admin", "real_name": "自测账号改", "password": "", "password_confirm": "", "roles": [role_id], "sort": 97}}), "更新成功") users = call("/store/store.user/list", method="get", params={"page": 1})["data"]["list"]["data"] u2 = [u for u in users if u["store_user_id"] == user_id] check("管理员编辑生效", u2 and u2[0]["real_name"] == "自测账号改" and u2[0]["sort"] == 97, json.dumps(u2, ensure_ascii=False)[:200]) if new_token: renew = S.post(BASE + "/store/store.user/renew", headers={"Access-Token": new_token, "Content-Type": "application/json"}, data=json.dumps({"form": {"user_name": "autotest_admin", "real_name": "自测账号二改", "password": "", "password_confirm": ""}}), timeout=60).json() expect_ok("当前管理员改资料", renew, "更新成功") print("== 地址 CRUD") expect_ok("新增地址", call("/store/store.address/add", {"form": { "type": 10, "name": "自测联系人", "phone": "13800000000", "cascader": [110000, 110100, 110101], "detail": "自测地址1号", "sort": 99}}), "添加成功") all_addr = call("/store/store.address/all", method="get")["data"]["list"] addr = [a for a in all_addr if a["name"] == "自测联系人"] check("地址已落库且省市区拆分正确", len(addr) == 1 and addr[0]["province_id"] == 110000 and addr[0]["city_id"] == 110100 and addr[0]["region_id"] == 110101, json.dumps(addr, ensure_ascii=False)[:240]) addr_id = addr[0]["address_id"] if addr else None check("地址级联/全称已组装", addr and addr[0].get("cascader") == [110000, 110100, 110101] and str(addr[0].get("full_address", "")).endswith("自测地址1号"), str(addr[0].get("full_address")) if addr else "missing") expect_ok("编辑地址", call("/store/store.address/edit", {"addressId": addr_id, "form": { "type": 20, "name": "自测联系人改", "phone": "13900000000", "cascader": [310000, 310100, 310101], "detail": "自测地址2号", "sort": 98}}), "更新成功") addr2 = [a for a in call("/store/store.address/all", method="get")["data"]["list"] if a["address_id"] == addr_id] check("地址编辑生效", addr2 and addr2[0]["name"] == "自测联系人改" and addr2[0]["type"] == 20 and addr2[0]["province_id"] == 310000, json.dumps(addr2, ensure_ascii=False)[:240]) filtered = call("/store/store.address/all", method="get", params={"type": 20})["data"]["list"] check("地址 type 过滤生效", all(a["type"] == 20 for a in filtered) and len(filtered) >= 1, f"{len(filtered)} 条") searched = call("/store/store.address/all", method="get", params={"search": "13900000000"})["data"]["list"] check("地址 search 过滤生效", any(a["address_id"] == addr_id for a in searched), f"{len(searched)} 条") print("== 页面 CRUD") page_id = None page_payload = {"form": {"page": {"params": {"name": "自测页面", "title": "自测标题", "shareTitle": "自测分享"}, "style": {"titleTextColor": "black", "titleBackgroundColor": "#ffffff"}, "name": "页面设置", "type": "page"}, "items": [{"name": "富文本", "type": "richText", "params": {"content": "

你好 & 世界

"}, "style": {"paddingTop": 0, "paddingLeft": 0, "background": "#ffffff"}}]}} expect_ok("新增页面", call("/store/page/add", page_payload), "添加成功") pages = call("/store/page/list", method="get", params={"page": 1})["data"]["list"]["data"] created_page = [p for p in pages if p["page_name"] == "自测页面"] check("页面已落库且类型为自定义页", len(created_page) == 1 and created_page[0]["page_type"] == 20, json.dumps(created_page, ensure_ascii=False)[:200]) page_id = created_page[0]["page_id"] if created_page else None detail = call("/store/page/detail", method="get", params={"pageId": page_id})["data"]["detail"] items = detail["page_data"]["items"] check("页面详情合并默认组件参数", items and items[0]["type"] == "richText" and items[0]["style"]["paddingTop"] == 0 and "name" in items[0], json.dumps(items, ensure_ascii=False)[:200]) check("页面详情 page.params 合并正常", detail["page_data"]["page"]["params"]["name"] == "自测页面", json.dumps(detail["page_data"]["page"], ensure_ascii=False)[:200]) expect_fail("删除首页被拒绝", call("/store/page/delete", {"pageId": 10001}), "默认首页不可以删除") page_payload["form"]["page"]["params"]["name"] = "自测页面改" expect_ok("编辑页面", call("/store/page/edit", {"pageId": page_id, "form": page_payload["form"]}), "更新成功") pages = call("/store/page/list", method="get", params={"page": 1})["data"]["list"]["data"] check("页面编辑生效", any(p["page_id"] == page_id and p["page_name"] == "自测页面改" for p in pages)) print("== 清理") if page_id: expect_ok("删除页面", call("/store/page/delete", {"pageId": page_id}), "删除成功") if addr_id: expect_ok("删除地址", call("/store/store.address/delete", {"addressId": addr_id}), "删除成功") if user_id: expect_ok("删除管理员", call("/store/store.user/delete", {"userId": user_id}), "删除成功") if new_token: S.post(BASE + "/store/passport/logout", headers={"Access-Token": new_token}, timeout=60) if role_id: expect_ok("删除角色", call("/store/store.role/delete", {"roleId": role_id}), "删除成功") print("== 删除后复核") pages = call("/store/page/list", method="get", params={"page": 1})["data"]["list"]["data"] check("页面已软删不可见", all(p["page_id"] != page_id for p in pages)) addr = call("/store/store.address/all", method="get")["data"]["list"] check("地址已软删不可见", all(a["address_id"] != addr_id for a in addr)) users = call("/store/store.user/list", method="get", params={"page": 1})["data"]["list"]["data"] check("管理员已软删不可见", all(u["store_user_id"] != user_id for u in users)) roles = call("/store/store.role/list", method="get")["data"]["list"] check("角色已删除", all(r["role_id"] != role_id for r in roles)) print(f"\n汇总：通过 {ok_count}，失败 {fail_count}") if __name__ == "__main__": main()