309 lines
10 KiB
Python
309 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
中介导入唯一性校验测试脚本
|
||
|
||
测试场景：
|
||
1. 个人中介导入 - 证件号重复导入（非更新模式）应失败
|
||
2. 个人中介导入 - 证件号重复导入（更新模式）应成功
|
||
3. 机构中介导入 - 统一社会信用代码重复导入（非更新模式）应失败
|
||
4. 机构中介导入 - 统一社会信用代码重复导入（更新模式）应成功
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:8080"
|
||
USERNAME = "admin"
|
||
PASSWORD = "admin123"
|
||
|
||
# 全局变量存储token
|
||
token = None
|
||
|
||
|
||
def login():
|
||
"""登录获取token"""
|
||
global token
|
||
url = f"{BASE_URL}/login/test"
|
||
data = {
|
||
"username": USERNAME,
|
||
"password": PASSWORD
|
||
}
|
||
response = requests.post(url, data=data)
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
token = result.get("token")
|
||
print(f"✓ 登录成功，获取token: {token[:20]}...")
|
||
return True
|
||
else:
|
||
print(f"✗ 登录失败: {result}")
|
||
return False
|
||
|
||
|
||
def get_headers():
|
||
"""获取带token的请求头"""
|
||
return {
|
||
"Authorization": f"Bearer {token}"
|
||
}
|
||
|
||
|
||
def test_import_person_without_update(file_path, cert_no):
|
||
"""
|
||
测试场景1: 个人中介导入（非更新模式）- 证件号重复
|
||
期望：导入失败，提示证件号已存在
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"测试场景1: 个人中介导入（非更新模式）- 证件号 {cert_no} 重复")
|
||
print(f"{'='*60}")
|
||
|
||
url = f"{BASE_URL}/dpc/intermediary/importPersonData"
|
||
files = {"file": open(file_path, "rb")}
|
||
data = {"updateSupport": "false"}
|
||
|
||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||
result = response.json()
|
||
|
||
print(f"响应状态码: {response.status_code}")
|
||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||
|
||
# 验证结果
|
||
if result.get("code") == 500:
|
||
if "已存在" in result.get("msg", ""):
|
||
print(f"✓ 测试通过：系统正确拒绝了重复的证件号")
|
||
return True
|
||
else:
|
||
print(f"✗ 测试失败：错误信息不符合预期")
|
||
return False
|
||
else:
|
||
print(f"✗ 测试失败：系统应该拒绝重复的证件号")
|
||
return False
|
||
|
||
|
||
def test_import_person_with_update(file_path, cert_no):
|
||
"""
|
||
测试场景2: 个人中介导入（更新模式）- 证件号重复
|
||
期望：导入成功，更新已存在的记录
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"测试场景2: 个人中介导入（更新模式）- 证件号 {cert_no} 重复")
|
||
print(f"{'='*60}")
|
||
|
||
url = f"{BASE_URL}/dpc/intermediary/importPersonData"
|
||
files = {"file": open(file_path, "rb")}
|
||
data = {"updateSupport": "true"}
|
||
|
||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||
result = response.json()
|
||
|
||
print(f"响应状态码: {response.status_code}")
|
||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||
|
||
# 验证结果
|
||
if result.get("code") == 200:
|
||
print(f"✓ 测试通过：系统成功更新了已存在的记录")
|
||
return True
|
||
else:
|
||
print(f"✗ 测试失败：系统应该允许更新模式")
|
||
return False
|
||
|
||
|
||
def test_import_entity_without_update(file_path, credit_code):
|
||
"""
|
||
测试场景3: 机构中介导入（非更新模式）- 统一社会信用代码重复
|
||
期望：导入失败，提示统一社会信用代码已存在
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"测试场景3: 机构中介导入（非更新模式）- 统一社会信用代码 {credit_code} 重复")
|
||
print(f"{'='*60}")
|
||
|
||
url = f"{BASE_URL}/dpc/intermediary/importEntityData"
|
||
files = {"file": open(file_path, "rb")}
|
||
data = {"updateSupport": "false"}
|
||
|
||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||
result = response.json()
|
||
|
||
print(f"响应状态码: {response.status_code}")
|
||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||
|
||
# 验证结果
|
||
if result.get("code") == 500:
|
||
if "已存在" in result.get("msg", ""):
|
||
print(f"✓ 测试通过：系统正确拒绝了重复的统一社会信用代码")
|
||
return True
|
||
else:
|
||
print(f"✗ 测试失败：错误信息不符合预期")
|
||
return False
|
||
else:
|
||
print(f"✗ 测试失败：系统应该拒绝重复的统一社会信用代码")
|
||
return False
|
||
|
||
|
||
def test_import_entity_with_update(file_path, credit_code):
|
||
"""
|
||
测试场景4: 机构中介导入（更新模式）- 统一社会信用代码重复
|
||
期望：导入成功，更新已存在的记录
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"测试场景4: 机构中介导入（更新模式）- 统一社会信用代码 {credit_code} 重复")
|
||
print(f"{'='*60}")
|
||
|
||
url = f"{BASE_URL}/dpc/intermediary/importEntityData"
|
||
files = {"file": open(file_path, "rb")}
|
||
data = {"updateSupport": "true"}
|
||
|
||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||
result = response.json()
|
||
|
||
print(f"响应状态码: {response.status_code}")
|
||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||
|
||
# 验证结果
|
||
if result.get("code") == 200:
|
||
print(f"✓ 测试通过：系统成功更新了已存在的记录")
|
||
return True
|
||
else:
|
||
print(f"✗ 测试失败：系统应该允许更新模式")
|
||
return False
|
||
|
||
|
||
def create_test_person_excel(file_path, cert_no, name="测试用户"):
|
||
"""创建测试用的个人中介Excel文件"""
|
||
import openpyxl
|
||
from openpyxl.styles import Protection
|
||
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "个人中介黑名单"
|
||
|
||
# 表头
|
||
headers = ["姓名", "证件号码", "人员类型", "人员子类型", "性别", "证件类型", "手机号", "微信号",
|
||
"联系地址", "所在公司", "职位", "关联人员ID", "关联关系", "备注"]
|
||
ws.append(headers)
|
||
|
||
# 添加测试数据
|
||
ws.append([name, cert_no, "中介", "本人", "男", "身份证", "13800138000",
|
||
"test_wxh", "测试地址", "测试公司", "测试职位", "", "", "测试备注"])
|
||
|
||
wb.save(file_path)
|
||
print(f"✓ 创建测试Excel文件: {file_path}")
|
||
|
||
|
||
def create_test_entity_excel(file_path, credit_code, name="测试机构"):
|
||
"""创建测试用的机构中介Excel文件"""
|
||
import openpyxl
|
||
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "机构中介黑名单"
|
||
|
||
# 表头
|
||
headers = ["机构名称", "统一社会信用代码", "主体类型", "企业性质", "行业分类", "所属行业", "成立日期",
|
||
"注册地址", "法定代表人", "法定代表人证件类型", "法定代表人证件号码", "股东1", "股东2",
|
||
"股东3", "股东4", "股东5", "备注"]
|
||
ws.append(headers)
|
||
|
||
# 添加测试数据
|
||
ws.append([name, credit_code, "有限责任公司", "民企", "金融业", "银行业", "2020-01-01",
|
||
"北京市测试区测试路123号", "张三", "身份证", "110101199001011234",
|
||
"股东A", "股东B", "股东C", "股东D", "股东E", "测试备注"])
|
||
|
||
wb.save(file_path)
|
||
print(f"✓ 创建测试Excel文件: {file_path}")
|
||
|
||
|
||
def main():
|
||
"""主测试流程"""
|
||
print(f"\n{'#'*60}")
|
||
print(f"# 中介导入唯一性校验测试")
|
||
print(f"# 测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"{'#'*60}")
|
||
|
||
# 登录
|
||
if not login():
|
||
return
|
||
|
||
# 测试参数
|
||
test_cert_no = f"TEST{int(time.time())}" # 生成唯一测试证件号
|
||
test_credit_code = f"91{int(time.time())}001" # 生成唯一测试统一社会信用代码
|
||
|
||
test_results = []
|
||
|
||
# 准备测试文件
|
||
person_file = "test_person_uniqueness.xlsx"
|
||
entity_file = "test_entity_uniqueness.xlsx"
|
||
|
||
# ========== 场景1: 先导入一条个人数据 ==========
|
||
print(f"\n{'='*60}")
|
||
print(f"准备步骤: 首次导入个人中介数据（证件号: {test_cert_no}）")
|
||
print(f"{'='*60}")
|
||
|
||
create_test_person_excel(person_file, test_cert_no)
|
||
|
||
url = f"{BASE_URL}/dpc/intermediary/importPersonData"
|
||
files = {"file": open(person_file, "rb")}
|
||
data = {"updateSupport": "false"}
|
||
|
||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
print(f"✓ 首次导入成功")
|
||
else:
|
||
print(f"✗ 首次导入失败: {result}")
|
||
return
|
||
|
||
# ========== 场景1: 非更新模式导入重复个人数据 ==========
|
||
test_results.append(test_import_person_without_update(person_file, test_cert_no))
|
||
|
||
# ========== 场景2: 更新模式导入重复个人数据 ==========
|
||
test_results.append(test_import_person_with_update(person_file, test_cert_no))
|
||
|
||
# ========== 准备: 首次导入机构数据 ==========
|
||
print(f"\n{'='*60}")
|
||
print(f"准备步骤: 首次导入机构中介数据（统一社会信用代码: {test_credit_code}）")
|
||
print(f"{'='*60}")
|
||
|
||
create_test_entity_excel(entity_file, test_credit_code)
|
||
|
||
url = f"{BASE_URL}/dpc/intermediary/importEntityData"
|
||
files = {"file": open(entity_file, "rb")}
|
||
data = {"updateSupport": "false"}
|
||
|
||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||
result = response.json()
|
||
|
||
if result.get("code") == 200:
|
||
print(f"✓ 首次导入成功")
|
||
else:
|
||
print(f"✗ 首次导入失败: {result}")
|
||
return
|
||
|
||
# ========== 场景3: 非更新模式导入重复机构数据 ==========
|
||
test_results.append(test_import_entity_without_update(entity_file, test_credit_code))
|
||
|
||
# ========== 场景4: 更新模式导入重复机构数据 ==========
|
||
test_results.append(test_import_entity_with_update(entity_file, test_credit_code))
|
||
|
||
# ========== 输出测试报告 ==========
|
||
print(f"\n{'='*60}")
|
||
print(f"测试报告汇总")
|
||
print(f"{'='*60}")
|
||
print(f"测试场景总数: {len(test_results)}")
|
||
print(f"通过数量: {sum(test_results)}")
|
||
print(f"失败数量: {len(test_results) - sum(test_results)}")
|
||
|
||
if all(test_results):
|
||
print(f"\n✓ 所有测试通过！")
|
||
else:
|
||
print(f"\n✗ 部分测试失败，请查看上方详细日志")
|
||
|
||
print(f"\n测试完成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|