feat: 从插件测试中提取环境信息

src/providers/docker_test/__init__.py

@@ -20,14 +20,23 @@ class Metadata(TypedDict):
 class DockerTestResult(BaseModel):
     """Docker 测试结果"""
 
-    run: bool  # 是否运行
-    load: bool  # 是否加载成功
+    run: bool
+    """ 是否运行测试 """
+    load: bool
+    """ 是否加载成功 """
     version: str | None = None
+    """ 测试版本 """
     config: str = ""
-    # 测试环境 python==3.10 pytest==6.2.5 nonebot2==2.0.0a1 ...
+    """ 测试配置 """
     test_env: str = Field(default="unknown")
+    """测试环境
+
+    python==3.12 nonebot2==2.4.0 pydantic==2.10.0
+    """
     metadata: SkipValidation[Metadata] | None
+    """ 插件元数据 """
     outputs: list[str]
+    """ 测试输出 """
 
     @field_validator("config", mode="before")
     @classmethod
@@ -77,5 +86,4 @@ async def run(self, version: str) -> DockerTestResult:
         ).decode()
 
         data = json.loads(output)
-        data["test_env"] = f"python=={version}"
         return DockerTestResult(**data)

src/providers/docker_test/plugin_test.py

@@ -14,6 +14,7 @@
 import json
 import os
 import re
+import sys
 from asyncio import create_subprocess_shell, subprocess
 from pathlib import Path
 from urllib.request import urlopen
@@ -160,12 +161,23 @@
     with urlopen(PLUGINS_URL) as response:
         plugins = json.loads(response.read())
 
-    return {plugin["project_link"]: plugin["module_name"] for plugin in plugins}
+    return {
+        canonicalize_name(plugin["project_link"]): plugin["module_name"]
+        for plugin in plugins
+    }
 
 
 _canonicalize_regex = re.compile(r"[-_.]+")
 
 
+def canonicalize_name(name: str) -> str:
+    """规范化名称
+
+    packaging.utils 中的 canonicalize_name 实现
+    """
+    return _canonicalize_regex.sub("-", name).lower()
+
+
 def extract_version(output: str, project_link: str) -> str | None:
     """提取插件版本"""
     output = strip_ansi(output)
@@ -177,7 +189,7 @@
 
     # poetry 使用 packaging.utils 中的 canonicalize_name 规范化名称
     # 在这里我们也需要规范化名称，以正确匹配版本号
