将注释翻译为中文以提高代码可读性
This commit is contained in:
longpeng
parent
9aa99178ab
commit
c4d6e553ed
@ -6,9 +6,9 @@ import asyncio
|
||||
|
||||
from temporalio import activity
|
||||
|
||||
# 确保能导入 gen 模块
|
||||
# 确保能导入 gen 模块，将gen目录添加到Python路径
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'gen')))
|
||||
# 全局变量来存储 protobuf 模块
|
||||
# 导入protobuf生成的模块和其他依赖
|
||||
from gen import common_test_pb2 as pb
|
||||
from api_tests import execute_api_test_case
|
||||
from ui_tests import execute_ui_test_case
|
||||
@ -16,112 +16,157 @@ from utils import upload_file_to_s3, scalar_map_to_dict
|
||||
|
||||
|
||||
class TestActivities:
|
||||
"""
|
||||
测试活动类，包含API测试和UI测试的Temporal Activity实现
|
||||
"""
|
||||
|
||||
async def _heartbeat_task(self, interval_seconds=30):
|
||||
"""心跳任务，定期发送心跳信号"""
|
||||
"""
|
||||
心跳任务，定期发送心跳信号，防止长时间运行的Activity被Temporal服务器认为已死亡
|
||||
|
||||
Args:
|
||||
interval_seconds (int): 心跳发送间隔，默认30秒
|
||||
"""
|
||||
while True:
|
||||
try:
|
||||
# 等待指定间隔时间
|
||||
await asyncio.sleep(interval_seconds)
|
||||
# 发送心跳信号告知Temporal服务器Activity仍在运行
|
||||
activity.heartbeat()
|
||||
activity.logger.debug("Activity heartbeat sent")
|
||||
except asyncio.CancelledError:
|
||||
# 心跳任务被取消，正常退出
|
||||
activity.logger.debug("Heartbeat task cancelled")
|
||||
break
|
||||
except Exception as e:
|
||||
# 心跳发送失败，记录警告但继续尝试
|
||||
activity.logger.warning(f"Failed to send heartbeat: {e}")
|
||||
|
||||
@activity.defn(name="run_api_test")
|
||||
async def run_api_test(self, req: pb.ApiTestRequest) -> pb.ApiTestResult:
|
||||
"""执行API测试的Temporal Activity实现"""
|
||||
"""
|
||||
执行API测试的Temporal Activity实现
|
||||
|
||||
Args:
|
||||
req (pb.ApiTestRequest): API测试请求对象，包含测试用例ID、端点、HTTP方法等信息
|
||||
|
||||
Returns:
|
||||
pb.ApiTestResult: API测试结果对象，包含测试状态、响应数据等信息
|
||||
"""
|
||||
activity.logger.info(f"Received API Test Request: {req.test_case_id}")
|
||||
# 记录测试开始时间，用于计算执行时长
|
||||
start_time = time.time()
|
||||
# 初始化测试结果对象
|
||||
result = pb.ApiTestResult()
|
||||
result.base_result.test_case_id = req.test_case_id
|
||||
|
||||
# 启动心跳任务
|
||||
# 启动后台心跳任务，确保长时间运行的测试不会超时
|
||||
heartbeat_task = asyncio.create_task(self._heartbeat_task())
|
||||
|
||||
try:
|
||||
# 发送初始心跳
|
||||
# 发送初始心跳信号
|
||||
activity.heartbeat()
|
||||
|
||||
# 调用实际的API测试逻辑
|
||||
# 调用实际的API测试逻辑，执行HTTP请求并验证响应
|
||||
api_test_success, actual_status, response_body, log_output = execute_api_test_case(
|
||||
req.test_case_id, req.endpoint, req.http_method, scalar_map_to_dict(req.headers), req.request_body,
|
||||
req.expected_status_code
|
||||
)
|
||||
|
||||
# 填充测试结果
|
||||
result.base_result.success = api_test_success
|
||||
result.actual_status_code = actual_status
|
||||
# 处理响应体，确保为字符串格式
|
||||
result.response_body = response_body.decode('utf-8') if isinstance(response_body, bytes) else str(response_body)
|
||||
result.base_result.log_output = log_output
|
||||
result.base_result.message = "API Test Passed" if api_test_success else "API Test Failed"
|
||||
|
||||
except Exception as e:
|
||||
# 捕获测试执行过程中的异常
|
||||
activity.logger.error(f"API Test Failed for {req.test_case_id}: {e}")
|
||||
result.base_result.success = False
|
||||
result.base_result.message = f"API Test Error: {e}"
|
||||
result.base_result.error_details = str(e)
|
||||
finally:
|
||||
# 取消心跳任务
|
||||
# 清理工作：取消心跳任务
|
||||
heartbeat_task.cancel()
|
||||
try:
|
||||
# 等待心跳任务完全结束
|
||||
await heartbeat_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
# 计算并记录测试执行时长
|
||||
result.base_result.duration_seconds = time.time() - start_time
|
||||
return result
|
||||
|
||||
@activity.defn(name="run_ui_test")
|
||||
async def run_ui_test(self, req: pb.UiTestRequest) -> pb.UiTestResult:
|
||||
"""执行UI测试的Temporal Activity实现"""
|
||||
"""
|
||||
执行UI测试的Temporal Activity实现
|
||||
|
||||
Args:
|
||||
req (pb.UiTestRequest): UI测试请求对象，包含测试用例ID、URL路径、浏览器类型等信息
|
||||
|
||||
Returns:
|
||||
pb.UiTestResult: UI测试结果对象，包含测试状态、截图URL、报告URL等信息
|
||||
"""
|
||||
activity.logger.info(f"Received UI Test Request: {req.test_case_id}")
|
||||
# 记录测试开始时间
|
||||
start_time = time.time()
|
||||
# 初始化测试结果对象
|
||||
result = pb.UiTestResult()
|
||||
result.base_result.test_case_id = req.test_case_id
|
||||
|
||||
# 启动心跳任务
|
||||
# 启动后台心跳任务
|
||||
heartbeat_task = asyncio.create_task(self._heartbeat_task())
|
||||
|
||||
try:
|
||||
# 发送初始心跳
|
||||
# 发送初始心跳信号
|
||||
activity.heartbeat()
|
||||
|
||||
# 调用实际的UI测试逻辑，返回本地文件路径
|
||||
# 调用实际的UI测试逻辑，执行浏览器自动化测试并返回本地文件路径
|
||||
ui_test_success, log_output, screenshot_path, html_report_path = await execute_ui_test_case(
|
||||
req.test_case_id, req.url_path, req.browser_type, req.headless, scalar_map_to_dict(req.user_data)
|
||||
)
|
||||
|
||||
# 填充基本测试结果
|
||||
result.base_result.success = ui_test_success
|
||||
result.base_result.log_output = log_output
|
||||
result.base_result.message = "UI Test Passed" if ui_test_success else "UI Test Failed"
|
||||
|
||||
# 上传截图和报告到对象存储，并返回URL
|
||||
# 处理测试生成的文件：上传截图和报告到对象存储，并返回URL
|
||||
if screenshot_path and os.path.exists(screenshot_path):
|
||||
# 在长时间操作前发送心跳
|
||||
# 在长时间操作前发送心跳，防止超时
|
||||
activity.heartbeat()
|
||||
# 上传截图到S3并获取访问URL
|
||||
result.screenshot_url = await upload_file_to_s3(screenshot_path, f"screenshots/{req.test_case_id}.png")
|
||||
os.remove(screenshot_path) # 清理本地文件
|
||||
# 清理本地临时文件
|
||||
os.remove(screenshot_path)
|
||||
|
||||
if html_report_path and os.path.exists(html_report_path):
|
||||
# 在长时间操作前发送心跳
|
||||
activity.heartbeat()
|
||||
# 上传HTML报告到S3并获取访问URL
|
||||
result.html_report_url = await upload_file_to_s3(html_report_path, f"reports/{req.test_case_id}.html")
|
||||
os.remove(html_report_path) # 清理本地文件
|
||||
# 清理本地临时文件
|
||||
os.remove(html_report_path)
|
||||
|
||||
except Exception as e:
|
||||
# 捕获UI测试执行过程中的异常
|
||||
activity.logger.error(f"UI Test Failed for {req.test_case_id}: {e}")
|
||||
result.base_result.success = False
|
||||
result.base_result.message = f"UI Test Error: {e}"
|
||||
result.base_result.error_details = str(e)
|
||||
finally:
|
||||
# 取消心跳任务
|
||||
# 清理工作：取消心跳任务
|
||||
heartbeat_task.cancel()
|
||||
try:
|
||||
# 等待心跳任务完全结束
|
||||
await heartbeat_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
# 计算并记录测试执行时长
|
||||
result.base_result.duration_seconds = time.time() - start_time
|
||||
return result
|
||||
Loading…
Reference in New Issue
Block a user