feat: 移动端开发规范重大更新
- 🔥 完全重写 .cursor/cursorrules-mobile.md (940行) * 基于实际mobile工程Vue 3.4.21+uniapp+uview-plus架构 * 移除所有JeeSite相关内容，精准匹配实际项目 * 新增11个完整章节：技术栈、项目结构、页面开发、组件规范等 - 📚 创建完整规范文档 docs/开发文档/移动端开发规范总结.md (800+行) * 涵盖从项目架构到部署的全流程指导 * 提供可直接使用的页面和组件模板 * 包含现代CSS变量设计系统和用户体验优化 - 🔄 同步更新主配置文件 .cursorrules * 更新移动端技术栈描述 * 调整为Vue 3.4.21+uniapp+uview-plus 3.4.45架构 - 📝 完善变更记录 * 详细记录规范更新背景和技术价值 * 为后续维护提供完整历史记录 核心价值：解决规范与实际项目不匹配问题，为Cursor AI提供准确的移动端开发上下文，确保生成代码完全符合实际项目架构。
This commit is contained in:
zlwu committed
1 parent
e8ece1b612
commit
28232d5d46
131 files changed
+17170
-5479
No files matched your search
@@ -0,0 +1,717 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
场地预约管理API测试脚本 (强制AES+RSA双重加密登录版)
|
||||
|
||||
功能说明:
|
||||
1. 测试场地信息管理的完整API
|
||||
2. 测试场地预约管理的完整API
|
||||
3. 测试时间冲突检查功能
|
||||
4. 测试权限控制和业务规则
|
||||
5. 强制使用AES+RSA双重加密登录 (高安全性)
|
||||
|
||||
使用方法:
|
||||
python 场地预约API测试脚本.py
|
||||
|
||||
测试环境:
|
||||
- 后端服务地址: http://localhost/dev-api (按照API测试文档)
|
||||
- 前端地址: http://localhost
|
||||
- 测试用户: admin/admin123
|
||||
|
||||
依赖安装 (必需):
|
||||
pip install requests pycryptodome
|
||||
|
||||
登录方式:
|
||||
⚡ 唯一登录方式: AES+RSA双重加密登录
|
||||
❌ 不支持简化登录
|
||||
❌ 不支持无认证访问
|
||||
|
||||
安全特性:
|
||||
🔐 AES-ECB模式数据加密
|
||||
🛡️ RSA-PKCS1_v1_5密钥加密
|
||||
🔑 32位随机AES密钥
|
||||
📋 标准化认证Headers
|
||||
|
||||
注意事项:
|
||||
- 必须安装pycryptodome加密库，否则脚本将直接退出
|
||||
- 确保后端服务已启动并运行在 http://localhost/dev-api
|
||||
- 严格按照cursorrules-api-test.md文档配置
|
||||
- 登录失败将直接终止测试
|
||||
- 测试完成后会自动清理创建的测试数据
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
import requests
|
||||
from datetime import datetime, timedelta
|
||||
import base64
|
||||
import hashlib
|
||||
import random
|
||||
import string
|
||||
|
||||
# 强制要求加密库，如果失败则退出
|
||||
try:
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Util.Padding import pad
|
||||
from Crypto.PublicKey import RSA
|
||||
from Crypto.Cipher import PKCS1_v1_5
|
||||
CRYPTO_AVAILABLE = True
|
||||
print("✅ 加密库已加载，将使用AES+RSA双重加密登录")
|
||||
except ImportError:
|
||||
CRYPTO_AVAILABLE = False
|
||||
print("❌ 错误：缺少必需的加密库!")
|
||||
print("🔧 请安装加密库：pip install pycryptodome")
|
||||
print("📖 参考文档：.cursor/cursorrules-api-test.md")
|
||||
print("🚫 脚本退出，必须使用加密登录方式")
|
||||
exit(1)
|
||||
|
||||
class VenueReservationAPITest:
|
||||
def __init__(self, base_url="http://localhost/dev-api"):
|
||||
self.base_url = base_url
|
||||
self.session = requests.Session()
|
||||
self.token = None
|
||||
self.user_info = None
|
||||
|
||||
# API测试配置 - 参考cursorrules-api-test.md
|
||||
self.config = {
|
||||
'clientId': 'e5cd7e4891bf95d1d19206ce24a7b32e',
|
||||
'tenantId': '000000',
|
||||
'referer': 'http://localhost/login?redirect=%252Findex',
|
||||
'rsaKey': 'MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBAKoR8mX0rGKLqzcWmOzbfj64K8ZIgOdHnzkXSOVOZbFu/TJhZ7rFAN+eaGkl3C4buccQd/EjEsj9ir7ijT7h96MCAwEAAQ=='
|
||||
}
|
||||
|
||||
# 设置默认header - 按照API测试文档要求
|
||||
self.session.headers.update({
|
||||
'Content-Type': 'application/json;charset=utf-8',
|
||||
'clientid': self.config['clientId'],
|
||||
'referer': self.config['referer'],
|
||||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||||
'isToken': 'false',
|
||||
'isEncrypt': 'false',
|
||||
'repeatSubmit': 'false'
|
||||
})
|
||||
|
||||
def log(self, message):
|
||||
"""输出测试日志"""
|
||||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
print(f"[{timestamp}] {message}")
|
||||
|
||||
def generate_aes_key(self):
|
||||
"""生成32位随机AES密钥"""
|
||||
chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789'
|
||||
return ''.join(random.choices(chars, k=32))
|
||||
|
||||
def aes_encrypt(self, data, key):
|
||||
"""AES加密 - ECB模式"""
|
||||
if not CRYPTO_AVAILABLE:
|
||||
return data # 如果加密库不可用，返回原始数据
|
||||
|
||||
try:
|
||||
# 转换密钥为字节
|
||||
key_bytes = key.encode('utf-8')
|
||||
|
||||
# 填充数据到16字节的倍数
|
||||
data_bytes = data.encode('utf-8')
|
||||
padded_data = pad(data_bytes, AES.block_size)
|
||||
|
||||
# AES-ECB加密
|
||||
cipher = AES.new(key_bytes, AES.MODE_ECB)
|
||||
encrypted_data = cipher.encrypt(padded_data)
|
||||
|
||||
# Base64编码
|
||||
return base64.b64encode(encrypted_data).decode('utf-8')
|
||||
except Exception as e:
|
||||
self.log(f"AES加密失败: {str(e)}")
|
||||
return data
|
||||
|
||||
def rsa_encrypt(self, aes_key):
|
||||
"""RSA加密AES密钥"""
|
||||
if not CRYPTO_AVAILABLE:
|
||||
return base64.b64encode(aes_key.encode('utf-8')).decode('utf-8') # 简化处理
|
||||
|
||||
try:
|
||||
# 构造RSA公钥
|
||||
pem_key = f"-----BEGIN PUBLIC KEY-----\n{self.config['rsaKey']}\n-----END PUBLIC KEY-----"
|
||||
rsa_key = RSA.import_key(pem_key)
|
||||
|
||||
# AES密钥Base64编码
|
||||
aes_key_base64 = base64.b64encode(aes_key.encode('utf-8')).decode('utf-8')
|
||||
|
||||
# RSA加密
|
||||
cipher = PKCS1_v1_5.new(rsa_key)
|
||||
encrypted_data = cipher.encrypt(aes_key_base64.encode('utf-8'))
|
||||
|
||||
# Base64编码
|
||||
return base64.b64encode(encrypted_data).decode('utf-8')
|
||||
except Exception as e:
|
||||
self.log(f"RSA加密失败: {str(e)}")
|
||||
return base64.b64encode(aes_key.encode('utf-8')).decode('utf-8')
|
||||
|
||||
def get_server_info(self):
|
||||
"""获取服务器信息，用于检测API是否可用"""
|
||||
try:
|
||||
self.log(f"正在检测服务可用性: {self.base_url}")
|
||||
|
||||
# 尝试访问一个公开的接口来检测服务是否可用
|
||||
test_endpoints = [
|
||||
"/actuator/health",
|
||||
"/health",
|
||||
"/api/health",
|
||||
"/venue/info/list", # 业务接口
|
||||
"/auth/login" # 登录接口
|
||||
]
|
||||
|
||||
for endpoint in test_endpoints:
|
||||
try:
|
||||
response = self.session.get(f"{self.base_url}{endpoint}", timeout=5)
|
||||
status_code = response.status_code
|
||||
|
||||
# 200正常，401/403说明服务可用但需要认证，404说明接口存在但路径不对
|
||||
if status_code in [200, 401, 403, 404]:
|
||||
self.log(f"✓ 服务可用，测试接口: {endpoint}, 状态码: {status_code}")
|
||||
return True
|
||||
|
||||
except requests.exceptions.ConnectTimeout:
|
||||
self.log(f"连接超时: {endpoint}")
|
||||
continue
|
||||
except requests.exceptions.ConnectionError:
|
||||
continue
|
||||
except Exception as e:
|
||||
continue
|
||||
|
||||
self.log("✗ 无法连接到后端服务")
|
||||
self.log("🔧 请检查:")
|
||||
self.log(" 1. 后端服务是否已启动")
|
||||
self.log(" 2. 服务地址是否正确 (http://localhost/dev-api)")
|
||||
self.log(" 3. 网络连接是否正常")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"检测服务异常: {str(e)}")
|
||||
return False
|
||||
|
||||
def login(self, username="admin", password="admin123"):
|
||||
"""用户登录 - 强制使用AES+RSA双重加密登录"""
|
||||
self.log("🔐 开始AES+RSA双重加密登录...")
|
||||
self.log(f"用户名: {username}")
|
||||
|
||||
try:
|
||||
# 只使用AES+RSA双重加密登录
|
||||
if self._encrypted_login(username, password):
|
||||
return True
|
||||
else:
|
||||
self.log("❌ AES+RSA双重加密登录失败")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"❌ 登录异常: {str(e)}")
|
||||
return False
|
||||
|
||||
def _encrypted_login(self, username, password):
|
||||
"""AES+RSA双重加密登录"""
|
||||
try:
|
||||
self.log("🔐 开始AES+RSA双重加密登录流程...")
|
||||
|
||||
# 生成AES密钥
|
||||
aes_key = self.generate_aes_key()
|
||||
self.log(f"📝 生成AES密钥: {aes_key[:8]}...")
|
||||
|
||||
# 构造登录数据
|
||||
login_data = {
|
||||
"tenantId": self.config['tenantId'],
|
||||
"username": username,
|
||||
"password": password,
|
||||
"clientId": self.config['clientId'],
|
||||
"grantType": "password"
|
||||
}
|
||||
self.log(f"📋 登录数据: username={username}, tenantId={login_data['tenantId']}, clientId={login_data['clientId']}")
|
||||
|
||||
# AES加密登录数据
|
||||
login_json = json.dumps(login_data, separators=(',', ':'))
|
||||
encrypted_data = self.aes_encrypt(login_json, aes_key)
|
||||
self.log(f"🔒 AES加密完成，数据长度: {len(encrypted_data)}")
|
||||
|
||||
# RSA加密AES密钥
|
||||
encrypted_key = self.rsa_encrypt(aes_key)
|
||||
self.log(f"🛡️ RSA加密密钥完成，长度: {len(encrypted_key)}")
|
||||
|
||||
# 设置加密登录的特殊headers
|
||||
headers = {
|
||||
'Content-Type': 'application/json;charset=utf-8',
|
||||
'clientid': self.config['clientId'],
|
||||
'encrypt-key': encrypted_key,
|
||||
'referer': self.config['referer'],
|
||||
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
|
||||
'isToken': 'false',
|
||||
'isEncrypt': 'true',
|
||||
'repeatSubmit': 'false'
|
||||
}
|
||||
|
||||
login_url = f"{self.base_url}/auth/login"
|
||||
self.log(f"🌐 发送登录请求: {login_url}")
|
||||
self.log(f"📤 Headers: clientid={headers['clientid']}, isEncrypt={headers['isEncrypt']}")
|
||||
|
||||
# 发送登录请求
|
||||
response = self.session.post(
|
||||
login_url,
|
||||
data=encrypted_data,
|
||||
headers=headers,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
self.log(f"📥 响应状态码: {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
try:
|
||||
result = response.json()
|
||||
self.log(f"📊 响应结果: code={result.get('code')}, msg={result.get('msg')}")
|
||||
|
||||
if result.get('code') == 200 and result.get('data'):
|
||||
# 获取Token
|
||||
token_fields = ['access_token', 'token', 'accessToken']
|
||||
for field in token_fields:
|
||||
if field in result['data']:
|
||||
self.token = result['data'][field]
|
||||
break
|
||||
|
||||
if self.token:
|
||||
self.user_info = result['data']
|
||||
|
||||
# 设置认证header
|
||||
self.session.headers.update({
|
||||
'Authorization': f'Bearer {self.token}',
|
||||
'isToken': 'true',
|
||||
'isEncrypt': 'false'
|
||||
})
|
||||
|
||||
self.log("✅ AES+RSA双重加密登录成功!")
|
||||
self.log(f"👤 用户: {username}")
|
||||
self.log(f"🎫 Token: {self.token[:50]}...")
|
||||
if 'username' in result['data']:
|
||||
self.log(f"📋 用户信息: {result['data']['username']}")
|
||||
return True
|
||||
else:
|
||||
self.log("❌ 登录失败: 响应中未找到Token字段")
|
||||
self.log(f"📊 可用字段: {list(result.get('data', {}).keys())}")
|
||||
else:
|
||||
self.log(f"❌ 登录失败: {result.get('msg', '未知错误')}")
|
||||
if result.get('code'):
|
||||
self.log(f"📊 错误代码: {result['code']}")
|
||||
|
||||
except json.JSONDecodeError:
|
||||
self.log("❌ 响应不是有效的JSON格式")
|
||||
self.log(f"📝 响应内容: {response.text[:200]}...")
|
||||
|
||||
else:
|
||||
self.log(f"❌ HTTP请求失败: {response.status_code}")
|
||||
self.log(f"📝 响应内容: {response.text[:200]}...")
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"❌ 加密登录异常: {str(e)}")
|
||||
import traceback
|
||||
self.log(f"🔍 异常详情: {traceback.format_exc()}")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
|
||||
def api_request(self, method, url, data=None, params=None):
|
||||
"""API请求封装"""
|
||||
try:
|
||||
full_url = f"{self.base_url}{url}"
|
||||
|
||||
if method.upper() == 'GET':
|
||||
response = self.session.get(full_url, params=params)
|
||||
elif method.upper() == 'POST':
|
||||
response = self.session.post(full_url, json=data, params=params)
|
||||
elif method.upper() == 'PUT':
|
||||
response = self.session.put(full_url, json=data)
|
||||
elif method.upper() == 'DELETE':
|
||||
response = self.session.delete(full_url)
|
||||
else:
|
||||
raise ValueError(f"不支持的HTTP方法: {method}")
|
||||
|
||||
# 记录请求日志
|
||||
self.log(f"{method} {url} - HTTP {response.status_code}")
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
return result
|
||||
else:
|
||||
self.log(f"请求失败: {response.text}")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"请求异常: {str(e)}")
|
||||
return None
|
||||
|
||||
def test_venue_management(self):
|
||||
"""测试场地信息管理"""
|
||||
self.log("=" * 50)
|
||||
self.log("开始测试场地信息管理API")
|
||||
|
||||
# 1. 查询场地列表
|
||||
self.log("1. 测试查询场地列表")
|
||||
result = self.api_request('GET', '/venue/info/list', params={'pageNum': 1, 'pageSize': 10})
|
||||
if result and result.get('code') == 200:
|
||||
self.log(f"✓ 查询场地列表成功，共{result.get('total', 0)}条数据")
|
||||
else:
|
||||
self.log("✗ 查询场地列表失败")
|
||||
|
||||
# 2. 获取可用场地列表
|
||||
self.log("2. 测试获取可用场地列表")
|
||||
result = self.api_request('GET', '/venue/info/available')
|
||||
available_venues = []
|
||||
if result and result.get('code') == 200:
|
||||
available_venues = result.get('data', [])
|
||||
self.log(f"✓ 获取可用场地列表成功，共{len(available_venues)}个可用场地")
|
||||
else:
|
||||
self.log("✗ 获取可用场地列表失败")
|
||||
|
||||
# 3. 新增场地信息
|
||||
self.log("3. 测试新增场地信息")
|
||||
test_venue_code = f"TEST{int(time.time())}"
|
||||
venue_data = {
|
||||
"venueName": "测试会议室",
|
||||
"venueCode": test_venue_code,
|
||||
"venueLocation": "测试楼1楼",
|
||||
"venueCapacity": 20,
|
||||
"venueType": "meeting_room",
|
||||
"venueStatus": "1",
|
||||
"venueDescription": "自动化测试创建的会议室",
|
||||
"venueManagerName": "测试管理员"
|
||||
}
|
||||
|
||||
result = self.api_request('POST', '/venue/info', data=venue_data)
|
||||
venue_id = None
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 新增场地信息成功")
|
||||
|
||||
# 查询刚创建的场地获取ID - 通过列表查询带编码过滤
|
||||
time.sleep(1) # 等待1秒确保数据已保存
|
||||
list_result = self.api_request('GET', '/venue/info/list', params={
|
||||
'venueCode': test_venue_code,
|
||||
'pageNum': 1,
|
||||
'pageSize': 1
|
||||
})
|
||||
if list_result and list_result.get('code') == 200 and list_result.get('rows'):
|
||||
venue_id = list_result['rows'][0]['id']
|
||||
self.log(f"✓ 获取新创建场地ID: {venue_id}")
|
||||
else:
|
||||
self.log("✗ 无法获取新创建场地的ID")
|
||||
else:
|
||||
self.log("✗ 新增场地信息失败")
|
||||
|
||||
# 4. 查询场地详情
|
||||
if venue_id:
|
||||
self.log("4. 测试查询场地详情")
|
||||
result = self.api_request('GET', f'/venue/info/{venue_id}')
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 查询场地详情成功")
|
||||
else:
|
||||
self.log("✗ 查询场地详情失败")
|
||||
|
||||
# 5. 修改场地信息
|
||||
if venue_id:
|
||||
self.log("5. 测试修改场地信息")
|
||||
update_data = venue_data.copy()
|
||||
update_data['id'] = venue_id
|
||||
update_data['venueName'] = "测试会议室(已修改)"
|
||||
update_data['venueCapacity'] = 25
|
||||
|
||||
result = self.api_request('PUT', '/venue/info', data=update_data)
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 修改场地信息成功")
|
||||
else:
|
||||
self.log("✗ 修改场地信息失败")
|
||||
|
||||
return venue_id, available_venues
|
||||
|
||||
def test_reservation_management(self, venue_id=None, available_venues=None):
|
||||
"""测试场地预约管理"""
|
||||
self.log("=" * 50)
|
||||
self.log("开始测试场地预约管理API")
|
||||
|
||||
# 选择测试用的场地ID
|
||||
test_venue_id = venue_id
|
||||
if not test_venue_id and available_venues:
|
||||
test_venue_id = available_venues[0]['id']
|
||||
|
||||
if not test_venue_id:
|
||||
self.log("✗ 没有可用的场地进行预约测试")
|
||||
return None
|
||||
|
||||
# 1. 查询预约列表
|
||||
self.log("1. 测试查询预约列表")
|
||||
result = self.api_request('GET', '/venue/reservation/list', params={'pageNum': 1, 'pageSize': 10})
|
||||
if result and result.get('code') == 200:
|
||||
self.log(f"✓ 查询预约列表成功，共{result.get('total', 0)}条数据")
|
||||
else:
|
||||
self.log("✗ 查询预约列表失败")
|
||||
|
||||
# 2. 查询我的预约列表
|
||||
self.log("2. 测试查询我的预约列表")
|
||||
result = self.api_request('GET', '/venue/reservation/my', params={'pageNum': 1, 'pageSize': 10})
|
||||
if result and result.get('code') == 200:
|
||||
self.log(f"✓ 查询我的预约列表成功，共{result.get('total', 0)}条数据")
|
||||
else:
|
||||
self.log("✗ 查询我的预约列表失败")
|
||||
|
||||
# 3. 时间冲突检查
|
||||
self.log("3. 测试时间冲突检查")
|
||||
tomorrow = datetime.now() + timedelta(days=1)
|
||||
start_time = tomorrow.replace(hour=14, minute=0, second=0, microsecond=0)
|
||||
end_time = tomorrow.replace(hour=16, minute=0, second=0, microsecond=0)
|
||||
|
||||
conflict_params = {
|
||||
'venueId': test_venue_id,
|
||||
'startTime': start_time.strftime('%Y-%m-%d %H:%M:%S'),
|
||||
'endTime': end_time.strftime('%Y-%m-%d %H:%M:%S')
|
||||
}
|
||||
|
||||
result = self.api_request('GET', '/venue/reservation/conflict', params=conflict_params)
|
||||
if result and result.get('code') == 200:
|
||||
conflicts = result.get('data', [])
|
||||
self.log(f"✓ 时间冲突检查成功，发现{len(conflicts)}个冲突")
|
||||
else:
|
||||
self.log("✗ 时间冲突检查失败")
|
||||
|
||||
# 4. 新增预约
|
||||
self.log("4. 测试新增预约")
|
||||
reservation_data = {
|
||||
"venueId": test_venue_id,
|
||||
"applicantPhone": "13800138000",
|
||||
"reservationDate": tomorrow.strftime('%Y-%m-%d'),
|
||||
"startTime": start_time.strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"endTime": end_time.strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"reservationPurpose": "自动化测试预约",
|
||||
"participantCount": 10,
|
||||
"specialRequirements": "测试用预约，无特殊需求"
|
||||
}
|
||||
|
||||
result = self.api_request('POST', '/venue/reservation', data=reservation_data)
|
||||
reservation_id = None
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 新增预约成功")
|
||||
|
||||
# 查询我的预约获取ID
|
||||
result = self.api_request('GET', '/venue/reservation/my', params={'pageNum': 1, 'pageSize': 1})
|
||||
if result and result.get('code') == 200 and result.get('rows'):
|
||||
reservation_id = result['rows'][0]['id']
|
||||
self.log(f"✓ 获取新创建预约ID: {reservation_id}")
|
||||
else:
|
||||
self.log("✗ 新增预约失败")
|
||||
|
||||
# 5. 查询预约详情
|
||||
if reservation_id:
|
||||
self.log("5. 测试查询预约详情")
|
||||
result = self.api_request('GET', f'/venue/reservation/{reservation_id}')
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 查询预约详情成功")
|
||||
else:
|
||||
self.log("✗ 查询预约详情失败")
|
||||
|
||||
# 6. 修改预约
|
||||
if reservation_id:
|
||||
self.log("6. 测试修改预约")
|
||||
update_reservation_data = reservation_data.copy()
|
||||
update_reservation_data['id'] = reservation_id
|
||||
update_reservation_data['reservationPurpose'] = "自动化测试预约(已修改)"
|
||||
update_reservation_data['participantCount'] = 15
|
||||
|
||||
result = self.api_request('PUT', '/venue/reservation', data=update_reservation_data)
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 修改预约成功")
|
||||
else:
|
||||
self.log("✗ 修改预约失败")
|
||||
|
||||
# 7. 查询待审批预约列表
|
||||
self.log("7. 测试查询待审批预约列表")
|
||||
result = self.api_request('GET', '/venue/reservation/pending', params={'pageNum': 1, 'pageSize': 10})
|
||||
if result and result.get('code') == 200:
|
||||
self.log(f"✓ 查询待审批预约列表成功，共{result.get('total', 0)}条数据")
|
||||
else:
|
||||
self.log("✗ 查询待审批预约列表失败")
|
||||
|
||||
# 8. 审批预约
|
||||
if reservation_id:
|
||||
self.log("8. 测试审批预约")
|
||||
approve_data = {
|
||||
"id": reservation_id,
|
||||
"reservationStatus": "1", # 1-已批准
|
||||
"approveRemark": "自动化测试审批通过"
|
||||
}
|
||||
|
||||
result = self.api_request('POST', '/venue/reservation/approve', data=approve_data)
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 审批预约成功")
|
||||
else:
|
||||
self.log("✗ 审批预约失败")
|
||||
|
||||
return reservation_id
|
||||
|
||||
def test_business_rules(self, venue_id=None):
|
||||
"""测试业务规则"""
|
||||
self.log("=" * 50)
|
||||
self.log("开始测试业务规则")
|
||||
|
||||
if not venue_id:
|
||||
self.log("✗ 没有可用的场地进行业务规则测试")
|
||||
return
|
||||
|
||||
# 1. 测试预约过去时间（应该失败）
|
||||
self.log("1. 测试预约过去时间")
|
||||
yesterday = datetime.now() - timedelta(days=1)
|
||||
past_reservation_data = {
|
||||
"venueId": venue_id,
|
||||
"applicantPhone": "13800138000",
|
||||
"reservationDate": yesterday.strftime('%Y-%m-%d'),
|
||||
"startTime": yesterday.strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"endTime": (yesterday + timedelta(hours=2)).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"reservationPurpose": "测试过去时间预约",
|
||||
"participantCount": 5
|
||||
}
|
||||
|
||||
result = self.api_request('POST', '/venue/reservation', data=past_reservation_data)
|
||||
if result and result.get('code') != 200:
|
||||
self.log("✓ 预约过去时间被正确拒绝")
|
||||
else:
|
||||
self.log("✗ 预约过去时间应该被拒绝")
|
||||
|
||||
# 2. 测试开始时间晚于结束时间（应该失败）
|
||||
self.log("2. 测试开始时间晚于结束时间")
|
||||
tomorrow = datetime.now() + timedelta(days=1)
|
||||
invalid_time_data = {
|
||||
"venueId": venue_id,
|
||||
"applicantPhone": "13800138000",
|
||||
"reservationDate": tomorrow.strftime('%Y-%m-%d'),
|
||||
"startTime": tomorrow.replace(hour=16).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"endTime": tomorrow.replace(hour=14).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
"reservationPurpose": "测试无效时间",
|
||||
"participantCount": 5
|
||||
}
|
||||
|
||||
result = self.api_request('POST', '/venue/reservation', data=invalid_time_data)
|
||||
if result and result.get('code') != 200:
|
||||
self.log("✓ 无效时间范围被正确拒绝")
|
||||
else:
|
||||
self.log("✗ 无效时间范围应该被拒绝")
|
||||
|
||||
def cleanup_test_data(self, venue_id=None, reservation_id=None):
|
||||
"""清理测试数据"""
|
||||
self.log("=" * 50)
|
||||
self.log("开始清理测试数据")
|
||||
|
||||
# 删除测试预约
|
||||
if reservation_id:
|
||||
result = self.api_request('DELETE', f'/venue/reservation/{reservation_id}')
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 删除测试预约成功")
|
||||
else:
|
||||
self.log("✗ 删除测试预约失败")
|
||||
|
||||
# 删除测试场地
|
||||
if venue_id:
|
||||
result = self.api_request('DELETE', f'/venue/info/{venue_id}')
|
||||
if result and result.get('code') == 200:
|
||||
self.log("✓ 删除测试场地成功")
|
||||
else:
|
||||
self.log("✗ 删除测试场地失败")
|
||||
|
||||
def run_all_tests(self):
|
||||
"""运行所有测试"""
|
||||
self.log("场地预约管理API自动化测试开始")
|
||||
self.log(f"测试环境: {self.base_url}")
|
||||
|
||||
start_time = time.time()
|
||||
|
||||
# 1. 检测服务可用性
|
||||
if not self.get_server_info():
|
||||
self.log("后端服务不可用，请确保服务已启动，测试终止")
|
||||
return False
|
||||
|
||||
# 2. 强制加密登录
|
||||
if not self.login():
|
||||
self.log("❌ AES+RSA双重加密登录失败，测试终止")
|
||||
self.log("💡 请检查:")
|
||||
self.log(" 1. 用户名密码是否正确 (admin/admin123)")
|
||||
self.log(" 2. 后端服务是否正常运行")
|
||||
self.log(" 3. 加密库是否正确安装 (pycryptodome)")
|
||||
self.log(" 4. 参考API测试文档配置")
|
||||
return False
|
||||
|
||||
venue_id = None
|
||||
reservation_id = None
|
||||
available_venues = None
|
||||
|
||||
try:
|
||||
# 2. 测试场地管理
|
||||
venue_id, available_venues = self.test_venue_management()
|
||||
|
||||
# 3. 测试预约管理
|
||||
reservation_id = self.test_reservation_management(venue_id, available_venues)
|
||||
|
||||
# 4. 测试业务规则
|
||||
self.test_business_rules(venue_id)
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"测试过程中发生异常: {str(e)}")
|
||||
|
||||
finally:
|
||||
# 5. 清理测试数据
|
||||
self.cleanup_test_data(venue_id, reservation_id)
|
||||
|
||||
end_time = time.time()
|
||||
duration = round(end_time - start_time, 2)
|
||||
|
||||
self.log("=" * 50)
|
||||
self.log(f"场地预约管理API自动化测试完成，耗时 {duration} 秒")
|
||||
|
||||
return True
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("🚀 场地预约管理API自动化测试脚本 (强制加密登录版)")
|
||||
print("=" * 70)
|
||||
print("📋 功能特性:")
|
||||
print("1. 🔐 强制使用AES+RSA双重加密登录 (严格按照API测试文档)")
|
||||
print("2. 🛡️ 高安全性认证，拒绝简化登录")
|
||||
print("3. 🌐 标准API接口地址 (http://localhost/dev-api)")
|
||||
print("4. 🧪 完整的场地预约功能测试")
|
||||
print("5. 🧹 自动数据清理")
|
||||
print("=" * 70)
|
||||
print("⚠️ 重要提醒:")
|
||||
print(" 🔧 必须安装加密库: pip install pycryptodome")
|
||||
print(" 🔐 必须使用AES+RSA双重加密登录")
|
||||
print(" 📖 参考文档: .cursor/cursorrules-api-test.md")
|
||||
print("=" * 70)
|
||||
|
||||
# 创建测试实例
|
||||
tester = VenueReservationAPITest()
|
||||
|
||||
# 运行测试
|
||||
success = tester.run_all_tests()
|
||||
|
||||
if success:
|
||||
print("\n✅ 测试执行完成！请查看上方日志了解测试结果。")
|
||||
print("\n📊 测试结果说明:")
|
||||
print(" ✅ = 测试通过")
|
||||
print(" ❌ = 测试失败")
|
||||
print(" 🔓 = 无认证访问")
|
||||
print(" 🔐 = 加密登录")
|
||||
else:
|
||||
print("\n❌ 测试执行失败！")
|
||||
print("\n🔧 故障排除:")
|
||||
print("1. 确保后端服务运行在 http://localhost/dev-api")
|
||||
print("2. 检查网络连接")
|
||||
print("3. 安装完整依赖: pip install requests pycryptodome")
|
||||
print("4. 参考 cursorrules-api-test.md 文档")
|
||||
return 1
|
||||
|
||||
return 0
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("🔍 参考文档: .cursor/cursorrules-api-test.md")
|
||||
print("📧 基于API测试文档优化，支持AES+RSA双重加密登录\n")
|
||||
exit(main())
|
||||
Reference in new issue
Block a user