-    project_link = _canonicalize_regex.sub("-", project_link).lower()
+    project_link = canonicalize_name(project_link)
 
     # 匹配版本解析失败的情况
     match = re.search(
@@ -192,6 +204,20 @@
         return match.group(1).strip()
 
 
+def parse_requirements(requirements: str) -> dict[str, str]:
+    """解析 requirements.txt 文件"""
+    # anyio==3.6.2 ; python_version >= "3.11" and python_version < "4.0"
+    # pydantic[dotenv]==1.10.6 ; python_version >= "3.10" and python_version < "4.0"
+    results = {}
+    for line in requirements.strip().splitlines():
+        match = re.match(r"^(.+?)(?:\[.+\])?==(.+) ;", line)
+        if match:
+            package_name = match.group(1)
+            version = match.group(2)
+            results[package_name] = version
+    return results
+
+
 class PluginTest:
     def __init__(self, project_info: str, config: str | None = None) -> None:
         """插件测试构造函数
@@ -213,7 +239,8 @@
         self._lines_output = []
 
         # 插件测试目录
-        self.test_dir = Path("plugin_test")
+        self._test_dir = Path("plugin_test")
+        self._test_env = []
 
     @property
     def key(self) -> str:
@@ -229,7 +256,7 @@
         """插件测试目录"""
         # 替换 : 为 -，防止文件名不合法
         key = self.key.replace(":", "-")
-        return self.test_dir / f"{key}"
+        return self._test_dir / f"{key}"
 
     @property
     def env(self) -> dict[str, str]:
@@ -254,8 +281,8 @@
         """插件测试入口"""
 
         # 创建测试目录
-        if not self.test_dir.exists():
-            self.test_dir.mkdir()
+        if not self._test_dir.exists():
+            self._test_dir.mkdir()
 
         # 创建插件测试项目
         await self.create_poetry_project()
@@ -272,21 +299,18 @@
             with open(self.path / "metadata.json", encoding="utf-8") as f:
                 metadata = json.load(f)
 
+        result = {
+            "metadata": metadata,
+            "outputs": self._lines_output,
+            "load": self._run,
+            "run": self._create,
+            "version": self._version,
+            "config": self.config,
+            "test_env": " ".join(self._test_env),
+        }
         # 输出测试结果
-        print(
-            json.dumps(
-                {
-                    "metadata": metadata,
-                    "outputs": self._lines_output,
-                    "load": self._run,
-                    "run": self._create,
-                    "version": self._version,
-                    "config": self.config,
-                }
-            )
-        )
-
-        return self._run, self._lines_output
+        print(json.dumps(result, ensure_ascii=False))
+        return result
 
     async def command(self, cmd: str, timeout: int = 300) -> tuple[bool, str, str]:
         """执行命令
@@ -328,7 +352,7 @@
 
             if self._create:
                 self._log_output(f"项目 {self.project_link} 创建成功。")
-                self._std_output(stdout, "")
+                self._std_output(stdout)
             else:
                 # 创建失败时尝试从报错中获取插件版本号
                 self._version = extract_version(stdout + stderr, self.project_link)
@@ -352,7 +376,7 @@
 
                 # 记录插件信息至输出
                 self._log_output(f"插件 {self.project_link} 的信息如下：")
-                self._std_output(stdout, "")
+                self._std_output(stdout)
             else:
                 self._log_output(f"插件 {self.project_link} 信息获取失败。")
                 self._std_output(stdout, stderr)
@@ -387,7 +411,7 @@
 
             if self._run:
                 self._log_output(f"插件 {self.module_name} 加载正常：")
-                self._std_output(stdout, "")
+                self._std_output(stdout)
             else:
                 self._log_output(f"插件 {self.module_name} 加载出错：")
                 self._std_output(stdout, stderr)
@@ -399,55 +423,64 @@
 
             if code:
                 self._log_output(f"插件 {self.project_link} 依赖的插件如下：")
-                for i in stdout.strip().splitlines():
-                    module_name = self._get_plugin_module_name(i)
-                    if module_name:
-                        self._deps.append(module_name)
+                requirements = parse_requirements(stdout)
+                self._deps = self._get_deps(requirements)
+                self._test_env = self._get_test_env(requirements)
                 self._log_output(f"    {', '.join(self._deps)}")
             else:
                 self._log_output(f"插件 {self.project_link} 依赖获取失败。")
                 self._std_output(stdout, stderr)
 
     @property
     def plugin_list(self) -> dict[str, str]:
-        """
-        获取插件列表
-        """
+        """获取插件列表"""
         if self._plugin_list is None:
             self._plugin_list = get_plugin_list()
         return self._plugin_list
 
-    def _std_output(self, stdout: str, stderr: str):
-        """
-        将标准输出流与标准错误流记录并输出
-        """
+    def _std_output(self, stdout: str, stderr: str = ""):
+        """将标准输出流与标准错误流记录并输出"""
         _out = stdout.strip().splitlines()
         _err = stderr.strip().splitlines()
+
         for i in _out:
             self._log_output(f"    {i}")
-
         for i in _err:
             self._log_output(f"    {i}")
 
-    def _get_plugin_module_name(self, require: str) -> str | None:
-        """
-        解析插件的依赖名称
-        """
-        # anyio==3.6.2 ; python_version >= "3.11" and python_version < "4.0"
-        # pydantic[dotenv]==1.10.6 ; python_version >= "3.10" and python_version < "4.0"
-        match = re.match(r"^(.+?)(?:\[.+\])?==", require.strip())
-        if match:
-            package_name = match.group(1)
-            # 不用包括自己
-            if package_name in self.plugin_list and package_name != self.project_link:
-                return self.plugin_list[package_name]
+    def _get_deps(self, requirements: dict[str, str]) -> list[str]:
+        """获取插件依赖"""
+        deps = []
+        for package_name in requirements:
+            if (
+                package_name in self.plugin_list
+                # 不用包括插件自己
+                and package_name != canonicalize_name(self.project_link)
+            ):
+                module_name = self.plugin_list[package_name]
+                deps.append(module_name)
+        return deps
+
+    def _get_test_env(self, requirements: dict[str, str]) -> list[str]:
+        """获取测试环境"""
+        # python 版本
+        envs = [
+            f"python=={sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
+        ]
+        # 特定插件依赖
+        # 当前仅需记录 nonebot2 和 pydantic 的版本
+        if "nonebot2" in requirements:
+            envs.append(f"nonebot2=={requirements['nonebot2']}")
+        if "pydantic" in requirements:
+            envs.append(f"pydantic=={requirements['pydantic']}")
+        return envs
 
 
 def main():
-    """
-    根据传入的环境变量 PLUGIN_INFO 和 PLUGIN_CONFIG 进行测试
+    """根据传入的环境变量进行测试
 
     PLUGIN_INFO 即为该插件的 KEY
+    PLUGIN_CONFIG 即为该插件的配置
     """
 
     plugin_info = os.environ.get("PLUGIN_INFO", "")

tests/utils/docker_test/test_docker_plugin_test.py

@@ -17,6 +17,7 @@ async def test_docker_plugin_test(mocked_api: MockRouter, mocker: MockerFixture)
             "run": True,
             "version": "0.0.1",
             "config": "",
+            "test_env": "python==3.12",
         }
     ).encode()
     mocked_client = mocker.Mock()
@@ -75,6 +76,7 @@ async def test_docker_plugin_test_metadata_some_fields_empty(
             "run": True,
             "version": "0.0.1",
             "config": "",
+            "test_env": "python==3.12",
         }
     ).encode()
     mocked_client = mocker.Mock()
@@ -139,6 +141,7 @@ async def test_docker_plugin_test_metadata_some_fields_invalid(
             "run": True,
             "version": "0.0.1",
             "config": "",
+            "test_env": "python==3.12",
         }
     ).encode()
     mocked_client = mocker.Mock()

tests/utils/docker_test/test_parse_requirements.py

@@ -0,0 +1,21 @@
+def test_parse_requirements():
+    """解析 poetry export --without-hashes 的输出"""
+    from src.providers.docker_test.plugin_test import parse_requirements
+
+    output = """
+anyio==4.6.2.post1 ; python_version >= "3.9" and python_version < "4.0"
+nonebot2[httpx]==2.4.0 ; python_version >= "3.9" and python_version < "4.0"
+nonebug==0.4.2 ; python_version >= "3.9" and python_version < "4.0"
+pydantic-core==2.27.0 ; python_version >= "3.9" and python_version < "4.0"
+pydantic==2.10.0 ; python_version >= "3.9" and python_version < "4.0"
+"""
+
+    requirements = parse_requirements(output)
+
+    assert requirements == {
+        "anyio": "4.6.2.post1",
+        "nonebot2": "2.4.0",
+        "nonebug": "0.4.2",
+        "pydantic-core": "2.27.0",
+        "pydantic": "2.10.0",
+    }