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