wkc
47f9491941
- 将员工表org_no字段迁移至dept_id，关联系统部门表 - 更新员工信息相关DTO、VO和Controller，使用deptId替代orgNo - 添加员工信息管理OpenSpec规范文档(proposal/design/spec/tasks) - 更新API文档，反映部门关联变更 - 添加数据库迁移脚本employee_org_no_to_dept_id.sql - 新增员工信息分页接口测试脚本(PowerShell/Python) - 更新CLAUDE.md，添加MCP数据库工具使用说明 Co-Authored-By: Claude (glm-4.7) <noreply@anthropic.com>
438 lines
14 KiB
Python
438 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""分页接口总数测试脚本
|
||
测试接口：
|
||
1. /dpc/employee/list - 员工列表（MyBatis Plus分页）
|
||
2. /dpc/intermediary/list - 中介黑名单列表（若依startPage分页）
|
||
"""
|
||
|
||
import sys
|
||
import io
|
||
import requests
|
||
import json
|
||
from datetime import datetime
|
||
|
||
# 设置stdout编码为UTF-8
|
||
if sys.platform == 'win32':
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
|
||
|
||
BASE_URL = "http://localhost:8080"
|
||
LOGIN_URL = f"{BASE_URL}/login/test"
|
||
EMPLOYEE_LIST_URL = f"{BASE_URL}/dpc/employee/list"
|
||
INTERMEDIARY_LIST_URL = f"{BASE_URL}/dpc/intermediary/list"
|
||
|
||
# 测试结果存储
|
||
test_results = []
|
||
|
||
|
||
def login():
|
||
"""登录获取token"""
|
||
print("=" * 60)
|
||
print("步骤1: 获取认证Token")
|
||
print("=" * 60)
|
||
|
||
login_body = {
|
||
"username": "admin",
|
||
"password": "admin123"
|
||
}
|
||
|
||
try:
|
||
# 使用json参数发送JSON格式请求体
|
||
response = requests.post(LOGIN_URL, json=login_body)
|
||
print(f"请求URL: {LOGIN_URL}")
|
||
print(f"请求参数: {login_body}")
|
||
print(f"响应状态码: {response.status_code}")
|
||
|
||
data = response.json()
|
||
if data.get("code") == 200:
|
||
token = data.get("token")
|
||
print(f"✓ Token获取成功: {token[:20]}...")
|
||
return token
|
||
else:
|
||
print(f"✗ Token获取失败: {data.get('msg')}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"✗ 异常: {e}")
|
||
return None
|
||
|
||
|
||
def test_page(url, token, page_num, page_size, test_name, api_type):
|
||
"""测试指定分页参数的接口"""
|
||
print(f"\n========== 测试: {test_name} ==========")
|
||
print(f"API类型: {api_type}")
|
||
print(f"URL: {url}")
|
||
print(f"参数: pageNum={page_num}, pageSize={page_size}")
|
||
|
||
params = {
|
||
"pageNum": page_num,
|
||
"pageSize": page_size
|
||
}
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {token}"
|
||
}
|
||
|
||
try:
|
||
response = requests.get(url, params=params, headers=headers)
|
||
print(f"响应状态码: {response.status_code}")
|
||
|
||
if response.status_code != 200:
|
||
print(f"✗ HTTP错误: {response.text}")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "FAIL",
|
||
"error": f"HTTP {response.status_code}"
|
||
})
|
||
return None
|
||
|
||
data = response.json()
|
||
print(f"响应内容:\n{json.dumps(data, indent=2, ensure_ascii=False)}")
|
||
|
||
code = data.get("code", 0)
|
||
|
||
if code != 200:
|
||
print(f"✗ 业务错误: {data.get('msg')}")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "FAIL",
|
||
"error": data.get("msg", "Unknown error")
|
||
})
|
||
return None
|
||
|
||
# 检查响应结构 - 支持两种格式
|
||
# 格式1: {data: {total, rows}, code, msg}
|
||
# 格式2: {total, rows, code, msg}
|
||
if "data" in data:
|
||
response_data = data["data"]
|
||
rows = response_data.get("rows", [])
|
||
total = response_data.get("total")
|
||
else:
|
||
# 扁平结构格式
|
||
rows = data.get("rows", [])
|
||
total = data.get("total")
|
||
|
||
rows_count = len(rows)
|
||
print(f"\n--- 分页数据分析 ---")
|
||
print(f"返回行数(rows): {rows_count}")
|
||
print(f"总数(total): {total}")
|
||
|
||
# 验证total字段
|
||
if total is None:
|
||
print(f"✗ 响应缺少total字段")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "FAIL",
|
||
"error": "响应缺少total字段"
|
||
})
|
||
return None
|
||
|
||
if not isinstance(total, int):
|
||
print(f"✗ total类型错误: {type(total)}")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "FAIL",
|
||
"error": f"total类型错误: {type(total)}"
|
||
})
|
||
return None
|
||
|
||
if total < 0:
|
||
print(f"✗ total值无效: {total}")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "FAIL",
|
||
"error": f"total值无效: {total}"
|
||
})
|
||
return None
|
||
|
||
# 计算预期值
|
||
expected_total_pages = (total + page_size - 1) // page_size
|
||
if page_num < expected_total_pages:
|
||
expected_rows = page_size
|
||
elif page_num == expected_total_pages:
|
||
expected_rows = total - (page_size * (page_num - 1))
|
||
else:
|
||
expected_rows = 0
|
||
|
||
print(f"预期行数: {expected_rows}")
|
||
print(f"预期总页数: {expected_total_pages}")
|
||
|
||
# 验证行数是否正确
|
||
is_correct = rows_count == expected_rows
|
||
|
||
if is_correct:
|
||
print(f"✓ 测试通过 - 分页总数返回正常")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "PASS",
|
||
"page_num": page_num,
|
||
"page_size": page_size,
|
||
"rows_count": rows_count,
|
||
"total": total,
|
||
"expected_rows": expected_rows
|
||
})
|
||
else:
|
||
print(f"✗ 测试失败 - 预期{expected_rows}行，实际{rows_count}行")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "FAIL",
|
||
"page_num": page_num,
|
||
"page_size": page_size,
|
||
"rows_count": rows_count,
|
||
"total": total,
|
||
"expected_rows": expected_rows,
|
||
"error": f"行数不匹配"
|
||
})
|
||
|
||
# 显示返回的ID
|
||
if rows:
|
||
if "employeeId" in rows[0]:
|
||
ids = ', '.join([str(r.get("employeeId")) for r in rows])
|
||
print(f"员工ID: {ids}")
|
||
elif "intermediaryId" in rows[0]:
|
||
ids = ', '.join([str(r.get("intermediaryId")) for r in rows])
|
||
print(f"中介ID: {ids}")
|
||
|
||
return {
|
||
"success": True,
|
||
"rows": rows_count,
|
||
"total": total,
|
||
"expected_rows": expected_rows,
|
||
"is_correct": is_correct
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"✗ 异常: {e}")
|
||
test_results.append({
|
||
"test_name": test_name,
|
||
"api_type": api_type,
|
||
"status": "ERROR",
|
||
"error": str(e)
|
||
})
|
||
return None
|
||
|
||
|
||
def test_consistency(url, token, test_name, api_type):
|
||
"""测试不同pageSize下total是否一致"""
|
||
print(f"\n========== 测试总数一致性: {test_name} ==========")
|
||
|
||
page_sizes = [10, 20, 50]
|
||
totals = []
|
||
|
||
for size in page_sizes:
|
||
params = {"pageNum": 1, "pageSize": size}
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
response = requests.get(url, params=params, headers=headers)
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
if data.get("code") == 200:
|
||
# 支持两种响应格式
|
||
if "data" in data:
|
||
total = data.get("data", {}).get("total")
|
||
else:
|
||
total = data.get("total")
|
||
totals.append(total)
|
||
print(f"pageSize={size}: total={total}")
|
||
except Exception as e:
|
||
print(f"✗ 异常: {e}")
|
||
|
||
if len(set(totals)) == 1 and totals[0] is not None:
|
||
print(f"✓ 不同pageSize下总数一致: {totals[0]}")
|
||
return True
|
||
else:
|
||
print(f"✗ 不同pageSize下总数不一致: {totals}")
|
||
return False
|
||
|
||
|
||
def generate_report():
|
||
"""生成测试报告"""
|
||
print("\n" + "=" * 60)
|
||
print("测试报告")
|
||
print("=" * 60)
|
||
|
||
# 按API类型分组
|
||
employee_results = [r for r in test_results if "员工" in r.get("test_name", "")]
|
||
intermediary_results = [r for r in test_results if "中介" in r.get("test_name", "")]
|
||
|
||
pass_count = sum(1 for r in test_results if r["status"] == "PASS")
|
||
fail_count = sum(1 for r in test_results if r["status"] == "FAIL")
|
||
error_count = sum(1 for r in test_results if r["status"] == "ERROR")
|
||
|
||
print(f"\n总测试数: {len(test_results)}")
|
||
print(f"通过: {pass_count}")
|
||
print(f"失败: {fail_count}")
|
||
print(f"错误: {error_count}")
|
||
|
||
# 员工接口结果
|
||
print(f"\n--- 员工列表接口 (MyBatis Plus) ---")
|
||
print(f"测试数: {len(employee_results)}")
|
||
for r in employee_results:
|
||
status_icon = "✓" if r["status"] == "PASS" else "✗"
|
||
print(f"{status_icon} {r['test_name']}: {r['status']}")
|
||
if r["status"] == "PASS":
|
||
print(f" 页码: {r.get('page_num')}/{r.get('page_size')}, "
|
||
f"返回行数: {r.get('rows_count')}, 总数: {r.get('total')}")
|
||
else:
|
||
print(f" 错误: {r.get('error', 'Unknown')}")
|
||
|
||
# 中介黑名单接口结果
|
||
print(f"\n--- 中介黑名单接口 (若依startPage) ---")
|
||
print(f"测试数: {len(intermediary_results)}")
|
||
for r in intermediary_results:
|
||
status_icon = "✓" if r["status"] == "PASS" else "✗"
|
||
print(f"{status_icon} {r['test_name']}: {r['status']}")
|
||
if r["status"] == "PASS":
|
||
print(f" 页码: {r.get('page_num')}/{r.get('page_size')}, "
|
||
f"返回行数: {r.get('rows_count')}, 总数: {r.get('total')}")
|
||
else:
|
||
print(f" 错误: {r.get('error', 'Unknown')}")
|
||
|
||
# 总体结论
|
||
print(f"\n--- 测试结论 ---")
|
||
if fail_count == 0 and error_count == 0:
|
||
print("✓ 所有分页接口总数返回正常")
|
||
else:
|
||
print("✗ 存在分页接口总数返回异常")
|
||
|
||
# 保存报告到文件
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
report_file = f"test/pagination_test_report_{timestamp}.txt"
|
||
|
||
with open(report_file, "w", encoding="utf-8") as f:
|
||
f.write("=" * 60 + "\n")
|
||
f.write("分页接口总数测试报告\n")
|
||
f.write("=" * 60 + "\n\n")
|
||
f.write(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
|
||
|
||
f.write("测试统计:\n")
|
||
f.write(f" 总测试数: {len(test_results)}\n")
|
||
f.write(f" 通过: {pass_count}\n")
|
||
f.write(f" 失败: {fail_count}\n")
|
||
f.write(f" 错误: {error_count}\n\n")
|
||
|
||
f.write("测试接口:\n")
|
||
f.write(" 1. /dpc/employee/list - 员工列表（MyBatis Plus分页）\n")
|
||
f.write(" 2. /dpc/intermediary/list - 中介黑名单列表（若依startPage分页）\n\n")
|
||
|
||
f.write("-" * 60 + "\n")
|
||
f.write("详细结果:\n")
|
||
f.write("-" * 60 + "\n\n")
|
||
|
||
for r in test_results:
|
||
f.write(f"测试: {r['test_name']}\n")
|
||
f.write(f"API类型: {r['api_type']}\n")
|
||
f.write(f"状态: {r['status']}\n")
|
||
if r['status'] == 'PASS':
|
||
f.write(f" 页码: {r.get('page_num')}/{r.get('page_size')}\n")
|
||
f.write(f" 返回行数: {r.get('rows_count')}\n")
|
||
f.write(f" 总数: {r.get('total')}\n")
|
||
f.write(f" 预期行数: {r.get('expected_rows')}\n")
|
||
else:
|
||
f.write(f" 错误: {r.get('error', 'Unknown')}\n")
|
||
f.write("\n")
|
||
|
||
f.write("-" * 60 + "\n")
|
||
f.write("测试结论:\n")
|
||
if fail_count == 0 and error_count == 0:
|
||
f.write("✓ 所有分页接口总数返回正常\n")
|
||
else:
|
||
f.write("✗ 存在分页接口总数返回异常\n")
|
||
|
||
print(f"\n报告已保存至: {report_file}")
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("\n" + "=" * 60)
|
||
print("分页接口总数测试")
|
||
print("=" * 60)
|
||
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
# 获取token
|
||
token = login()
|
||
if not token:
|
||
print("\n✗ 无法获取token，测试终止")
|
||
return
|
||
|
||
print("\n✓ 登录成功，开始测试")
|
||
|
||
# 员工列表接口测试用例
|
||
employee_test_cases = [
|
||
{"page_num": 1, "page_size": 10, "desc": "员工列表 - 第1页(10条/页)"},
|
||
{"page_num": 2, "page_size": 10, "desc": "员工列表 - 第2页(10条/页)"},
|
||
{"page_num": 1, "page_size": 5, "desc": "员工列表 - 第1页(5条/页)"},
|
||
{"page_num": 1, "page_size": 20, "desc": "员工列表 - 第1页(20条/页)"},
|
||
]
|
||
|
||
print("\n" + "=" * 60)
|
||
print("测试员工列表接口（MyBatis Plus分页）")
|
||
print("=" * 60)
|
||
|
||
for test_case in employee_test_cases:
|
||
test_page(
|
||
EMPLOYEE_LIST_URL,
|
||
token,
|
||
test_case["page_num"],
|
||
test_case["page_size"],
|
||
test_case["desc"],
|
||
"MyBatis Plus"
|
||
)
|
||
|
||
# 测试总数一致性
|
||
test_consistency(
|
||
EMPLOYEE_LIST_URL,
|
||
token,
|
||
"员工列表-总数一致性",
|
||
"MyBatis Plus"
|
||
)
|
||
|
||
# 中介黑名单接口测试用例
|
||
intermediary_test_cases = [
|
||
{"page_num": 1, "page_size": 10, "desc": "中介黑名单 - 第1页(10条/页)"},
|
||
{"page_num": 2, "page_size": 10, "desc": "中介黑名单 - 第2页(10条/页)"},
|
||
{"page_num": 1, "page_size": 5, "desc": "中介黑名单 - 第1页(5条/页)"},
|
||
{"page_num": 1, "page_size": 20, "desc": "中介黑名单 - 第1页(20条/页)"},
|
||
]
|
||
|
||
print("\n" + "=" * 60)
|
||
print("测试中介黑名单接口（若依startPage分页）")
|
||
print("=" * 60)
|
||
|
||
for test_case in intermediary_test_cases:
|
||
test_page(
|
||
INTERMEDIARY_LIST_URL,
|
||
token,
|
||
test_case["page_num"],
|
||
test_case["page_size"],
|
||
test_case["desc"],
|
||
"若依startPage"
|
||
)
|
||
|
||
# 测试总数一致性
|
||
test_consistency(
|
||
INTERMEDIARY_LIST_URL,
|
||
token,
|
||
"中介黑名单-总数一致性",
|
||
"若依startPage"
|
||
)
|
||
|
||
# 生成报告
|
||
generate_report()
|
||
|
||
print("\n" + "=" * 60)
|
||
print("测试完成")
|
||
print("=" * 60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|