baqing-shop/e2e-test/test_admin_flow.py

# ============================================================
# 平台管理员端 — 全量端到端测试
# 覆盖 20 个 Controller，约 70 个 API 端点
# ============================================================
import pytest
import json
from config import KEEP_TEST_DATA
from api_client import ApiClient


# ============================================================
# 平台端 — 账号管理 (AccountManageController)
# ============================================================
class TestAdminAccountManage:

    def test_01_account_list(self, client, admin_token):
        """平台端账号列表查询"""
        resp = client.get("/agri/accountManage/list")
        client.assert_ok("查询账号列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 账号列表 total={data.get('total','?')}")

    def test_02_account_detail(self, client, admin_token):
        """平台端账号详情"""
        # 从列表取第一个
        resp = client.get("/agri/accountManage/list", params={"pageSize": 1, "pageNum": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无账号数据")
        uid = rows[0].get("userId")
        resp = client.get(f"/agri/accountManage/{uid}")
        client.assert_ok("查询账号详情失败")
        print(f"  [OK] 账号详情 userId={uid}")


# ============================================================
# 平台端 — 管理员管理 (PlatformAdminController)
# ============================================================
class TestAdminPlatformAdmin:

    def test_03_admin_list(self, client, admin_token):
        """平台管理员列表"""
        resp = client.get("/agri/platformAdmin/list")
        client.assert_ok("查询管理员列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 管理员列表 total={data.get('total','?')}")

    def test_04_admin_role_options(self, client, admin_token):
        """平台管理员角色选项"""
        resp = client.get("/agri/platformAdmin/roleOptions")
        client.assert_ok("查询角色选项失败")
        roles = client.extract_data() or []
        print(f"  [OK] 角色选项 count={len(roles)}")

    def test_05_admin_detail(self, client, admin_token):
        """平台管理员详情"""
        resp = client.get("/agri/platformAdmin/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无管理员数据")
        uid = rows[0].get("userId")
        resp = client.get(f"/agri/platformAdmin/{uid}")
        client.assert_ok("查询管理员详情失败")
        print(f"  [OK] 管理员详情 userId={uid}")


# ============================================================
# 平台端 — 会员管理 (MemberController)
# ============================================================
class TestAdminMember:

    def test_06_member_list(self, client, admin_token):
        """平台端会员列表"""
        resp = client.get("/agri/member/list")
        client.assert_ok("查询会员列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 会员列表 total={data.get('total','?')}")

    def test_07_member_levels(self, client, admin_token):
        """平台端会员等级列表"""
        resp = client.get("/agri/member/levels")
        client.assert_ok("查询会员等级失败")
        data = client.extract_data() or []
        if isinstance(data, list):
            print(f"  [OK] 会员等级 count={len(data)}")
        else:
            print(f"  [OK] 会员等级正常")

    def test_08_member_detail(self, client, admin_token):
        """平台端会员详情"""
        resp = client.get("/agri/member/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无会员数据")
        mid = rows[0].get("memberId")
        resp = client.get(f"/agri/member/{mid}")
        client.assert_ok("查询会员详情失败")
        print(f"  [OK] 会员详情 memberId={mid}")


# ============================================================
# 平台端 — 商家管理 (MerchantController)
# ============================================================
class TestAdminMerchant:

    def test_09_merchant_list(self, client, admin_token):
        """平台端商家列表"""
        resp = client.get("/agri/merchant/list")
        client.assert_ok("查询商家列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 商家列表 total={data.get('total','?')}")

    def test_10_merchant_detail(self, client, admin_token):
        """平台端商家详情"""
        resp = client.get("/agri/merchant/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无商家数据")
        mid = rows[0].get("merchantId")
        resp = client.get(f"/agri/merchant/{mid}")
        client.assert_ok("查询商家详情失败")
        print(f"  [OK] 商家详情 merchantId={mid}")


# ============================================================
# 平台端 — 入驻审核 (MerchantEntryApplyController)
# ============================================================
class TestAdminEntryAudit:

    def test_11_entry_apply_list(self, client, admin_token):
        """入驻申请列表"""
        resp = client.get("/agri/merchantEntryApply/list")
        client.assert_ok("查询入驻申请失败")
        data = client.extract_data() or {}
        total = data.get("total", 0) if isinstance(data, dict) else 0
        print(f"  [OK] 入驻申请 total={total}")

    def test_12_entry_pending_count(self, client, admin_token):
        """待审核入驻数量"""
        resp = client.get("/agri/merchantEntryApply/pendingCount")
        data = client.assert_ok("查询待审核入驻数量失败")
        cnt = (data.get("data") or {}).get("pendingCount", 0)
        print(f"  [OK] 待审核入驻 count={cnt}")

    def test_13_entry_apply_detail(self, client, admin_token):
        """入驻申请详情"""
        resp = client.get("/agri/merchantEntryApply/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无入驻申请")
        aid = rows[0].get("applyId")
        resp = client.get(f"/agri/merchantEntryApply/{aid}")
        client.assert_ok("查询入驻申请详情失败")
        print(f"  [OK] 入驻申请详情 applyId={aid}")

    def test_14_entry_approve_flow(self, client, admin_token):
        """入驻审核通过流程（需有待审核数据）"""
        resp = client.get("/agri/merchantEntryApply/list", params={
            "pageSize": 1, "pageNum": 1
        })
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无入驻申请")

        apply = rows[0]
        aid = apply.get("applyId")
        status = apply.get("applyStatus", "")
        print(f"  [INFO] 入驻申请 applyId={aid} status={status}")

        if status == "PENDING":
            resp = client.put(f"/agri/merchantEntryApply/approve/{aid}", json={})
            client.assert_ok("入驻审核通过失败")
            print(f"  [OK] 入驻申请 {aid} 审核通过")
        elif status == "APPROVED":
            resp = client.put(f"/agri/merchantEntryApply/completePublicity/{aid}")
            client.assert_ok("完成公示失败")
            print(f"  [OK] 入驻申请 {aid} 完成公示")
        else:
            print(f"  [SKIP] 当前状态 {status}，跳过")


# ============================================================
# 平台端 — 商品管理 (GoodsController + GoodsAuditController)
# ============================================================
class TestAdminGoods:

    def test_15_goods_list(self, client, admin_token):
        """平台端商品列表"""
        resp = client.get("/agri/goods/list")
        client.assert_ok("查询商品列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 商品列表 total={data.get('total','?')}")

    def test_16_goods_pending_count(self, client, admin_token):
        """待审核商品数量"""
        resp = client.get("/agri/goods/pendingCount")
        data = client.assert_ok("查询待审核商品数量失败")
        cnt = (data.get("data") or {}).get("pendingCount", 0)
        print(f"  [OK] 待审核商品 count={cnt}")

    def test_17_goods_detail(self, client, admin_token):
        """平台端商品详情"""
        resp = client.get("/agri/goods/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            resp = client.get("/agri/goodsAudit/list", params={"pageSize": 1})
            data = client.assert_ok()
            rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无商品数据")
        gid = rows[0].get("goodsId")
        resp = client.get(f"/agri/goods/{gid}")
        client.assert_ok("查询商品详情失败")
        print(f"  [OK] 商品详情 goodsId={gid}")

    def test_18_goods_audit_list(self, client, admin_token):
        """商品审核列表"""
        resp = client.get("/agri/goodsAudit/list")
        client.assert_ok("查询商品审核列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 商品审核列表 total={data.get('total','?')}")

    def test_19_goods_audit_pending_count(self, client, admin_token):
        """商品审核待审数量"""
        resp = client.get("/agri/goodsAudit/pendingCount")
        data = client.assert_ok("查询审核待审数量失败")
        cnt = (data.get("data") or {}).get("pendingCount", 0)
        print(f"  [OK] 审核待审商品 count={cnt}")

    def test_20_goods_audit_detail(self, client, admin_token):
        """商品审核详情"""
        resp = client.get("/agri/goodsAudit/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无待审商品")
        gid = rows[0].get("goodsId")
        resp = client.get(f"/agri/goodsAudit/{gid}")
        client.assert_ok("查询审核商品详情失败")
        print(f"  [OK] 审核商品详情 goodsId={gid}")


# ============================================================
# 平台端 — 分类管理 (PlatformCategoryController)
# ============================================================
class TestAdminCategory:

    def test_21_category_list(self, client, admin_token):
        """平台分类列表"""
        resp = client.get("/agri/category/list")
        client.assert_ok("查询分类列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 分类列表 total={data.get('total','?')}")

    def test_22_category_tree(self, client, admin_token):
        """平台分类树"""
        resp = client.get("/agri/category/tree")
        client.assert_ok("查询分类树失败")
        tree = client.extract_data() or []
        print(f"  [OK] 分类树 nodes={len(tree)}")

    def test_23_category_detail(self, client, admin_token):
        """平台分类详情"""
        resp = client.get("/agri/category/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无分类数据")
        cid = rows[0].get("categoryId")
        resp = client.get(f"/agri/category/{cid}")
        client.assert_ok("查询分类详情失败")
        print(f"  [OK] 分类详情 categoryId={cid}")


# ============================================================
# 平台端 — Banner管理 (BannerController)
# ============================================================
class TestAdminBanner:

    def test_24_banner_list(self, client, admin_token):
        """Banner列表"""
        resp = client.get("/agri/banner/list")
        client.assert_ok("查询Banner列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] Banner列表 total={data.get('total','?')}")

    def test_25_banner_detail(self, client, admin_token):
        """Banner详情"""
        resp = client.get("/agri/banner/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无Banner数据")
        bid = rows[0].get("bannerId")
        resp = client.get(f"/agri/banner/{bid}")
        client.assert_ok("查询Banner详情失败")
        print(f"  [OK] Banner详情 bannerId={bid}")


# ============================================================
# 平台端 — 商城设置 (MallSettingController)
# ============================================================
class TestAdminMallSetting:

    def test_26_mall_setting_get(self, client, admin_token):
        """商城设置查询"""
        resp = client.get("/agri/mallSetting")
        client.assert_ok("查询商城设置失败")
        setting = client.extract_data() or {}
        print(f"  [OK] 商城设置: {json.dumps(setting, ensure_ascii=False)[:100]}")


# ============================================================
# 平台端 — 服务协议 (MallServiceAgreementController)
# ============================================================
class TestAdminAgreement:

    def test_27_service_agreement_get(self, client, admin_token):
        """商城服务协议查询"""
        resp = client.get("/agri/mallServiceAgreement")
        client.assert_ok("查询服务协议失败")
        agmt = client.extract_data() or {}
        print(f"  [OK] 服务协议: {str(agmt.get('agreementType',''))[:80]}")

    def test_28_entry_agreement_get(self, client, admin_token):
        """入驻协议查询"""
        resp = client.get("/agri/merchantEntryAgreement")
        client.assert_ok("查询入驻协议失败")
        agmt = client.extract_data() or {}
        print(f"  [OK] 入驻协议: {str(agmt.get('agreementType',''))[:80]}")
        client._test_has_agreement = True


# ============================================================
# 平台端 — 地区管理 (RegionController)
# ============================================================
class TestAdminRegion:

    def test_29_region_tree(self, client, admin_token):
        """地区树查询"""
        resp = client.get("/agri/region/tree")
        client.assert_ok("查询地区树失败")
        tree = client.extract_data() or []
        print(f"  [OK] 地区树 nodes={len(tree) if isinstance(tree, list) else 'ok'}")


# ============================================================
# 平台端 — 店铺管理 (ShopController)
# ============================================================
class TestAdminShop:

    def test_30_shop_list(self, client, admin_token):
        """平台端店铺列表"""
        resp = client.get("/agri/shop/list")
        client.assert_ok("查询店铺列表失败")
        data = client.extract_data() or {}
        total = data.get("total", 0) if isinstance(data, dict) else 0
        print(f"  [OK] 店铺列表 total={total}")

    def test_31_shop_detail(self, client, admin_token):
        """平台端店铺详情"""
        resp = client.get("/agri/shop/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无店铺数据")
        sid = rows[0].get("shopId")
        resp = client.get(f"/agri/shop/{sid}")
        client.assert_ok("查询店铺详情失败")
        print(f"  [OK] 店铺详情 shopId={sid}")


# ============================================================
# 平台端 — 商品服务 (GoodsServiceController)
# ============================================================
class TestAdminGoodsService:

    def test_32_goods_service_list(self, client, admin_token):
        """商品服务列表"""
        resp = client.get("/agri/goodsService/list")
        client.assert_ok("查询商品服务列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 商品服务列表 total={data.get('total','?') if isinstance(data, dict) else 'ok'}")

    def test_33_goods_service_detail(self, client, admin_token):
        """商品服务详情"""
        resp = client.get("/agri/goodsService/list", params={"pageSize": 1})
        data = client.assert_ok()
        # 可能返回 list 或 {rows, total}
        rows = []
        if isinstance(data.get("data"), dict):
            rows = data["data"].get("rows", [])
        elif isinstance(data.get("data"), list):
            rows = data["data"]
        if not rows:
            pytest.skip("无商品服务数据")
        sid = rows[0].get("serviceId")
        resp = client.get(f"/agri/goodsService/{sid}")
        client.assert_ok("查询商品服务详情失败")
        print(f"  [OK] 商品服务详情 serviceId={sid}")


# ============================================================
# 平台端 — 财务管理 (FundOverview / WithdrawAudit / WithdrawSummary)
# ============================================================
class TestAdminFinance:

    def test_34_fund_summary(self, client, admin_token):
        """资金概览"""
        resp = client.get("/agri/finance/fundOverview/summary")
        client.assert_ok("查询资金概览失败")
        summary = client.extract_data() or {}
        print(f"  [OK] 资金概览: {json.dumps(summary, ensure_ascii=False)[:120]}")

    def test_35_fund_shops(self, client, admin_token):
        """资金概览-店铺列表"""
        resp = client.get("/agri/finance/fundOverview/shops")
        client.assert_ok("查询资金店铺列表失败")
        data = client.extract_data() or {}
        print(f"  [OK] 资金店铺列表: {json.dumps(data, ensure_ascii=False)[:120]}")

    def test_36_fund_details(self, client, admin_token):
        """资金概览-明细"""
        resp = client.get("/agri/finance/fundOverview/details", params={"pageSize": 1})
        client.assert_ok("查询资金明细失败")
        data = client.extract_data() or {}
        print(f"  [OK] 资金明细正常")

    def test_37_withdraw_audit_list(self, client, admin_token):
        """提现审核列表"""
        resp = client.get("/agri/finance/withdrawAudit/list")
        client.assert_ok("查询提现列表失败")
        data = client.extract_data() or {}
        total = data.get("total", 0) if isinstance(data, dict) else 0
        print(f"  [OK] 提现审核列表 total={total}")

    def test_38_withdraw_pending_count(self, client, admin_token):
        """提现待审数量"""
        resp = client.get("/agri/finance/withdrawAudit/pendingCount")
        client.assert_ok("查询提现待审数量失败")
        cnt = (client.extract_data() or {}).get("pendingCount", 0)
        print(f"  [OK] 提现待审 count={cnt}")

    def test_39_withdraw_summary_shops(self, client, admin_token):
        """提现汇总-店铺"""
        resp = client.get("/agri/finance/withdrawSummary/shops")
        client.assert_ok("查询提现汇总店铺失败")
        data = client.extract_data() or {}
        print(f"  [OK] 提现汇总-店铺正常")

    def test_40_withdraw_summary_details(self, client, admin_token):
        """提现汇总-明细"""
        resp = client.get("/agri/finance/withdrawSummary/details", params={"pageSize": 1})
        client.assert_ok("查询提现汇总明细失败")
        data = client.extract_data() or {}
        print(f"  [OK] 提现汇总-明细正常")


# ============================================================
# 平台端 — 微信支付配置 (PlatformShopWechatPayController)
# ============================================================
class TestAdminWechatPay:

    def test_41_wechat_pay_list(self, client, admin_token):
        """微信支付配置列表"""
        resp = client.get("/agri/shop/wechat-pay/list")
        client.assert_ok("查询支付配置列表失败")
        data = client.extract_data() or {}
        total = data.get("total", 0) if isinstance(data, dict) else 0
        print(f"  [OK] 支付配置列表 total={total}")

    def test_42_wechat_pay_pending_count(self, client, admin_token):
        """微信支付待审数量"""
        resp = client.get("/agri/shop/wechat-pay/pendingCount")
        client.assert_ok("查询支付待审数量失败")
        cnt = (client.extract_data() or {}).get("pendingCount", 0)
        print(f"  [OK] 支付待审 count={cnt}")

    def test_43_wechat_pay_detail(self, client, admin_token):
        """微信支付配置详情"""
        resp = client.get("/agri/shop/wechat-pay/list", params={"pageSize": 1})
        data = client.assert_ok()
        rows = (data.get("data") or {}).get("rows") or []
        if not rows:
            pytest.skip("无支付配置数据")
        sid = rows[0].get("shopId")
        resp = client.get(f"/agri/shop/wechat-pay/{sid}")
        client.assert_ok("查询支付配置详情失败")
        print(f"  [OK] 支付配置详情 shopId={sid}")


# ============================================================
# 平台端 — 全局设置 (ShopGlobalSettingController)
# ============================================================
class TestAdminGlobalSetting:

    def test_44_global_setting_get(self, client, admin_token):
        """全局设置查询"""
        resp = client.get("/agri/shopSetting")
        client.assert_ok("查询全局设置失败")
        setting = client.extract_data() or {}
        print(f"  [OK] 全局设置: {json.dumps(setting, ensure_ascii=False)[:100]}")