wkc
626f7d566b
- 添加缺失的认证参数：appId, appSecretCode, role - 修复 analysisType 和 departmentCode 参数 - 将所有接口改为使用 Form 参数（form-data 格式） - 更新服务层支持字典参数 - 更新所有测试代码 - 所有测试通过（7/7）
277 lines
8.4 KiB
Python
277 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
接口参数修复验证脚本
|
|
用于验证 GetToken 接口参数修复是否成功
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import sys
|
|
from typing import Dict, Any
|
|
|
|
# 配置
|
|
BASE_URL = "http://localhost:8000"
|
|
TIMEOUT = 10
|
|
|
|
|
|
def print_separator(title: str = ""):
|
|
"""打印分隔线"""
|
|
if title:
|
|
print(f"\n{'='*60}")
|
|
print(f" {title}")
|
|
print(f"{'='*60}")
|
|
else:
|
|
print(f"{'='*60}")
|
|
|
|
|
|
def print_result(response: requests.Response, test_name: str):
|
|
"""打印测试结果"""
|
|
print(f"\n测试: {test_name}")
|
|
print(f"状态码: {response.status_code}")
|
|
|
|
try:
|
|
data = response.json()
|
|
print(f"响应数据:")
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
# 检查是否成功
|
|
if data.get("code") == "200":
|
|
print(f"\n✅ 测试通过")
|
|
return True
|
|
else:
|
|
print(f"\n❌ 测试失败: {data.get('message', '未知错误')}")
|
|
return False
|
|
except Exception as e:
|
|
print(f"\n❌ 解析响应失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def test_token_with_all_params():
|
|
"""测试包含所有必填参数的 Token 请求"""
|
|
print_separator("测试1: 完整参数的 GetToken 请求")
|
|
|
|
request_data = {
|
|
"projectNo": "test_full_params_001",
|
|
"entityName": "测试企业-完整参数",
|
|
"userId": "902001",
|
|
"userName": "张三",
|
|
"appId": "remote_app",
|
|
"appSecretCode": "test_secret_code_12345",
|
|
"role": "VIEWER",
|
|
"orgCode": "902000",
|
|
"departmentCode": "902000",
|
|
"entityId": "91110000MA00ABCD12",
|
|
"xdRelatedPersons": json.dumps([
|
|
{"relatedPerson": "关联企业1", "relation": "股东"},
|
|
{"relatedPerson": "关联人1", "relation": "董事"}
|
|
], ensure_ascii=False),
|
|
"jzDataDateId": "0",
|
|
"innerBSStartDateId": "0",
|
|
"innerBSEndDateId": "0",
|
|
"analysisType": "-1"
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{BASE_URL}/account/common/getToken",
|
|
json=request_data,
|
|
timeout=TIMEOUT
|
|
)
|
|
return print_result(response, "完整参数请求")
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"\n❌ 请求失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def test_token_with_required_params_only():
|
|
"""测试仅包含必填参数的 Token 请求"""
|
|
print_separator("测试2: 仅必填参数的 GetToken 请求")
|
|
|
|
request_data = {
|
|
"projectNo": "test_required_params_002",
|
|
"entityName": "测试企业-仅必填参数",
|
|
"userId": "902001",
|
|
"userName": "李四",
|
|
"appId": "remote_app",
|
|
"appSecretCode": "test_secret_code_67890",
|
|
"role": "VIEWER",
|
|
"orgCode": "902000",
|
|
"departmentCode": "902000",
|
|
"analysisType": "-1"
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{BASE_URL}/account/common/getToken",
|
|
json=request_data,
|
|
timeout=TIMEOUT
|
|
)
|
|
return print_result(response, "必填参数请求")
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"\n❌ 请求失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def test_token_error_scenario():
|
|
"""测试错误场景触发"""
|
|
print_separator("测试3: 错误场景触发 (40101)")
|
|
|
|
request_data = {
|
|
"projectNo": "test_error_40101", # 包含错误标记
|
|
"entityName": "测试错误场景",
|
|
"userId": "902001",
|
|
"userName": "王五",
|
|
"appId": "remote_app",
|
|
"appSecretCode": "test_secret_code",
|
|
"role": "VIEWER",
|
|
"orgCode": "902000",
|
|
"departmentCode": "902000",
|
|
"analysisType": "-1"
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{BASE_URL}/account/common/getToken",
|
|
json=request_data,
|
|
timeout=TIMEOUT
|
|
)
|
|
|
|
print(f"\n测试: 错误场景触发")
|
|
print(f"状态码: {response.status_code}")
|
|
|
|
data = response.json()
|
|
print(f"响应数据:")
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
|
|
# 检查是否返回了预期的错误码
|
|
if data.get("code") == "40101":
|
|
print(f"\n✅ 测试通过 - 成功触发错误码 40101")
|
|
return True
|
|
else:
|
|
print(f"\n⚠️ 警告: 未触发预期错误码")
|
|
return False
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"\n❌ 请求失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def test_token_missing_required_param():
|
|
"""测试缺少必填参数的情况"""
|
|
print_separator("测试4: 缺少必填参数验证")
|
|
|
|
# 故意缺少 departmentCode
|
|
request_data = {
|
|
"projectNo": "test_missing_param_003",
|
|
"entityName": "测试缺少参数",
|
|
"userId": "902001",
|
|
"userName": "赵六",
|
|
"appId": "remote_app",
|
|
"appSecretCode": "test_secret_code",
|
|
"role": "VIEWER",
|
|
"orgCode": "902000",
|
|
"analysisType": "-1"
|
|
# 缺少 departmentCode
|
|
}
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{BASE_URL}/account/common/getToken",
|
|
json=request_data,
|
|
timeout=TIMEOUT
|
|
)
|
|
|
|
print(f"\n测试: 缺少必填参数")
|
|
print(f"状态码: {response.status_code}")
|
|
|
|
# 应该返回 422 Unprocessable Entity
|
|
if response.status_code == 422:
|
|
print(f"✅ 测试通过 - 正确拒绝了缺少必填参数的请求")
|
|
print(f"验证错误信息:")
|
|
print(json.dumps(response.json(), indent=2, ensure_ascii=False))
|
|
return True
|
|
else:
|
|
print(f"⚠️ 警告: 服务器接受了不完整的请求")
|
|
print(f"响应数据:")
|
|
print(json.dumps(response.json(), indent=2, ensure_ascii=False))
|
|
return False
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"\n❌ 请求失败: {str(e)}")
|
|
return False
|
|
|
|
|
|
def check_server_status():
|
|
"""检查服务器状态"""
|
|
print_separator("检查服务器状态")
|
|
|
|
try:
|
|
response = requests.get(f"{BASE_URL}/health", timeout=TIMEOUT)
|
|
if response.status_code == 200:
|
|
print(f"✅ 服务器运行中")
|
|
print(f"健康状态: {response.json()}")
|
|
return True
|
|
else:
|
|
print(f"❌ 服务器状态异常: {response.status_code}")
|
|
return False
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"❌ 无法连接到服务器: {str(e)}")
|
|
print(f"\n请确保服务器已启动:")
|
|
print(f" python main.py")
|
|
print(f" 或")
|
|
print(f" uvicorn main:app --reload --host 0.0.0.0 --port 8000")
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print_separator("接口参数修复验证")
|
|
print("\n此脚本用于验证 GetToken 接口参数修复是否成功")
|
|
print(f"服务器地址: {BASE_URL}\n")
|
|
|
|
# 检查服务器状态
|
|
if not check_server_status():
|
|
print_separator("验证失败")
|
|
print("请先启动服务器，然后重新运行此脚本")
|
|
sys.exit(1)
|
|
|
|
# 运行测试
|
|
results = []
|
|
results.append(("完整参数测试", test_token_with_all_params()))
|
|
results.append(("必填参数测试", test_token_with_required_params_only()))
|
|
results.append(("错误场景测试", test_token_error_scenario()))
|
|
results.append(("参数校验测试", test_token_missing_required_param()))
|
|
|
|
# 打印总结
|
|
print_separator("测试总结")
|
|
|
|
passed = sum(1 for _, result in results if result)
|
|
total = len(results)
|
|
|
|
print(f"\n总测试数: {total}")
|
|
print(f"通过: {passed}")
|
|
print(f"失败: {total - passed}")
|
|
|
|
print("\n详细结果:")
|
|
for name, result in results:
|
|
status = "✅ 通过" if result else "❌ 失败"
|
|
print(f" - {name}: {status}")
|
|
|
|
# 最终结论
|
|
print_separator()
|
|
if passed == total:
|
|
print("\n🎉 所有测试通过！接口参数修复成功！\n")
|
|
print("修复内容:")
|
|
print(" ✅ 添加 appId 参数")
|
|
print(" ✅ 添加 appSecretCode 参数")
|
|
print(" ✅ 添加 role 参数")
|
|
print(" ✅ 修复 analysisType 类型")
|
|
print(" ✅ 修复 departmentCode 必填性")
|
|
print()
|
|
sys.exit(0)
|
|
else:
|
|
print("\n⚠️ 部分测试失败，请检查修复是否完整\n")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|