From 4dcbef8307bb8dce1429e62c3befa26acc372948 Mon Sep 17 00:00:00 2001 From: longpeng Date: Wed, 25 Jun 2025 18:43:06 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=BC=BAAPI=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E9=80=BB=E8=BE=91=EF=BC=8C=E6=96=B0=E5=A2=9E=E5=93=8D=E5=BA=94?= =?UTF-8?q?=E5=A4=B4=E5=A4=84=E7=90=86=EF=BC=8C=E4=BC=98=E5=8C=96=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E6=A0=BC=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- proto/common_test.proto | 1 + workers/python/activities.py | 26 +++++++++++++++++--------- workers/python/api_tests.py | 15 ++++++++++----- 3 files changed, 28 insertions(+), 14 deletions(-) diff --git a/proto/common_test.proto b/proto/common_test.proto index 598bac2..bdb9346 100644 --- a/proto/common_test.proto +++ b/proto/common_test.proto @@ -51,6 +51,7 @@ message ApiTestResult { BaseTestResult base_result = 1; int32 actual_status_code = 2; string response_body = 3; + map headers = 4; } // UI测试结果 diff --git a/workers/python/activities.py b/workers/python/activities.py index 759898a..4a2fd82 100644 --- a/workers/python/activities.py +++ b/workers/python/activities.py @@ -1,8 +1,8 @@ # 实现 Temporal Activity 逻辑 +import asyncio import os import sys import time -import asyncio from temporalio import activity @@ -19,6 +19,7 @@ class TestActivities: """ 测试活动类，包含API测试和UI测试的Temporal Activity实现 """ + @staticmethod async def _heartbeat_task(interval_seconds=30): """ @@ -62,13 +63,13 @@ class TestActivities: # 启动后台心跳任务，确保长时间运行的测试不会超时 heartbeat_task = asyncio.create_task(self._heartbeat_task()) - + try: # 发送初始心跳信号 activity.heartbeat() - + # 调用实际的API测试逻辑，执行HTTP请求并验证响应 - api_test_success, actual_status, response_body, log_output = execute_api_test_case( + 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 ) @@ -77,10 +78,17 @@ class TestActivities: 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.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}") @@ -120,11 +128,11 @@ class TestActivities: # 启动后台心跳任务 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) @@ -143,7 +151,7 @@ class TestActivities: 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() @@ -169,4 +177,4 @@ class TestActivities: # 计算并记录测试执行时长 result.base_result.duration_seconds = time.time() - start_time - return result \ No newline at end of file + return result diff --git a/workers/python/api_tests.py b/workers/python/api_tests.py index 8c6cdcd..835da6d 100644 --- a/workers/python/api_tests.py +++ b/workers/python/api_tests.py @@ -1,19 +1,23 @@ # 接口测试具体实现 -import requests import json -def execute_api_test_case(test_case_id: str, endpoint: str, http_method: str, headers: dict, request_body: str, expected_status_code: int): +import requests + + +def execute_api_test_case(test_case_id: str, endpoint: str, http_method: str, headers: dict, request_body: str, + expected_status_code: int): """ 实际执行API测试的函数。 可以集成 pytest, requests 等库。 """ - base_url = "http://101.89.127.197:9080" # 假设 API 服务的基地址 + base_url = "http://101.89.127.197:9080" # 假设 API 服务的基地址 full_url = f"{base_url}{endpoint}" log_output = [] success = False actual_status = 0 response_body = b"" + response_headers = {} log_output.append(f"Executing API test: {test_case_id} - {http_method} {full_url}") @@ -29,9 +33,10 @@ def execute_api_test_case(test_case_id: str, endpoint: str, http_method: str, he actual_status = response.status_code response_body = response.content + response_headers = response.headers log_output.append(f"Response Status: {actual_status}") - log_output.append(f"Response Body: {response_body.decode('utf-8')[:500]}...") # 只显示前500字符 + log_output.append(f"Response Body: {response_body.decode('utf-8')[:500]}...") # 只显示前500字符 if actual_status == expected_status_code: success = True @@ -47,4 +52,4 @@ def execute_api_test_case(test_case_id: str, endpoint: str, http_method: str, he log_output.append(f"API Test Failed (Unexpected Error): {e}") success = False - return success, actual_status, response_body, "\n".join(log_output) \ No newline at end of file + return success, actual_status, response_headers, response_body, "\n".join(log_output)