refactor: 将容器内测试数据的传输方式由标准输入输出流改为优先通过文件传输

src/providers/constants.py

@@ -1,4 +1,5 @@
 import os
+from pathlib import Path
 from zoneinfo import ZoneInfo
 
 TIME_ZONE = ZoneInfo("Asia/Shanghai")
@@ -38,6 +39,9 @@
 # https://github.com/orgs/nonebot/packages/container/package/nonetest
 DOCKER_IMAGES_VERSION = os.environ.get("DOCKER_IMAGES_VERSION") or "latest"
 DOCKER_IMAGES = f"ghcr.io/nonebot/nonetest:{DOCKER_IMAGES_VERSION}"
+DOCKER_BIND_RESULT_PATH = "/app/plugin_test/test_result.json"
+
+PLUGIN_TEST_DIR = Path("plugin_test")
 
 # Artifact 相关常量
 REGISTRY_DATA_NAME = "registry_data.json"

src/providers/docker_test/__init__.py

@@ -1,10 +1,17 @@
 import json
+import traceback
 from typing import TypedDict
 
 import docker
 from pydantic import BaseModel, SkipValidation, field_validator
 
-from src.providers.constants import DOCKER_IMAGES
+from src.providers.constants import (
+    DOCKER_BIND_RESULT_PATH,
+    DOCKER_IMAGES,
+    PLUGIN_TEST_DIR,
+    PYPI_KEY_TEMPLATE,
+)
+from src.providers.utils import pypi_key_to_path
 
 
 class Metadata(TypedDict):
@@ -50,6 +57,9 @@ def __init__(self, project_link: str, module_name: str, config: str = ""):
         self.module_name = module_name
         self.config = config
 
+        if not PLUGIN_TEST_DIR.exists():
+            PLUGIN_TEST_DIR.mkdir(parents=True, exist_ok=True)
+
     async def run(self, version: str) -> DockerTestResult:
         """运行 Docker 容器测试插件
 
@@ -61,6 +71,13 @@ async def run(self, version: str) -> DockerTestResult:
         """
         # 连接 Docker 环境
         client = docker.DockerClient(base_url="unix://var/run/docker.sock")
+        key = PYPI_KEY_TEMPLATE.format(
+            project_link=self.project_link, module_name=self.module_name
+        )
+        plugin_test_result = PLUGIN_TEST_DIR / f"{pypi_key_to_path(key)}.json"
+
+        # 创建文件，以确保 Docker 容器内可以写入
+        plugin_test_result.touch(exist_ok=True)
 
         try:
             # 运行 Docker 容器，捕获输出。 容器内运行的代码拥有超时设限，此处无需设置超时
@@ -76,20 +93,42 @@ async def run(self, version: str) -> DockerTestResult:
                 },
                 detach=False,
                 remove=True,
-            ).decode()
+                volumes={
+                    plugin_test_result.resolve(strict=False).as_posix(): {
+                        "bind": DOCKER_BIND_RESULT_PATH,
+                        "mode": "rw",
+                    }
+                },
+            )
 
             try:
-                data = json.loads(output)
-            except json.JSONDecodeError:
-                data = {
-                    "run": True,
-                    "load": False,
-                    "output": f"插件测试结果解析失败，输出内容非 JSON 格式。\n输出内容：{output}",
-                }
+                # 若测试结果文件存在且可解析，则优先使用测试结果文件
+                data = json.loads(plugin_test_result.read_text(encoding="utf-8"))
+            except Exception as e:
+                # 如果测试结果文件不存在或不可解析，则尝试使用容器输出内容
+                # 这个时候才需要解码容器输出内容，避免不必要的解码开销
+                try:
+                    data = json.loads(output.decode(encoding="utf-8"))
+                except json.JSONDecodeError:
+                    data = {
+                        "run": True,
+                        "load": False,
+                        "output": f"""
+                        测试结果文件解析失败，输出内容写入失败。
+                        {e}
+                        输出内容：{output}
+                        """,
+                    }
         except Exception as e:
+            # 格式化异常堆栈信息
+            trackback = "".join(traceback.format_exception(type(e), e, e.__traceback__))
+            MAX_OUTPUT = 2000
+            if len(trackback) > MAX_OUTPUT:
+                trackback = trackback[-MAX_OUTPUT:]
+
             data = {
                 "run": False,
                 "load": False,
-                "output": str(e),
+                "output": trackback,
             }
         return DockerTestResult(**data)

src/providers/docker_test/plugin_test.py

@@ -13,7 +13,11 @@
 
 import httpx
 
-from src.providers.constants import REGISTRY_PLUGINS_URL
+from src.providers.constants import (
+    DOCKER_BIND_RESULT_PATH,
+    PLUGIN_TEST_DIR,
+    REGISTRY_PLUGINS_URL,
+)
 
 from .render import render_fake, render_runner
 
@@ -115,7 +119,7 @@ def __init__(
         self.config = config
 
         self._plugin_list = None
-        self._test_dir = Path("plugin_test")
+        self._test_dir = PLUGIN_TEST_DIR
         # 插件信息
         self._version = None
         # 插件测试结果
@@ -190,6 +194,13 @@ async def run(self):
             "config": self.config,
             "test_env": " ".join(self._test_env),
         }
+        # 写入测试结果文件
+        try:
+            result_path = Path(DOCKER_BIND_RESULT_PATH)
+            with open(result_path, "w", encoding="utf-8") as f:
+                json.dump(result, f, ensure_ascii=False)
+        except Exception as e:
+            self._log_output(f"写入测试结果文件失败，错误信息：{e}")
         # 输出测试结果
         print(json.dumps(result, ensure_ascii=False))
         return result

src/providers/utils.py

@@ -125,6 +125,11 @@ def add_step_summary(summary: str):
     logger.debug(f"已添加作业摘要：{summary}")
 
 
+def pypi_key_to_path(key: str) -> str:
+    """将 PyPI 键名转换为路径字符"""
+    return key.replace(":", "-").replace(".", "-").replace("_", "-")
+
+
 @cache
 def get_author_name(author_id: int) -> str:
     """通过作者的ID获取作者名字"""