shop-java/tools/test_batch3_writes.py

# -*- 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": "<p>你好 &amp; 世界</p>"},
                                        "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()