505 lines
17 KiB
Python
505 lines
17 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
邮箱系统性能测试脚本 (轻量版)
|
|
用于测试系统基本功能的响应时间和资源使用情况
|
|
"""
|
|
|
|
import requests
|
|
import smtplib
|
|
import time
|
|
import datetime
|
|
import random
|
|
import string
|
|
import argparse
|
|
import psutil
|
|
import threading
|
|
import json
|
|
import os
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
# 颜色输出
|
|
class Colors:
|
|
GREEN = '\033[92m'
|
|
YELLOW = '\033[93m'
|
|
RED = '\033[91m'
|
|
BLUE = '\033[94m'
|
|
CYAN = '\033[96m'
|
|
BOLD = '\033[1m'
|
|
ENDC = '\033[0m'
|
|
|
|
def print_header(title):
|
|
"""打印格式化的标题"""
|
|
print(f"\n{Colors.BLUE}{Colors.BOLD}{'=' * 60}{Colors.ENDC}")
|
|
print(f"{Colors.BLUE}{Colors.BOLD}{title.center(60)}{Colors.ENDC}")
|
|
print(f"{Colors.BLUE}{Colors.BOLD}{'=' * 60}{Colors.ENDC}\n")
|
|
|
|
def print_info(message):
|
|
"""打印信息消息"""
|
|
print(f"{Colors.CYAN}[INFO] {message}{Colors.ENDC}")
|
|
|
|
def print_success(message):
|
|
"""打印成功消息"""
|
|
print(f"{Colors.GREEN}[SUCCESS] {message}{Colors.ENDC}")
|
|
|
|
def print_warning(message):
|
|
"""打印警告消息"""
|
|
print(f"{Colors.YELLOW}[WARNING] {message}{Colors.ENDC}")
|
|
|
|
def print_error(message):
|
|
"""打印错误消息"""
|
|
print(f"{Colors.RED}[ERROR] {message}{Colors.ENDC}")
|
|
|
|
def generate_random_string(length=8):
|
|
"""生成随机字符串"""
|
|
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
|
|
|
|
def monitor_resources(duration=10, interval=1):
|
|
"""监控系统资源使用"""
|
|
print_info(f"开始监控系统资源使用 ({duration}秒)")
|
|
|
|
# 初始化结果数据
|
|
samples = []
|
|
start_time = time.time()
|
|
end_time = start_time + duration
|
|
|
|
# 收集样本
|
|
while time.time() < end_time:
|
|
# 获取当前进程
|
|
current_process = psutil.Process(os.getpid())
|
|
|
|
# 系统CPU和内存
|
|
cpu_percent = psutil.cpu_percent(interval=0.1)
|
|
memory = psutil.virtual_memory()
|
|
|
|
# 当前进程的CPU和内存
|
|
process_cpu = current_process.cpu_percent(interval=0.1)
|
|
process_memory = current_process.memory_info().rss / 1024 / 1024 # MB
|
|
|
|
# 记录样本
|
|
sample = {
|
|
'timestamp': time.time() - start_time,
|
|
'system_cpu': cpu_percent,
|
|
'system_memory': memory.percent,
|
|
'process_cpu': process_cpu,
|
|
'process_memory': process_memory
|
|
}
|
|
samples.append(sample)
|
|
|
|
# 等待下一个间隔
|
|
time.sleep(interval)
|
|
|
|
# 计算平均值
|
|
if samples:
|
|
avg_system_cpu = sum(s['system_cpu'] for s in samples) / len(samples)
|
|
avg_system_memory = sum(s['system_memory'] for s in samples) / len(samples)
|
|
avg_process_cpu = sum(s['process_cpu'] for s in samples) / len(samples)
|
|
avg_process_memory = sum(s['process_memory'] for s in samples) / len(samples)
|
|
|
|
print_info(f"资源监控完成，平均值:")
|
|
print(f" 系统CPU使用率: {avg_system_cpu:.1f}%")
|
|
print(f" 系统内存使用率: {avg_system_memory:.1f}%")
|
|
print(f" 进程CPU使用率: {avg_process_cpu:.1f}%")
|
|
print(f" 进程内存使用: {avg_process_memory:.1f}MB")
|
|
|
|
return {
|
|
'samples': samples,
|
|
'averages': {
|
|
'system_cpu': avg_system_cpu,
|
|
'system_memory': avg_system_memory,
|
|
'process_cpu': avg_process_cpu,
|
|
'process_memory': avg_process_memory
|
|
}
|
|
}
|
|
|
|
return None
|
|
|
|
def test_api_responsiveness(api_base_url="http://localhost:5000/api", num_requests=10):
|
|
"""测试API响应性能"""
|
|
print_info(f"开始测试API响应性能 ({num_requests}个请求)")
|
|
|
|
endpoints = [
|
|
'/status',
|
|
'/domains',
|
|
'/mailboxes'
|
|
]
|
|
|
|
results = {}
|
|
for endpoint in endpoints:
|
|
url = f"{api_base_url}{endpoint}"
|
|
response_times = []
|
|
success_count = 0
|
|
|
|
print_info(f"测试端点: {endpoint}")
|
|
for i in range(num_requests):
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.get(url, timeout=5)
|
|
end_time = time.time()
|
|
|
|
response_time = (end_time - start_time) * 1000 # 毫秒
|
|
response_times.append(response_time)
|
|
|
|
if response.status_code == 200:
|
|
success_count += 1
|
|
print(f" 请求 {i+1}/{num_requests}: {response_time:.2f}ms (成功)")
|
|
else:
|
|
print(f" 请求 {i+1}/{num_requests}: {response_time:.2f}ms (失败: {response.status_code})")
|
|
except Exception as e:
|
|
print(f" 请求 {i+1}/{num_requests}: 失败 ({str(e)})")
|
|
|
|
# 计算结果
|
|
if response_times:
|
|
avg_time = sum(response_times) / len(response_times)
|
|
min_time = min(response_times)
|
|
max_time = max(response_times)
|
|
success_rate = (success_count / num_requests) * 100
|
|
|
|
results[endpoint] = {
|
|
'avg_time': avg_time,
|
|
'min_time': min_time,
|
|
'max_time': max_time,
|
|
'success_rate': success_rate
|
|
}
|
|
|
|
print(f" 结果: 平均 {avg_time:.2f}ms, 最小 {min_time:.2f}ms, 最大 {max_time:.2f}ms, 成功率 {success_rate:.1f}%")
|
|
else:
|
|
results[endpoint] = {
|
|
'error': '所有请求失败'
|
|
}
|
|
print_error(" 所有请求失败")
|
|
|
|
return results
|
|
|
|
def send_test_email(smtp_host, smtp_port, from_addr, to_addr):
|
|
"""发送测试邮件并计时"""
|
|
message_id = f"{generate_random_string()}@test.local"
|
|
subject = f"测试邮件 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
|
|
|
# 创建邮件
|
|
msg = MIMEMultipart()
|
|
msg['From'] = from_addr
|
|
msg['To'] = to_addr
|
|
msg['Subject'] = subject
|
|
msg['Message-ID'] = message_id
|
|
|
|
# 添加文本内容
|
|
body = f"""
|
|
这是一封测试邮件。
|
|
|
|
发送时间: {datetime.datetime.now().isoformat()}
|
|
唯一ID: {message_id}
|
|
|
|
此邮件用于测试系统性能。
|
|
"""
|
|
msg.attach(MIMEText(body, 'plain'))
|
|
|
|
try:
|
|
# 连接SMTP服务器
|
|
start_time = time.time()
|
|
server = smtplib.SMTP(smtp_host, smtp_port)
|
|
|
|
# 发送邮件
|
|
server.sendmail(from_addr, to_addr, msg.as_string())
|
|
|
|
# 关闭连接
|
|
server.quit()
|
|
end_time = time.time()
|
|
|
|
# 计算响应时间
|
|
response_time = (end_time - start_time) * 1000 # 毫秒
|
|
|
|
return {
|
|
'success': True,
|
|
'response_time': response_time,
|
|
'message_id': message_id,
|
|
'subject': subject
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'error': str(e)
|
|
}
|
|
|
|
def test_email_sending(smtp_host="localhost", smtp_port=3825, num_emails=5, concurrent=False):
|
|
"""测试邮件发送性能"""
|
|
from_addr = "sender@example.com"
|
|
to_addr = "recipient@example.com"
|
|
|
|
print_info(f"开始测试邮件发送性能 ({num_emails}封邮件, {'并发' if concurrent else '顺序'})")
|
|
|
|
results = []
|
|
start_time = time.time()
|
|
|
|
if concurrent:
|
|
# 并发发送
|
|
with ThreadPoolExecutor(max_workers=min(num_emails, 10)) as executor:
|
|
futures = []
|
|
for i in range(num_emails):
|
|
futures.append(executor.submit(
|
|
send_test_email, smtp_host, smtp_port, from_addr, to_addr
|
|
))
|
|
|
|
for i, future in enumerate(futures):
|
|
result = future.result()
|
|
results.append(result)
|
|
|
|
if result['success']:
|
|
print(f" 邮件 {i+1}/{num_emails}: {result['response_time']:.2f}ms (成功)")
|
|
else:
|
|
print(f" 邮件 {i+1}/{num_emails}: 失败 ({result['error']})")
|
|
else:
|
|
# 顺序发送
|
|
for i in range(num_emails):
|
|
result = send_test_email(smtp_host, smtp_port, from_addr, to_addr)
|
|
results.append(result)
|
|
|
|
if result['success']:
|
|
print(f" 邮件 {i+1}/{num_emails}: {result['response_time']:.2f}ms (成功)")
|
|
else:
|
|
print(f" 邮件 {i+1}/{num_emails}: 失败 ({result['error']})")
|
|
|
|
end_time = time.time()
|
|
total_time = (end_time - start_time) * 1000 # 毫秒
|
|
|
|
# 计算结果
|
|
success_count = sum(1 for r in results if r['success'])
|
|
success_rate = (success_count / num_emails) * 100
|
|
|
|
if success_count > 0:
|
|
response_times = [r['response_time'] for r in results if r['success']]
|
|
avg_time = sum(response_times) / len(response_times)
|
|
min_time = min(response_times)
|
|
max_time = max(response_times)
|
|
|
|
print_info(f"邮件发送测试完成:")
|
|
print(f" 成功率: {success_rate:.1f}% ({success_count}/{num_emails})")
|
|
print(f" 平均响应时间: {avg_time:.2f}ms")
|
|
print(f" 最小响应时间: {min_time:.2f}ms")
|
|
print(f" 最大响应时间: {max_time:.2f}ms")
|
|
print(f" 总执行时间: {total_time:.2f}ms")
|
|
|
|
return {
|
|
'success_rate': success_rate,
|
|
'avg_time': avg_time,
|
|
'min_time': min_time,
|
|
'max_time': max_time,
|
|
'total_time': total_time,
|
|
'results': results
|
|
}
|
|
else:
|
|
print_error("所有邮件发送失败")
|
|
return {
|
|
'success_rate': 0,
|
|
'error': '所有发送失败',
|
|
'results': results
|
|
}
|
|
|
|
def test_creating_mailboxes(api_base_url="http://localhost:5000/api", domain_id=1, num_mailboxes=5, concurrent=False):
|
|
"""测试创建邮箱的性能"""
|
|
print_info(f"开始测试创建邮箱性能 ({num_mailboxes}个邮箱, {'并发' if concurrent else '顺序'})")
|
|
|
|
results = []
|
|
start_time = time.time()
|
|
|
|
def create_mailbox(index):
|
|
"""创建单个邮箱"""
|
|
address = f"testuser{generate_random_string()}_{index}"
|
|
data = {
|
|
"domain_id": domain_id,
|
|
"address": address
|
|
}
|
|
|
|
try:
|
|
start_time = time.time()
|
|
response = requests.post(f"{api_base_url}/mailboxes", json=data, timeout=5)
|
|
end_time = time.time()
|
|
|
|
response_time = (end_time - start_time) * 1000 # 毫秒
|
|
|
|
return {
|
|
'success': response.status_code in (200, 201),
|
|
'status_code': response.status_code,
|
|
'response_time': response_time,
|
|
'address': address,
|
|
'response': response.json() if response.status_code in (200, 201) else None
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
'success': False,
|
|
'error': str(e),
|
|
'address': address
|
|
}
|
|
|
|
if concurrent:
|
|
# 并发创建
|
|
with ThreadPoolExecutor(max_workers=min(num_mailboxes, 10)) as executor:
|
|
futures = []
|
|
for i in range(num_mailboxes):
|
|
futures.append(executor.submit(create_mailbox, i))
|
|
|
|
for i, future in enumerate(futures):
|
|
result = future.result()
|
|
results.append(result)
|
|
|
|
if result['success']:
|
|
print(f" 邮箱 {i+1}/{num_mailboxes}: {result['response_time']:.2f}ms (成功)")
|
|
else:
|
|
error_msg = result.get('error', f"状态码: {result.get('status_code')}")
|
|
print(f" 邮箱 {i+1}/{num_mailboxes}: 失败 ({error_msg})")
|
|
else:
|
|
# 顺序创建
|
|
for i in range(num_mailboxes):
|
|
result = create_mailbox(i)
|
|
results.append(result)
|
|
|
|
if result['success']:
|
|
print(f" 邮箱 {i+1}/{num_mailboxes}: {result['response_time']:.2f}ms (成功)")
|
|
else:
|
|
error_msg = result.get('error', f"状态码: {result.get('status_code')}")
|
|
print(f" 邮箱 {i+1}/{num_mailboxes}: 失败 ({error_msg})")
|
|
|
|
end_time = time.time()
|
|
total_time = (end_time - start_time) * 1000 # 毫秒
|
|
|
|
# 计算结果
|
|
success_count = sum(1 for r in results if r['success'])
|
|
success_rate = (success_count / num_mailboxes) * 100
|
|
|
|
if success_count > 0:
|
|
response_times = [r['response_time'] for r in results if r['success']]
|
|
avg_time = sum(response_times) / len(response_times)
|
|
min_time = min(response_times)
|
|
max_time = max(response_times)
|
|
|
|
print_info(f"创建邮箱测试完成:")
|
|
print(f" 成功率: {success_rate:.1f}% ({success_count}/{num_mailboxes})")
|
|
print(f" 平均响应时间: {avg_time:.2f}ms")
|
|
print(f" 最小响应时间: {min_time:.2f}ms")
|
|
print(f" 最大响应时间: {max_time:.2f}ms")
|
|
print(f" 总执行时间: {total_time:.2f}ms")
|
|
|
|
return {
|
|
'success_rate': success_rate,
|
|
'avg_time': avg_time,
|
|
'min_time': min_time,
|
|
'max_time': max_time,
|
|
'total_time': total_time,
|
|
'results': results
|
|
}
|
|
else:
|
|
print_error("所有邮箱创建失败")
|
|
return {
|
|
'success_rate': 0,
|
|
'error': '所有创建失败',
|
|
'results': results
|
|
}
|
|
|
|
def save_results(results, filename="performance_results.json"):
|
|
"""保存测试结果到文件"""
|
|
try:
|
|
with open(filename, 'w') as f:
|
|
json.dump(results, f, indent=2)
|
|
print_success(f"结果已保存到 {filename}")
|
|
return True
|
|
except Exception as e:
|
|
print_error(f"保存结果失败: {str(e)}")
|
|
return False
|
|
|
|
def run_all_tests(args):
|
|
"""运行所有测试"""
|
|
results = {
|
|
'timestamp': datetime.datetime.now().isoformat(),
|
|
'config': vars(args),
|
|
'tests': {}
|
|
}
|
|
|
|
# 测试API响应
|
|
print_header("API响应性能测试")
|
|
api_results = test_api_responsiveness(
|
|
api_base_url=args.api_url,
|
|
num_requests=args.api_requests
|
|
)
|
|
results['tests']['api_responsiveness'] = api_results
|
|
|
|
# 测试邮件发送
|
|
print_header("邮件发送性能测试")
|
|
email_results = test_email_sending(
|
|
smtp_host=args.smtp_host,
|
|
smtp_port=args.smtp_port,
|
|
num_emails=args.email_count,
|
|
concurrent=args.concurrent
|
|
)
|
|
results['tests']['email_sending'] = email_results
|
|
|
|
# 测试创建邮箱
|
|
if not args.skip_mailbox:
|
|
print_header("创建邮箱性能测试")
|
|
mailbox_results = test_creating_mailboxes(
|
|
api_base_url=args.api_url,
|
|
domain_id=args.domain_id,
|
|
num_mailboxes=args.mailbox_count,
|
|
concurrent=args.concurrent
|
|
)
|
|
results['tests']['mailbox_creation'] = mailbox_results
|
|
|
|
# 监控资源使用
|
|
print_header("系统资源监控")
|
|
resource_results = monitor_resources(
|
|
duration=args.monitor_duration,
|
|
interval=1
|
|
)
|
|
results['tests']['resources'] = resource_results
|
|
|
|
# 保存结果
|
|
if args.save:
|
|
save_results(results, args.output)
|
|
|
|
return results
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='邮箱系统性能测试脚本 (轻量版)')
|
|
parser.add_argument('--api-url', default='http://localhost:5000/api', help='API基础URL')
|
|
parser.add_argument('--smtp-host', default='localhost', help='SMTP服务器主机')
|
|
parser.add_argument('--smtp-port', type=int, default=3825, help='SMTP服务器端口')
|
|
parser.add_argument('--domain-id', type=int, default=1, help='用于创建邮箱的域名ID')
|
|
parser.add_argument('--api-requests', type=int, default=10, help='API测试请求数量')
|
|
parser.add_argument('--email-count', type=int, default=5, help='测试邮件数量')
|
|
parser.add_argument('--mailbox-count', type=int, default=5, help='测试创建的邮箱数量')
|
|
parser.add_argument('--monitor-duration', type=int, default=10, help='资源监控持续时间(秒)')
|
|
parser.add_argument('--concurrent', action='store_true', help='启用并发测试')
|
|
parser.add_argument('--skip-mailbox', action='store_true', help='跳过邮箱创建测试')
|
|
parser.add_argument('--save', action='store_true', help='保存测试结果')
|
|
parser.add_argument('--output', default='performance_results.json', help='结果输出文件')
|
|
args = parser.parse_args()
|
|
|
|
print_header("邮箱系统性能测试 (轻量版)")
|
|
print_info("测试配置:")
|
|
print(f" API URL: {args.api_url}")
|
|
print(f" SMTP服务器: {args.smtp_host}:{args.smtp_port}")
|
|
print(f" 域名ID: {args.domain_id}")
|
|
print(f" API请求数: {args.api_requests}")
|
|
print(f" 邮件数量: {args.email_count}")
|
|
print(f" 邮箱数量: {args.mailbox_count}")
|
|
print(f" 并发模式: {'是' if args.concurrent else '否'}")
|
|
print(f" 资源监控时间: {args.monitor_duration}秒")
|
|
|
|
try:
|
|
run_all_tests(args)
|
|
print_header("测试完成")
|
|
return 0
|
|
except KeyboardInterrupt:
|
|
print_warning("\n测试被用户中断")
|
|
return 1
|
|
except Exception as e:
|
|
print_error(f"\n测试过程中出错: {str(e)}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
exit(main())
|