179 lines
5.4 KiB
Python
179 lines
5.4 KiB
Python
#!/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()
|