181 lines
7.5 KiB
Python
181 lines
7.5 KiB
Python
# 实现 Temporal Activity 逻辑
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
from temporalio import activity
|
||
|
||
# 确保能导入 gen 模块，将gen目录添加到Python路径
|
||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'gen')))
|
||
# 导入protobuf生成的模块和其他依赖
|
||
from pb import common_test_pb2 as pb
|
||
from api_tests import execute_api_test_case
|
||
from ui_tests import execute_ui_test_case
|
||
from utils import upload_file_to_s3, scalar_map_to_dict
|
||
|
||
|
||
class TestActivities:
|
||
"""
|
||
测试活动类，包含API测试和UI测试的Temporal Activity实现
|
||
"""
|
||
|
||
@staticmethod
|
||
async def _heartbeat_task(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="RunApiTest")
|
||
async def run_api_test(self, req: pb.ApiTestRequest) -> pb.ApiTestResult:
|
||
"""
|
||
执行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测试逻辑，执行HTTP请求并验证响应
|
||
api_test_success, actual_status, response_headers, 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"
|
||
|
||
# 处理响应头信息
|
||
if response_headers:
|
||
for key, value in response_headers.items():
|
||
# 确保键和值都是字符串类型
|
||
result.headers[str(key)] = str(value)
|
||
|
||
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="RunUiTest")
|
||
async def run_ui_test(self, req: pb.UiTestRequest) -> pb.UiTestResult:
|
||
"""
|
||
执行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_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
|
||
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)
|
||
|
||
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)
|
||
|
||
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
|