暂存

go.mod

@@ -5,6 +5,7 @@ go 1.24.4
 require (
 	github.com/google/uuid v1.6.0
 	go.temporal.io/sdk v1.34.0
+	google.golang.org/protobuf v1.36.5
 )
 
 require (
@@ -28,6 +29,5 @@ require (
 	google.golang.org/genproto/googleapis/api v0.0.0-20240827150818-7e3bb234dfed // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240827150818-7e3bb234dfed // indirect
 	google.golang.org/grpc v1.66.0 // indirect
-	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

server/main.go

@@ -16,6 +16,7 @@ func main() {
 	// 创建 Temporal 客户端
 	c, err := client.Dial(client.Options{
 		HostPort:  "temporal.newai.day:17233", // 根据你的 Temporal Server 配置
+		Namespace: "default",
 	})
 	if err != nil {
 		log.Fatalf("Unable to create Temporal client: %v", err)

worker/activities.py

@@ -1,23 +1,22 @@
 # 实现 Temporal Activity 逻辑
 import os
-# 确保能导入 proto_gen 模块
 import sys
 import time
 
 from temporalio import activity
 
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'proto_gen')))
-
+# 确保能导入 gen 模块
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'gen')))
+# 全局变量来存储 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
-from utils import upload_file_to_s3  # 假设有这个函数
-
-from worker.utils import scalar_map_to_dict
+from utils import upload_file_to_s3, scalar_map_to_dict
 
 
-@activity.defn
-async def RunApiTest(req: pb.ApiTestRequest) -> pb.ApiTestResult:
+class TestActivities:
+    @activity.defn
+    async def run_api_test(self,req: pb.ApiTestRequest) -> pb.ApiTestResult:
         """执行API测试的Temporal Activity实现"""
         activity.logger.info(f"Received API Test Request: {req.test_case_id}")
         start_time = time.time()
@@ -33,7 +32,7 @@ async def RunApiTest(req: pb.ApiTestRequest) -> pb.ApiTestResult:
 
             result.base_result.success = api_test_success
             result.actual_status_code = actual_status
-        result.response_body = response_body.decode('utf-8')  # 假设是UTF-8
+            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"
 
@@ -42,15 +41,13 @@ async def RunApiTest(req: pb.ApiTestRequest) -> pb.ApiTestResult:
             result.base_result.success = False
             result.base_result.message = f"API Test Error: {e}"
             result.base_result.error_details = str(e)
-        # 如果是业务逻辑上的不可重试错误，可以抛出 ApplicationError
-        # raise ApplicationError("NonRetryableErrorType", details=str(e))
 
         result.base_result.duration_seconds = time.time() - start_time
         return result
 
 
-@activity.defn
-async def RunUiTest(req: pb.UiTestRequest) -> pb.UiTestResult:
+    @activity.defn
+    async def run_ui_test(self,req: pb.UiTestRequest) -> pb.UiTestResult:
         """执行UI测试的Temporal Activity实现"""
         activity.logger.info(f"Received UI Test Request: {req.test_case_id}")
         start_time = time.time()
@@ -68,10 +65,10 @@ async def RunUiTest(req: pb.UiTestRequest) -> pb.UiTestResult:
             result.base_result.message = "UI Test Passed" if ui_test_success else "UI Test Failed"
 
             # 上传截图和报告到对象存储，并返回URL
-        if screenshot_path:
+            if screenshot_path and os.path.exists(screenshot_path):
                 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:
+            if html_report_path and os.path.exists(html_report_path):
                 result.html_report_url = await upload_file_to_s3(html_report_path, f"reports/{req.test_case_id}.html")
                 os.remove(html_report_path)  # 清理本地文件
 
@@ -80,7 +77,6 @@ async def RunUiTest(req: pb.UiTestRequest) -> pb.UiTestResult:
             result.base_result.success = False
             result.base_result.message = f"UI Test Error: {e}"
             result.base_result.error_details = str(e)
-        # 同样，可以抛出 ApplicationError
 
         result.base_result.duration_seconds = time.time() - start_time
         return result

worker/main.py

@@ -6,20 +6,21 @@ import sys
 from temporalio.client import Client
 from temporalio.worker import Worker
 
-# 确保能导入 proto_gen 模块
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'proto_gen')))
+# 确保能导入 gen 模块
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), 'gen')))
 
-from activities import RunApiTest, RunUiTest # 导入定义的 Activity
+from activities import TestActivities # 导入定义的 Activity
 
 async def main():
     # 连接 Temporal Server
-    client = await Client.connect("localhost:7233") # 根据你的 Temporal Server 配置
+    client = await Client.connect("temporal.newai.day:17233", namespace="default") # 根据你的 Temporal Server 配置
 
+    activities = TestActivities()
     # 创建 Worker
     worker = Worker(
         client,
         task_queue="test-task-queue", # 保持与 Go Client 一致
-        activities=[RunApiTest, RunUiTest],
+        activities=[activities.run_ui_test, activities.run_api_test]
     )
     print("Starting Python Temporal Worker...")
     await worker.run()

worker/ui_tests.py

@@ -1,6 +1,5 @@
 # UI 测试具体实现 (使用 Playwright)
-import asyncio
-from playwright.async_api import Playwright, async_playwright, expect
+from playwright.async_api import async_playwright, expect
 import os
 import datetime
 
@@ -15,13 +14,15 @@ async def execute_ui_test_case(test_case_id: str, url_path: str, browser_type: s
     log_output = []
     success = False
     screenshot_path = None
-    html_report_path = None # Playwright 默认生成HTML报告，通常在测试结束后生成
+    html_report_path = None
+
+    browser = None
+    page = None
 
     log_output.append(f"Executing UI test: {test_case_id} - {full_url} with {browser_type}")
 
     try:
         async with async_playwright() as p:
-            browser = None
             if browser_type == "chromium":
                 browser = await p.chromium.launch(headless=headless)
             elif browser_type == "firefox":
@@ -65,24 +66,19 @@ async def execute_ui_test_case(test_case_id: str, url_path: str, browser_type: s
         success = False
     finally:
         if page:
+            try:
                 timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
                 screenshot_filename = f"screenshot_{test_case_id}_{timestamp}.png"
-            screenshot_path = os.path.join("/tmp", screenshot_filename) # 临时保存路径
+                screenshot_path = os.path.join("/tmp", screenshot_filename)
                 await page.screenshot(path=screenshot_path)
                 log_output.append(f"Screenshot saved to: {screenshot_path}")
+            except Exception as e:
+                log_output.append(f"Failed to take screenshot: {e}")
+
         if browser:
+            try:
                 await browser.close()
+            except Exception as e:
+                log_output.append(f"Failed to close browser: {e}")
 
-    return success, "\n".join(log_output), screenshot_path, html_report_path # html_report_path 留空，因为Playwright通常在测试套件结束后生成
-
-# 辅助函数，模拟对象存储上传
-async def upload_file_to_s3(local_path: str, remote_path: str) -> str:
-    """
-    模拟将文件上传到S3（或任何对象存储）并返回可访问URL。
-    在实际项目中，这里会调用 AWS SDK, MinIO SDK 等。
-    """
-    # 模拟上传延迟
-    await asyncio.sleep(0.1)
-    print(f"Mock Uploaded {local_path} to S3 bucket/path: {remote_path}")
-    # 返回一个模拟的URL
-    return f"https://your-s3-bucket.com/{remote_path}"
+    return success, "\n".join(log_output), screenshot_path, html_report_path