wuzhen
661a513f57
- 新增SysUserController.listUsersByPermission接口 - 新增SysUserService.selectUserListByPermission方法 - 支持多租户和权限关联查询 - 更新变更记录文档
631 lines
25 KiB
Python
631 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
班级信息管理API测试脚本
|
|
功能：测试班级信息管理的所有API接口
|
|
作者：Cursor AI 辅助
|
|
版本：V1.0.0
|
|
时间：2025-01-14
|
|
|
|
API接口列表：
|
|
1. 登录获取Token
|
|
2. 查询班级信息列表
|
|
3. 新增班级信息
|
|
4. 获取班级信息详细信息
|
|
5. 修改班级信息
|
|
6. 根据班级代码查询班级信息
|
|
7. 查询学院班级列表
|
|
8. 获取班级选项列表
|
|
9. 检查班级代码是否存在
|
|
10. 更新班级人数统计
|
|
11. 删除班级信息
|
|
12. 导出班级信息
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import hashlib
|
|
import base64
|
|
import time
|
|
import uuid
|
|
from datetime import datetime
|
|
from Crypto.Cipher import AES
|
|
from Crypto.PublicKey import RSA
|
|
from Crypto.Cipher import PKCS1_v1_5
|
|
from Crypto.Util.Padding import pad
|
|
|
|
class ClassInfoAPITester:
|
|
def __init__(self):
|
|
self.base_url = "http://localhost:8080"
|
|
self.session = requests.Session()
|
|
self.token = None
|
|
self.client_id = "e5cd7e4891bf95d1d19206ce24a7b32e"
|
|
self.test_class_id = None # 存储测试过程中创建的班级ID
|
|
self.test_results = []
|
|
|
|
def log_test_result(self, test_name, success, message, response_data=None):
|
|
"""记录测试结果"""
|
|
result = {
|
|
'test_name': test_name,
|
|
'success': success,
|
|
'message': message,
|
|
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
|
'response_data': response_data
|
|
}
|
|
self.test_results.append(result)
|
|
|
|
status = "✅ PASS" if success else "❌ FAIL"
|
|
print(f"[{result['timestamp']}] {status} - {test_name}")
|
|
print(f" 消息: {message}")
|
|
if response_data:
|
|
print(f" 响应: {json.dumps(response_data, ensure_ascii=False, indent=2)}")
|
|
print("-" * 80)
|
|
|
|
def get_rsa_public_key(self):
|
|
"""获取RSA公钥"""
|
|
try:
|
|
response = self.session.get(f"{self.base_url}/auth/getPublicKey")
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
return data.get('data')
|
|
return None
|
|
except Exception as e:
|
|
print(f"获取RSA公钥失败: {e}")
|
|
return None
|
|
|
|
def rsa_encrypt(self, data, public_key_str):
|
|
"""RSA加密"""
|
|
try:
|
|
# 格式化公钥
|
|
public_key_str = f"-----BEGIN PUBLIC KEY-----\n{public_key_str}\n-----END PUBLIC KEY-----"
|
|
public_key = RSA.import_key(public_key_str)
|
|
cipher = PKCS1_v1_5.new(public_key)
|
|
|
|
# RSA加密
|
|
encrypted_data = cipher.encrypt(data.encode('utf-8'))
|
|
return base64.b64encode(encrypted_data).decode('utf-8')
|
|
except Exception as e:
|
|
print(f"RSA加密失败: {e}")
|
|
return None
|
|
|
|
def aes_encrypt(self, data, key):
|
|
"""AES加密"""
|
|
try:
|
|
# 使用MD5哈希作为AES密钥
|
|
key_hash = hashlib.md5(key.encode('utf-8')).digest()
|
|
|
|
# AES加密
|
|
cipher = AES.new(key_hash, AES.MODE_ECB)
|
|
padded_data = pad(data.encode('utf-8'), 16)
|
|
encrypted_data = cipher.encrypt(padded_data)
|
|
|
|
return base64.b64encode(encrypted_data).decode('utf-8')
|
|
except Exception as e:
|
|
print(f"AES加密失败: {e}")
|
|
return None
|
|
|
|
def login(self, username="admin", password="admin123"):
|
|
"""登录获取Token"""
|
|
try:
|
|
# 第一步：获取RSA公钥
|
|
public_key = self.get_rsa_public_key()
|
|
if not public_key:
|
|
raise Exception("获取RSA公钥失败")
|
|
|
|
# 第二步：生成AES密钥
|
|
aes_key = str(uuid.uuid4()).replace('-', '')[:16]
|
|
|
|
# 第三步：AES加密用户名和密码
|
|
encrypted_username = self.aes_encrypt(username, aes_key)
|
|
encrypted_password = self.aes_encrypt(password, aes_key)
|
|
|
|
if not encrypted_username or not encrypted_password:
|
|
raise Exception("AES加密失败")
|
|
|
|
# 第四步：RSA加密AES密钥
|
|
encrypted_aes_key = self.rsa_encrypt(aes_key, public_key)
|
|
if not encrypted_aes_key:
|
|
raise Exception("RSA加密AES密钥失败")
|
|
|
|
# 第五步：准备登录请求
|
|
login_data = {
|
|
"username": encrypted_username,
|
|
"password": encrypted_password,
|
|
"key": encrypted_aes_key
|
|
}
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'clientid': self.client_id,
|
|
'referer': 'http://localhost'
|
|
}
|
|
|
|
# 第六步：发送登录请求
|
|
response = self.session.post(
|
|
f"{self.base_url}/auth/login",
|
|
json=login_data,
|
|
headers=headers
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
self.token = data.get('data', {}).get('access_token')
|
|
if self.token:
|
|
# 设置默认请求头
|
|
self.session.headers.update({
|
|
'Authorization': f'Bearer {self.token}',
|
|
'clientid': self.client_id,
|
|
'referer': 'http://localhost',
|
|
'Content-Type': 'application/json'
|
|
})
|
|
self.log_test_result("用户登录", True, f"登录成功，获取到Token: {self.token[:20]}...")
|
|
return True
|
|
else:
|
|
raise Exception("响应中未找到access_token")
|
|
else:
|
|
raise Exception(f"登录失败: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("用户登录", False, f"登录失败: {str(e)}")
|
|
return False
|
|
|
|
def test_get_class_list(self):
|
|
"""测试查询班级信息列表"""
|
|
try:
|
|
params = {
|
|
'pageNum': 1,
|
|
'pageSize': 10
|
|
}
|
|
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/list", params=params)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
rows = data.get('rows', [])
|
|
total = data.get('total', 0)
|
|
self.log_test_result(
|
|
"查询班级信息列表",
|
|
True,
|
|
f"查询成功，共{total}条记录，返回{len(rows)}条数据",
|
|
{'total': total, 'count': len(rows)}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("查询班级信息列表", False, f"查询失败: {str(e)}")
|
|
return False
|
|
|
|
def test_create_class(self):
|
|
"""测试新增班级信息"""
|
|
try:
|
|
# 生成唯一的班级代码
|
|
timestamp = int(time.time())
|
|
class_data = {
|
|
"classCode": f"TEST{timestamp}",
|
|
"className": f"测试班级{timestamp}",
|
|
"classType": "regular",
|
|
"grade": "2024级",
|
|
"enrollmentYear": 2024,
|
|
"graduationYear": 2028,
|
|
"majorCode": "CS001",
|
|
"majorName": "计算机科学与技术",
|
|
"collegeCode": "CS",
|
|
"collegeName": "计算机学院",
|
|
"classStatus": "1",
|
|
"maxStudentCount": 30,
|
|
"classDescription": "这是一个测试班级",
|
|
"remark": "API测试创建的班级"
|
|
}
|
|
|
|
response = self.session.post(f"{self.base_url}/xg/classInfo", json=class_data)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
class_id = data.get('data')
|
|
if class_id:
|
|
self.test_class_id = class_id # 保存班级ID供后续测试使用
|
|
self.log_test_result(
|
|
"新增班级信息",
|
|
True,
|
|
f"新增成功，班级ID: {class_id}",
|
|
{'classId': class_id, 'classCode': class_data['classCode']}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception("新增成功但未返回班级ID")
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("新增班级信息", False, f"新增失败: {str(e)}")
|
|
return False
|
|
|
|
def test_get_class_detail(self):
|
|
"""测试获取班级信息详细信息"""
|
|
if not self.test_class_id:
|
|
self.log_test_result("获取班级详细信息", False, "没有可用的测试班级ID")
|
|
return False
|
|
|
|
try:
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/{self.test_class_id}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
class_info = data.get('data')
|
|
if class_info:
|
|
self.log_test_result(
|
|
"获取班级详细信息",
|
|
True,
|
|
f"获取成功，班级名称: {class_info.get('className')}",
|
|
{'className': class_info.get('className'), 'classCode': class_info.get('classCode')}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception("未返回班级详细信息")
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("获取班级详细信息", False, f"获取失败: {str(e)}")
|
|
return False
|
|
|
|
def test_update_class(self):
|
|
"""测试修改班级信息"""
|
|
if not self.test_class_id:
|
|
self.log_test_result("修改班级信息", False, "没有可用的测试班级ID")
|
|
return False
|
|
|
|
try:
|
|
# 先获取班级信息
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/{self.test_class_id}")
|
|
if response.status_code != 200:
|
|
raise Exception("无法获取班级信息进行修改")
|
|
|
|
class_info = response.json().get('data')
|
|
if not class_info:
|
|
raise Exception("班级信息为空")
|
|
|
|
# 修改班级信息
|
|
class_info['className'] = f"{class_info['className']}_已修改"
|
|
class_info['classDescription'] = "这是修改后的班级描述"
|
|
class_info['maxStudentCount'] = 35
|
|
|
|
response = self.session.put(f"{self.base_url}/xg/classInfo", json=class_info)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
self.log_test_result(
|
|
"修改班级信息",
|
|
True,
|
|
f"修改成功，新班级名称: {class_info['className']}",
|
|
{'className': class_info['className']}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("修改班级信息", False, f"修改失败: {str(e)}")
|
|
return False
|
|
|
|
def test_get_class_by_code(self):
|
|
"""测试根据班级代码查询班级信息"""
|
|
if not self.test_class_id:
|
|
self.log_test_result("根据班级代码查询", False, "没有可用的测试班级")
|
|
return False
|
|
|
|
try:
|
|
# 先获取班级代码
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/{self.test_class_id}")
|
|
if response.status_code != 200:
|
|
raise Exception("无法获取班级代码")
|
|
|
|
class_info = response.json().get('data')
|
|
class_code = class_info.get('classCode')
|
|
|
|
if not class_code:
|
|
raise Exception("班级代码为空")
|
|
|
|
# 根据班级代码查询
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/byCode/{class_code}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
found_class = data.get('data')
|
|
if found_class:
|
|
self.log_test_result(
|
|
"根据班级代码查询",
|
|
True,
|
|
f"查询成功，找到班级: {found_class.get('className')}",
|
|
{'classCode': class_code, 'className': found_class.get('className')}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception("未找到对应班级")
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("根据班级代码查询", False, f"查询失败: {str(e)}")
|
|
return False
|
|
|
|
def test_get_class_by_college(self):
|
|
"""测试查询学院班级列表"""
|
|
try:
|
|
college_code = "CS" # 计算机学院
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/byCollege/{college_code}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
class_list = data.get('data', [])
|
|
self.log_test_result(
|
|
"查询学院班级列表",
|
|
True,
|
|
f"查询成功，{college_code}学院共有{len(class_list)}个班级",
|
|
{'collegeCode': college_code, 'classCount': len(class_list)}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("查询学院班级列表", False, f"查询失败: {str(e)}")
|
|
return False
|
|
|
|
def test_get_class_options(self):
|
|
"""测试获取班级选项列表"""
|
|
try:
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/options")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
options = data.get('data', [])
|
|
self.log_test_result(
|
|
"获取班级选项列表",
|
|
True,
|
|
f"获取成功，共{len(options)}个班级选项",
|
|
{'optionCount': len(options)}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("获取班级选项列表", False, f"获取失败: {str(e)}")
|
|
return False
|
|
|
|
def test_check_class_code_exists(self):
|
|
"""测试检查班级代码是否存在"""
|
|
try:
|
|
# 测试存在的班级代码
|
|
if self.test_class_id:
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/{self.test_class_id}")
|
|
if response.status_code == 200:
|
|
class_info = response.json().get('data')
|
|
class_code = class_info.get('classCode')
|
|
|
|
params = {'classCode': class_code}
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/checkClassCode", params=params)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
exists = data.get('data')
|
|
if exists:
|
|
self.log_test_result(
|
|
"检查班级代码存在性",
|
|
True,
|
|
f"检查成功，班级代码{class_code}存在",
|
|
{'classCode': class_code, 'exists': exists}
|
|
)
|
|
else:
|
|
raise Exception(f"班级代码{class_code}应该存在但检查结果为不存在")
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
# 测试不存在的班级代码
|
|
non_existent_code = f"NONEXISTENT{int(time.time())}"
|
|
params = {'classCode': non_existent_code}
|
|
response = self.session.get(f"{self.base_url}/xg/classInfo/checkClassCode", params=params)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
exists = data.get('data')
|
|
if not exists:
|
|
self.log_test_result(
|
|
"检查不存在班级代码",
|
|
True,
|
|
f"检查成功，班级代码{non_existent_code}不存在",
|
|
{'classCode': non_existent_code, 'exists': exists}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception(f"班级代码{non_existent_code}不应该存在但检查结果为存在")
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("检查班级代码是否存在", False, f"检查失败: {str(e)}")
|
|
return False
|
|
|
|
def test_update_class_student_count(self):
|
|
"""测试更新班级人数统计"""
|
|
if not self.test_class_id:
|
|
self.log_test_result("更新班级人数统计", False, "没有可用的测试班级ID")
|
|
return False
|
|
|
|
try:
|
|
response = self.session.put(f"{self.base_url}/xg/classInfo/{self.test_class_id}/updateCount")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
self.log_test_result(
|
|
"更新班级人数统计",
|
|
True,
|
|
f"更新成功，班级ID: {self.test_class_id}",
|
|
{'classId': self.test_class_id}
|
|
)
|
|
return True
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("更新班级人数统计", False, f"更新失败: {str(e)}")
|
|
return False
|
|
|
|
def test_delete_class(self):
|
|
"""测试删除班级信息"""
|
|
if not self.test_class_id:
|
|
self.log_test_result("删除班级信息", False, "没有可用的测试班级ID")
|
|
return False
|
|
|
|
try:
|
|
response = self.session.delete(f"{self.base_url}/xg/classInfo/{self.test_class_id}")
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data.get('code') == 200:
|
|
self.log_test_result(
|
|
"删除班级信息",
|
|
True,
|
|
f"删除成功，班级ID: {self.test_class_id}",
|
|
{'classId': self.test_class_id}
|
|
)
|
|
self.test_class_id = None # 清空测试班级ID
|
|
return True
|
|
else:
|
|
raise Exception(f"接口返回错误: {data.get('msg', '未知错误')}")
|
|
else:
|
|
raise Exception(f"HTTP请求失败: {response.status_code}")
|
|
|
|
except Exception as e:
|
|
self.log_test_result("删除班级信息", False, f"删除失败: {str(e)}")
|
|
return False
|
|
|
|
def generate_test_report(self):
|
|
"""生成测试报告"""
|
|
total_tests = len(self.test_results)
|
|
passed_tests = sum(1 for result in self.test_results if result['success'])
|
|
failed_tests = total_tests - passed_tests
|
|
|
|
report = f"""
|
|
==================== 班级信息管理API测试报告 ====================
|
|
|
|
测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
|
测试接口: 班级信息管理相关API
|
|
总测试数: {total_tests}
|
|
通过数量: {passed_tests}
|
|
失败数量: {failed_tests}
|
|
通过率: {(passed_tests/total_tests*100):.1f}%
|
|
|
|
{'='*60}
|
|
详细测试结果:
|
|
"""
|
|
|
|
for i, result in enumerate(self.test_results, 1):
|
|
status = "✅ 通过" if result['success'] else "❌ 失败"
|
|
report += f"\n{i:2d}. [{status}] {result['test_name']}"
|
|
report += f"\n 时间: {result['timestamp']}"
|
|
report += f"\n 结果: {result['message']}"
|
|
if result['response_data']:
|
|
report += f"\n 数据: {json.dumps(result['response_data'], ensure_ascii=False)}"
|
|
report += "\n"
|
|
|
|
report += f"\n{'='*60}"
|
|
|
|
if failed_tests > 0:
|
|
report += f"\n⚠️ 有 {failed_tests} 个测试失败，请检查相关接口！"
|
|
else:
|
|
report += f"\n🎉 所有测试通过！班级信息管理API功能正常。"
|
|
|
|
return report
|
|
|
|
def run_all_tests(self):
|
|
"""运行所有测试"""
|
|
print("开始班级信息管理API测试...")
|
|
print("=" * 80)
|
|
|
|
# 1. 登录
|
|
if not self.login():
|
|
print("❌ 登录失败，停止后续测试")
|
|
return False
|
|
|
|
# 2. 查询班级信息列表
|
|
self.test_get_class_list()
|
|
|
|
# 3. 新增班级信息
|
|
self.test_create_class()
|
|
|
|
# 4. 获取班级详细信息
|
|
self.test_get_class_detail()
|
|
|
|
# 5. 修改班级信息
|
|
self.test_update_class()
|
|
|
|
# 6. 根据班级代码查询
|
|
self.test_get_class_by_code()
|
|
|
|
# 7. 查询学院班级列表
|
|
self.test_get_class_by_college()
|
|
|
|
# 8. 获取班级选项列表
|
|
self.test_get_class_options()
|
|
|
|
# 9. 检查班级代码是否存在
|
|
self.test_check_class_code_exists()
|
|
|
|
# 10. 更新班级人数统计
|
|
self.test_update_class_student_count()
|
|
|
|
# 11. 删除班级信息（最后执行）
|
|
self.test_delete_class()
|
|
|
|
# 生成测试报告
|
|
report = self.generate_test_report()
|
|
print(report)
|
|
|
|
# 保存测试报告到文件
|
|
with open(f"test/班级信息管理API测试报告_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", 'w', encoding='utf-8') as f:
|
|
f.write(report)
|
|
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
tester = ClassInfoAPITester()
|
|
tester.run_all_tests() |