我使用docker compose搭建了8.3.1版本的domjudge评测系统，准备举行比赛，在此之前，我使用domjudge系统的api准备模拟用户并行提交进行压力测试，presstest.py是我的压测脚本，使用API登录成功了，但使用api提交代码是返回500错误，请问是为什么
#!/usr/bin/env python3
"""
DOMjudge 压力测试脚本
用于模拟多个用户同时提交代码
"""
import requests
import threading
import time
import random
import json
from requests.auth import HTTPBasicAuth
import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
class DOMjudgeStressTester:
def __init__(self, base_url, username_prefix, password, num_users=200):
self.base_url = base_url.rstrip('/')
self.username_prefix = username_prefix
self.password = password
self.num_users = num_users
self.results = []
self.lock = threading.Lock()
def submit_solution(self, user_id, problem_id, language_id, source_code):
"""提交解决方案"""
username = f"{self.username_prefix}{user_id+8:03d}"
# DOMjudge API 端点
submit_url = f"{self.base_url}/api/v4/contests/3/submissions"
# 准备提交数据
payload = {
"problem_id": problem_id,
"language_id": language_id,
"code": source_code
}
headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
auth = HTTPBasicAuth(username, self.password)
start_time = time.time()
try:
response = requests.post(
submit_url,
json=payload,
headers=headers,
auth=auth,
timeout=30
)
end_time = time.time()
response_time = end_time - start_time
if response.status_code == 200:
result = {
"user_id": user_id,
"username": username,
"status": "success",
"response_time": response_time,
"submission_id": response.json().get("id", "unknown")
}
print(f"✓ 用户 {username} 提交成功 (响应时间: {response_time:.2f}s)")
else:
result = {
"user_id": user_id,
"username": username,
"status": f"failed - HTTP {response.status_code}",
"response_time": response_time,
"error": response.text
}
print(f"✗ 用户 {username} 提交失败: {response.status_code}")
print(f"错误信息: {response.text}")
except requests.exceptions.RequestException as e:
end_time = time.time()
response_time = end_time - start_time
result = {
"user_id": user_id,
"username": username,
"status": f"error - {str(e)}",
"response_time": response_time
}
print(f"✗ 用户 {username} 请求异常: {str(e)}")
with self.lock:
self.results.append(result)
return result
def worker(self, user_id, problem_id, language_id, source_code):
"""工作线程函数"""
time.sleep(random.uniform(0, 2)) # 随机延迟，模拟真实用户行为
return self.submit_solution(user_id, problem_id, language_id, source_code)
def run_stress_test(self, problem_id, language_id, source_code_file):
"""运行压力测试"""
print(f"开始压力测试: {self.num_users} 个用户同时提交")
print("=" * 50)
# 读取源代码文件
try:
with open(source_code_file, 'r') as f:
source_code = f.read()
except FileNotFoundError:
print(f"错误: 找不到源代码文件 {source_code_file}")
return
# 创建并启动线程
threads = []
start_time = time.time()
for i in range(self.num_users):
thread = threading.Thread(
target=self.worker,
args=(i + 1, problem_id, language_id, source_code)
)
threads.append(thread)
thread.start()
# 等待所有线程完成
for thread in threads:
thread.join()
total_time = time.time() - start_time
# 打印统计结果
self.print_statistics(total_time)
def print_statistics(self, total_time):
"""打印统计信息"""
print("\n" + "=" * 50)
print("压力测试完成!")
print("=" * 50)
successful = len([r for r in self.results if r["status"] == "success"])
failed = len(self.results) - successful
print(f"总用户数: {self.num_users}")
print(f"成功提交: {successful}")
print(f"失败提交: {failed}")
print(f"总耗时: {total_time:.2f} 秒")
if successful > 0:
response_times = [r["response_time"] for r in self.results if r["status"] == "success"]
avg_response_time = sum(response_times) / len(response_times)
max_response_time = max(response_times)
min_response_time = min(response_times)
print(f"平均响应时间: {avg_response_time:.2f} 秒")
print(f"最小响应时间: {min_response_time:.2f} 秒")
print(f"最大响应时间: {max_response_time:.2f} 秒")
print(f"吞吐量: {successful / total_time:.2f} 提交/秒")
if __name__ == "__main__":
# 配置参数 - 请根据你的实际情况修改
CONFIG = {
"base_url": "http://192.168.53.202:8080", # DOMjudge 地址
"username_prefix": "team", # 用户名前缀
"password": "123456", # 用户密码
"num_users": 10, # 用户数量
"problem_id": "8", # 题目ID
"language_id": "cpp", # 语言ID (cpp, java, python等)
"source_code_file": "E://press/1.cpp" # 源代码文件路径
}
# 创建测试器并运行
tester = DOMjudgeStressTester(
base_url=CONFIG["base_url"],
username_prefix=CONFIG["username_prefix"],
password=CONFIG["password"],
num_users=CONFIG["num_users"]
)
tester.run_stress_test(
problem_id=CONFIG["problem_id"],
language_id=CONFIG["language_id"],
source_code_file=CONFIG["source_code_file"]
)
