hzdyjs/test_overseas_registration_list.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
出国登记列表测试脚本
用于诊断列表不显示数据的问题
"""

import requests
import json
from Crypto.Cipher import AES, PKCS1_v1_5
from Crypto.PublicKey import RSA
from Crypto.Util.Padding import pad
import base64
import time

# API基础配置
BASE_URL = "http://localhost:8080"
LOGIN_URL = f"{BASE_URL}/auth/login"
OVERSEAS_REGISTRATION_LIST_URL = f"{BASE_URL}/rs/overseas/registration/list"

# 测试账号
TEST_USER = {
    "username": "admin",
    "password": "admin123",
    "code": "0000",  # 验证码（测试环境可以用固定值）
    "uuid": ""
}

# 全局session
session = requests.Session()

# RSA公钥（从后端获取或硬编码）
RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCt9ejcgDRUDZr7qh/OfCQDPSZt
wMzxv7L8x9VnOuIElmH3bH5pSPmBGwkQhMKSbXGaWiW6pJXLhRQrLFpJYqD/n3z3
T8vHFsqCfwjlzJJDGXvT9w0dJhYT1WBmQ0GjQgx7bvOXJr6JWwEW3BXcB2NGZyF+
hIb5P9bXNBBdQhHOLQIDAQAB
-----END PUBLIC KEY-----"""

def aes_encrypt(data, key, iv):
    """AES加密"""
    cipher = AES.new(key.encode('utf-8'), AES.MODE_CBC, iv.encode('utf-8'))
    encrypted = cipher.encrypt(pad(data.encode('utf-8'), AES.block_size))
    return base64.b64encode(encrypted).decode('utf-8')

def rsa_encrypt(data, public_key):
    """RSA加密"""
    rsa_key = RSA.import_key(public_key)
    cipher = PKCS1_v1_5.new(rsa_key)
    encrypted = cipher.encrypt(data.encode('utf-8'))
    return base64.b64encode(encrypted).decode('utf-8')

def login():
    """登录并获取token"""
    print("=" * 80)
    print("步骤1: 登录系统")
    print("=" * 80)
    
    # AES密钥和IV
    aes_key = "1234567890abcdef"  # 16字节
    aes_iv = "1234567890abcdef"   # 16字节
    
    # 加密密码
    encrypted_password = aes_encrypt(TEST_USER["password"], aes_key, aes_iv)
    
    # 加密AES密钥
    encrypted_key = rsa_encrypt(aes_key, RSA_PUBLIC_KEY)
    
    # 登录请求体
    login_data = {
        "tenantId": "000000",
        "username": TEST_USER["username"],
        "password": encrypted_password,
        "code": TEST_USER["code"],
        "uuid": TEST_USER["uuid"],
        "clientid": "e5cd7e4891bf95d1d19206ce24a7b32e",
        "grant_type": "password",
        "key": encrypted_key
    }
    
    headers = {
        "Content-Type": "application/json",
        "clientid": "e5cd7e4891bf95d1d19206ce24a7b32e",
        "referer": "http://localhost"
    }
    
    print(f"📤 发送登录请求: {LOGIN_URL}")
    print(f"📋 用户名: {TEST_USER['username']}")
    
    try:
        response = session.post(LOGIN_URL, json=login_data, headers=headers)
        print(f"📡 响应状态码: {response.status_code}")
        
        if response.status_code == 200:
            result = response.json()
            print(f"📊 响应数据: {json.dumps(result, ensure_ascii=False, indent=2)}")
            
            if result.get("code") == 200:
                token = result.get("data", {}).get("access_token")
                if token:
                    print(f"✅ 登录成功! Token: {token[:50]}...")
                    return token
                else:
                    print("❌ 登录失败: 响应中没有token")
                    return None
            else:
                print(f"❌ 登录失败: {result.get('msg')}")
                return None
        else:
            print(f"❌ 登录失败: HTTP {response.status_code}")
            print(f"响应内容: {response.text}")
            return None
            
    except Exception as e:
        print(f"❌ 登录异常: {str(e)}")
        return None

def test_registration_list(token):
    """测试出国登记列表"""
    print("\n" + "=" * 80)
    print("步骤2: 查询出国登记列表")
    print("=" * 80)
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "clientid": "e5cd7e4891bf95d1d19206ce24a7b32e",
        "referer": "http://localhost"
    }
    
    # 测试不同的查询参数
    test_cases = [
        {
            "name": "无参数查询（查询所有）",
            "params": {
                "pageNum": 1,
                "pageSize": 10
            }
        },
        {
            "name": "状态过滤（已提交）",
            "params": {
                "pageNum": 1,
                "pageSize": 10,
                "status": "1"
            }
        },
        {
            "name": "状态过滤（草稿）",
            "params": {
                "pageNum": 1,
                "pageSize": 10,
                "status": "0"
            }
        }
    ]
    
    for i, test_case in enumerate(test_cases, 1):
        print(f"\n📋 测试案例 {i}: {test_case['name']}")
        print(f"📤 请求URL: {OVERSEAS_REGISTRATION_LIST_URL}")
        print(f"📋 查询参数: {json.dumps(test_case['params'], ensure_ascii=False)}")
        
        try:
            response = session.get(
                OVERSEAS_REGISTRATION_LIST_URL,
                params=test_case['params'],
                headers=headers
            )
            
            print(f"📡 响应状态码: {response.status_code}")
            
            if response.status_code == 200:
                result = response.json()
                print(f"📊 响应数据: {json.dumps(result, ensure_ascii=False, indent=2)}")
                
                if result.get("code") == 200:
                    rows = result.get("rows", [])
                    total = result.get("total", 0)
                    print(f"✅ 查询成功!")
                    print(f"📊 总记录数: {total}")
                    print(f"📊 当前页记录数: {len(rows)}")
                    
                    if len(rows) > 0:
                        print(f"\n📋 第一条记录详情:")
                        print(json.dumps(rows[0], ensure_ascii=False, indent=2))
                    else:
                        print("⚠️ 列表为空，没有数据")
                else:
                    print(f"❌ 查询失败: {result.get('msg')}")
            else:
                print(f"❌ 请求失败: HTTP {response.status_code}")
                print(f"响应内容: {response.text}")
                
        except Exception as e:
            print(f"❌ 请求异常: {str(e)}")
        
        if i < len(test_cases):
            print("\n" + "-" * 80)
            time.sleep(0.5)

def main():
    """主函数"""
    print("\n🚀 开始出国登记列表测试")
    print("=" * 80)
    
    # 1. 登录获取token
    token = login()
    if not token:
        print("\n❌ 测试终止: 登录失败")
        return
    
    # 2. 测试列表查询
    test_registration_list(token)
    
    print("\n" + "=" * 80)
    print("✅ 测试完成!")
    print("=" * 80)

if __name__ == "__main__":
    main()