feat: batch update - gift card breakdown spec, backend APIs, miniprogram pages, ETL finance recharge, docs & migrations
This commit is contained in:
178
scripts/ops/e2e_test_rns1.py
Normal file
178
scripts/ops/e2e_test_rns1.py
Normal file
@@ -0,0 +1,178 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
RNS1.1 端到端验证脚本
|
||||
|
||||
启动后端后，用真实 JWT 对 4 个接口发 HTTP 请求，验证完整链路：
|
||||
JWT 认证 → 路由 → 服务层 → fdw_queries（直连 ETL 库）→ RLS 视图 → 返回数据
|
||||
|
||||
测试用户：
|
||||
user_id=8777, site_id=2790685415443269, assistant_id=2793532503855173
|
||||
（E2E测试助教，在 test_zqyy_app 中创建）
|
||||
|
||||
用法：
|
||||
1. 启动后端：cd apps/backend && uvicorn app.main:app --port 8000
|
||||
2. 运行脚本：python scripts/ops/e2e_test_rns1.py
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
# 确保项目根目录在 sys.path 中，以便导入 app 模块
|
||||
_root = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(_root / "apps" / "backend"))
|
||||
os.environ.setdefault("NEOZQYY_ROOT", str(_root))
|
||||
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv(_root / ".env", override=False)
|
||||
|
||||
import requests
|
||||
from app.auth.jwt import create_access_token
|
||||
|
||||
# ── 测试参数 ──
|
||||
USER_ID = 8777
|
||||
SITE_ID = 2790685415443269
|
||||
ASSISTANT_ID = 2793532503855173
|
||||
BASE_URL = os.environ.get("E2E_BASE_URL", "http://localhost:8000")
|
||||
|
||||
# salary_calc 数据存在的月份（2025-11 ~ 2026-02）
|
||||
TEST_YEAR = 2026
|
||||
TEST_MONTH = 2
|
||||
|
||||
|
||||
def _token() -> str:
|
||||
return create_access_token(user_id=USER_ID, site_id=SITE_ID)
|
||||
|
||||
|
||||
def _headers() -> dict:
|
||||
return {"Authorization": f"Bearer {_token()}"}
|
||||
|
||||
|
||||
def _print_result(name: str, resp: requests.Response):
|
||||
status = resp.status_code
|
||||
ok = "✅" if status == 200 else "❌"
|
||||
print(f"\n{ok} [{name}] HTTP {status}")
|
||||
if status == 200:
|
||||
data = resp.json()
|
||||
# 只打印关键字段，避免刷屏
|
||||
if isinstance(data, dict):
|
||||
for k, v in data.items():
|
||||
if isinstance(v, list):
|
||||
print(f" {k}: [{len(v)} items]")
|
||||
elif isinstance(v, dict):
|
||||
print(f" {k}: {{...}}")
|
||||
else:
|
||||
print(f" {k}: {v}")
|
||||
else:
|
||||
print(f" {data}")
|
||||
else:
|
||||
print(f" body: {resp.text[:500]}")
|
||||
|
||||
|
||||
|
||||
# ── TASK-1: 任务列表 ──
|
||||
def test_task_list():
|
||||
"""GET /api/xcx/tasks?status=pending&page=1"""
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/xcx/tasks",
|
||||
params={"status": "pending", "page": 1, "page_size": 5},
|
||||
headers=_headers(),
|
||||
timeout=15,
|
||||
)
|
||||
_print_result("TASK-1 任务列表", resp)
|
||||
return resp.status_code == 200
|
||||
|
||||
|
||||
# ── TASK-2: 任务详情（需要先从列表拿到一个 task_id）──
|
||||
def test_task_detail(task_id: int | None = None):
|
||||
"""GET /api/xcx/tasks/{task_id}"""
|
||||
if task_id is None:
|
||||
# 先拿列表，取第一个
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/xcx/tasks",
|
||||
params={"status": "pending", "page": 1, "page_size": 1},
|
||||
headers=_headers(),
|
||||
timeout=15,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
print(f"\n❌ [TASK-2 任务详情] 无法获取列表: HTTP {resp.status_code}")
|
||||
return False
|
||||
data = resp.json()
|
||||
tasks = data.get("tasks") or data.get("items") or []
|
||||
if not tasks:
|
||||
print("\n⚠️ [TASK-2 任务详情] 列表为空，跳过（测试用户可能没有待办任务）")
|
||||
return True # 不算失败
|
||||
task_id = tasks[0].get("id") or tasks[0].get("task_id")
|
||||
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/xcx/tasks/{task_id}",
|
||||
headers=_headers(),
|
||||
timeout=15,
|
||||
)
|
||||
_print_result("TASK-2 任务详情", resp)
|
||||
return resp.status_code == 200
|
||||
|
||||
|
||||
# ── PERF-1: 绩效概览 ──
|
||||
def test_performance_overview():
|
||||
"""GET /api/xcx/performance?year=2026&month=2"""
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/xcx/performance",
|
||||
params={"year": TEST_YEAR, "month": TEST_MONTH},
|
||||
headers=_headers(),
|
||||
timeout=15,
|
||||
)
|
||||
_print_result("PERF-1 绩效概览", resp)
|
||||
return resp.status_code == 200
|
||||
|
||||
|
||||
# ── PERF-2: 绩效明细 ──
|
||||
def test_performance_records():
|
||||
"""GET /api/xcx/performance/records?year=2026&month=2"""
|
||||
resp = requests.get(
|
||||
f"{BASE_URL}/api/xcx/performance/records",
|
||||
params={"year": TEST_YEAR, "month": TEST_MONTH, "page": 1, "page_size": 5},
|
||||
headers=_headers(),
|
||||
timeout=15,
|
||||
)
|
||||
_print_result("PERF-2 绩效明细", resp)
|
||||
return resp.status_code == 200
|
||||
|
||||
|
||||
# ── 主入口 ──
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print("RNS1.1 端到端验证")
|
||||
print(f"BASE_URL: {BASE_URL}")
|
||||
print(f"USER_ID: {USER_ID}, SITE_ID: {SITE_ID}")
|
||||
print(f"测试月份: {TEST_YEAR}-{TEST_MONTH:02d}")
|
||||
print("=" * 60)
|
||||
|
||||
# 先检查后端是否可达
|
||||
try:
|
||||
r = requests.get(f"{BASE_URL}/docs", timeout=5)
|
||||
print(f"\n后端可达: HTTP {r.status_code}")
|
||||
except requests.ConnectionError:
|
||||
print(f"\n❌ 无法连接 {BASE_URL}，请先启动后端：")
|
||||
print(" cd apps/backend && uvicorn app.main:app --port 8000")
|
||||
sys.exit(1)
|
||||
|
||||
results = {}
|
||||
results["TASK-1"] = test_task_list()
|
||||
results["TASK-2"] = test_task_detail()
|
||||
results["PERF-1"] = test_performance_overview()
|
||||
results["PERF-2"] = test_performance_records()
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
passed = sum(1 for v in results.values() if v)
|
||||
total = len(results)
|
||||
print(f"结果: {passed}/{total} 通过")
|
||||
for name, ok in results.items():
|
||||
print(f" {'✅' if ok else '❌'} {name}")
|
||||
print("=" * 60)
|
||||
|
||||
sys.exit(0 if passed == total else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user