增强API测试逻辑，新增响应头处理，优化代码格式

proto/common_test.proto

@@ -51,6 +51,7 @@ message ApiTestResult {
   BaseTestResult base_result = 1;
   int32 actual_status_code = 2;
   string response_body = 3;
+  map<string, string> headers = 4;
 }
 
 // UI测试结果

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
+        return result

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)
+    return success, actual_status, response_headers, response_body, "\n".join(log_output)