refactor: 清理未使用IPC命令、修正point_service注释与扣费逻辑、修复camelToSnake正则、优化vidu import
- 删除8个未使用IPC命令,保留validate_media_path - file.rs返回类型优化为ApiResponse<()> - point_service.consume()注释与签名一致 - VideoGeneration改为拼接成功后扣费 - 添加漏扣费风险注释 - 删除过时测试文件 - 修复camelToSnake连续大写字母问题 - vidu.py import移至模块顶层 Refs: P1-1~P1-6 技术债务清理
This commit is contained in:
@@ -1,153 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
验证双 Token + 单设备踢人功能
|
||||
==============================
|
||||
|
||||
用法:
|
||||
cd python-api
|
||||
python -m scripts.test_auth --mobile 13800138000
|
||||
|
||||
流程:
|
||||
1. 发送验证码
|
||||
2. 设备 A 登录(获取 Token A)
|
||||
3. 用 Token A 调用 /me 验证 Access Token 有效
|
||||
4. 用 Token A 的 refresh_token 刷新(验证 Token 轮换)
|
||||
5. 设备 B 登录(同一手机号,获取 Token B)
|
||||
6. 验证 Token A 已失效(被踢)
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import sys
|
||||
import time
|
||||
|
||||
import httpx
|
||||
|
||||
# 测试环境 API 地址
|
||||
BASE_URL = "https://dev.tapi.meijiaka.cn/api/v1"
|
||||
|
||||
|
||||
async def send_code(mobile: str) -> None:
|
||||
"""发送验证码(开发环境验证码会打印到后端日志)"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(f"{BASE_URL}/auth/send-code", json={"mobile": mobile})
|
||||
print(f"[1] 发送验证码: {r.status_code} {r.text}")
|
||||
r.raise_for_status()
|
||||
|
||||
|
||||
async def login(mobile: str, code: str, device_id: str) -> dict:
|
||||
"""登录,返回 {access_token, refresh_token, user}"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{BASE_URL}/auth/login",
|
||||
json={
|
||||
"mobile": mobile,
|
||||
"code": code,
|
||||
"device_id": device_id,
|
||||
"device_name": f"测试设备-{device_id[-4:]}",
|
||||
"os_info": "test-script",
|
||||
"app_version": "0.1.0",
|
||||
},
|
||||
)
|
||||
print(f"[2] 登录 ({device_id}): {r.status_code}")
|
||||
r.raise_for_status()
|
||||
return r.json()["data"]
|
||||
|
||||
|
||||
async def call_me(access_token: str) -> dict:
|
||||
"""用 Access Token 调用 /me"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.get(
|
||||
f"{BASE_URL}/auth/me",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
)
|
||||
print(f"[3] /me: {r.status_code} {r.text[:200]}")
|
||||
r.raise_for_status()
|
||||
return r.json()["data"]
|
||||
|
||||
|
||||
async def refresh_token(refresh_token: str) -> dict:
|
||||
"""用 Refresh Token 换取新的 Token 对"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{BASE_URL}/auth/refresh",
|
||||
json={"refresh_token": refresh_token},
|
||||
)
|
||||
print(f"[4] 刷新 Token: {r.status_code}")
|
||||
r.raise_for_status()
|
||||
return r.json()["data"]
|
||||
|
||||
|
||||
async def verify_kicked(access_token: str) -> bool:
|
||||
"""验证旧 Access Token 是否已失效(被踢)"""
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.get(
|
||||
f"{BASE_URL}/auth/me",
|
||||
headers={"Authorization": f"Bearer {access_token}"},
|
||||
)
|
||||
print(f"[6] 旧 Token 调用 /me: {r.status_code} {r.text[:200]}")
|
||||
return r.status_code == 401
|
||||
|
||||
|
||||
async def main():
|
||||
parser = argparse.ArgumentParser(description="验证双 Token + 踢人功能")
|
||||
parser.add_argument("--mobile", required=True, help="手机号")
|
||||
parser.add_argument("--code", default="123456", help="验证码(默认 123456,需与日志一致)")
|
||||
args = parser.parse_args()
|
||||
|
||||
mobile = args.mobile
|
||||
code = args.code
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"测试手机号: {mobile}")
|
||||
print(f"验证码: {code}")
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
# 1. 发送验证码
|
||||
await send_code(mobile)
|
||||
print("⚠️ 请确认后端日志中的验证码与 --code 一致\n")
|
||||
|
||||
# 2. 设备 A 登录
|
||||
device_a = "device-a-test"
|
||||
token_a = await login(mobile, code, device_a)
|
||||
print(f" Access Token A: {token_a['access_token'][:40]}...")
|
||||
print(f" Refresh Token A: {token_a['refresh_token'][:40]}...")
|
||||
print()
|
||||
|
||||
# 3. 用 Token A 调用 /me
|
||||
me = await call_me(token_a["access_token"])
|
||||
print(f" 用户信息: {me['nickname']} ({me['mobile']})\n")
|
||||
|
||||
# 4. 用 Refresh Token A 刷新(验证 Token 轮换)
|
||||
token_a2 = await refresh_token(token_a["refresh_token"])
|
||||
print(f" New Access Token: {token_a2['access_token'][:40]}...")
|
||||
print(f" New Refresh Token: {token_a2['refresh_token'][:40]}...")
|
||||
print()
|
||||
|
||||
# 5. 设备 B 登录(同一手机号,触发踢人)
|
||||
device_b = "device-b-test"
|
||||
token_b = await login(mobile, code, device_b)
|
||||
print(f" Access Token B: {token_b['access_token'][:40]}...")
|
||||
print()
|
||||
|
||||
# 6. 验证 Token A(旧的)已失效
|
||||
is_kicked = await verify_kicked(token_a["access_token"])
|
||||
if is_kicked:
|
||||
print("✅ 旧 Token A 已失效(被踢)—— 单设备登录生效\n")
|
||||
else:
|
||||
print("❌ 旧 Token A 仍然有效 —— 踢人逻辑有问题\n")
|
||||
|
||||
# 7. 验证 Token A2(刷新后的)也应该失效(因为设备 B 登录会覆盖设备记录)
|
||||
is_kicked2 = await verify_kicked(token_a2["access_token"])
|
||||
if is_kicked2:
|
||||
print("✅ 刷新后的 Token A2 也已失效\n")
|
||||
else:
|
||||
print("❌ 刷新后的 Token A2 仍然有效\n")
|
||||
|
||||
print(f"{'='*60}")
|
||||
print("验证完成")
|
||||
print(f"{'='*60}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user