ccdi/verify_fix.py

#!/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()