refactor: 整理常量同时直接获取所需数据

bot.py

@@ -28,7 +28,8 @@ async def handle_github_action_event():
     try:
         config = driver.config
         # 从环境变量中获取事件信息
-        event_id = config.github_run_id
+        # 读取到的 gitub_run_id 会因为 nonebot 配置加载机制转成 int，需要转回 str
+        event_id = str(config.github_run_id)
         event_name = config.github_event_name
         github_event_path = Path(config.github_event_path)
         # 生成事件

examples/store-test.yml

@@ -46,13 +46,6 @@ jobs:
         run: |
           git checkout `git describe --abbrev=0 --tags`
           poetry install
-          mkdir -p plugin_test/store
-          curl -sSL https://raw.githubusercontent.com/nonebot/registry/results/results.json -o plugin_test/store/previous_results.json
-          curl -sSL https://raw.githubusercontent.com/nonebot/registry/results/plugins.json -o plugin_test/store/previous_plugins.json
-          curl -sSL https://raw.githubusercontent.com/nonebot/nonebot2/master/assets/adapters.json -o plugin_test/store/adapters.json
-          curl -sSL https://raw.githubusercontent.com/nonebot/nonebot2/master/assets/bots.json -o plugin_test/store/bots.json
-          curl -sSL https://raw.githubusercontent.com/nonebot/nonebot2/master/assets/drivers.json -o plugin_test/store/drivers.json
-          curl -sSL https://raw.githubusercontent.com/nonebot/nonebot2/master/assets/plugins.json -o plugin_test/store/plugins.json
 
       - name: Test plugin
         if: ${{ !contains(fromJSON('["Bot", "Adapter", "Plugin"]'), github.event.client_payload.type) }}

src/plugins/publish/__init__.py

@@ -13,7 +13,7 @@
 from src.utils.validation.models import PublishType
 
 from .config import plugin_config
-from .constants import BOT_MARKER, BRANCH_NAME_PREFIX, MAX_NAME_LENGTH
+from .constants import BOT_MARKER, BRANCH_NAME_PREFIX, TITLE_MAX_LENGTH
 from .depends import (
     get_installation_id,
     get_issue_number,
@@ -190,7 +190,7 @@ async def handle_publish_check(
 
         # 设置拉取请求与议题的标题
         # 限制标题长度，过长的标题不好看
-        title = f"{publish_type.value}: {result['name'][:MAX_NAME_LENGTH]}"
+        title = f"{publish_type.value}: {result['name'][:TITLE_MAX_LENGTH]}"
 
         # 分支命名示例 publish/issue123
         branch_name = f"{BRANCH_NAME_PREFIX}{issue_number}"

src/plugins/publish/constants.py

@@ -11,8 +11,8 @@
 
 BRANCH_NAME_PREFIX = "publish/issue"
 
-MAX_NAME_LENGTH = 50
-"""名称最大长度"""
+TITLE_MAX_LENGTH = 50
+"""标题最大长度"""
 
 # 匹配信息的正则表达式
 # 格式：### {标题}\n\n{内容}

src/utils/constants.py

@@ -0,0 +1,14 @@
+# NoneBot 插件商店测试结果
+# https://github.com/nonebot/registry/tree/results
+REGISTRY_BASE_URL = "https://raw.githubusercontent.com/nonebot/registry/results"
+REGISTRY_RESULTS_URL = f"{REGISTRY_BASE_URL}/results.json"
+REGISTRY_PLUGINS_URL = f"{REGISTRY_BASE_URL}/plugins.json"
+
+# NoneBot 插件商店
+# https://github.com/nonebot/nonebot2/tree/master/assets
+STORE_BASE_URL = "https://raw.githubusercontent.com/nonebot/nonebot2/master/assets"
+STORE_ADAPTERS_URL = f"{STORE_BASE_URL}/adapters.json"
+STORE_BOTS_URL = f"{STORE_BASE_URL}/bots.json"
+STORE_DRIVERS_URL = f"{STORE_BASE_URL}/drivers.json"
+STORE_PLUGINS_URL = f"{STORE_BASE_URL}/plugins.json"
+"""plugin_test.py 中也有一个常量，需要同时修改"""

src/utils/store_test/__main__.py

@@ -1,5 +1,5 @@
+import asyncio
 import os
-from asyncio import run
 
 import click
 
@@ -18,7 +18,7 @@ def main(limit: int, offset: int, force: bool, key: str | None):
     config = os.environ.get("PLUGIN_CONFIG")
     data = os.environ.get("PLUGIN_DATA")
 
-    run(test.run(key, config, data))
+    asyncio.run(test.run(key, config, data))
 
 
 if __name__ == "__main__":

src/utils/store_test/constants.py

@@ -17,18 +17,3 @@
 """ 生成的驱动器列表保存路径 """
 PLUGINS_PATH = TEST_DIR / "plugins.json"
 """ 生成的插件列表保存路径 """
-
-STORE_DIR = Path("plugin_test") / "store"
-""" 商店信息文件夹 """
-STORE_ADAPTERS_PATH = STORE_DIR / "adapters.json"
-""" 适配器列表文件路径 """
-STORE_BOTS_PATH = STORE_DIR / "bots.json"
-""" 机器人列表文件路径 """
-STORE_DRIVERS_PATH = STORE_DIR / "drivers.json"
-""" 驱动器列表文件路径 """
-STORE_PLUGINS_PATH = STORE_DIR / "plugins.json"
-""" 插件列表文件路径 """
-PREVIOUS_RESULTS_PATH = STORE_DIR / "previous_results.json"
-""" 上次测试生成的结果文件路径 """
-PREVIOUS_PLUGINS_PATH = STORE_DIR / "previous_plugins.json"
-""" 上次测试生成的插件列表文件路径 """

src/utils/store_test/store.py

@@ -1,18 +1,21 @@
 import click
 
+from src.utils.constants import (
+    REGISTRY_PLUGINS_URL,
+    REGISTRY_RESULTS_URL,
+    STORE_ADAPTERS_URL,
+    STORE_BOTS_URL,
+    STORE_DRIVERS_URL,
+    STORE_PLUGINS_URL,
+)
+
 from .constants import (
     ADAPTERS_PATH,
     BOTS_PATH,
     DRIVERS_PATH,
     PLUGIN_KEY_TEMPLATE,
     PLUGINS_PATH,
-    PREVIOUS_PLUGINS_PATH,
-    PREVIOUS_RESULTS_PATH,
     RESULTS_PATH,
-    STORE_ADAPTERS_PATH,
-    STORE_BOTS_PATH,
-    STORE_DRIVERS_PATH,
-    STORE_PLUGINS_PATH,
 )
 from .models import Plugin, StorePlugin, TestResult
 from .utils import dump_json, get_latest_version, load_json
@@ -33,24 +36,24 @@ def __init__(
         self._force = force
 
         # NoneBot 仓库中的数据
-        self._store_adapters = load_json(STORE_ADAPTERS_PATH)
-        self._store_bots = load_json(STORE_BOTS_PATH)
-        self._store_drivers = load_json(STORE_DRIVERS_PATH)
+        self._store_adapters = load_json(STORE_ADAPTERS_URL)
+        self._store_bots = load_json(STORE_BOTS_URL)
+        self._store_drivers = load_json(STORE_DRIVERS_URL)
         self._store_plugins: dict[str, StorePlugin] = {
             PLUGIN_KEY_TEMPLATE.format(
                 project_link=plugin["project_link"],
                 module_name=plugin["module_name"],
             ): plugin
-            for plugin in load_json(STORE_PLUGINS_PATH)
+            for plugin in load_json(STORE_PLUGINS_URL)
         }
         # 上次测试的结果
-        self._previous_results: dict[str, TestResult] = load_json(PREVIOUS_RESULTS_PATH)
+        self._previous_results: dict[str, TestResult] = load_json(REGISTRY_RESULTS_URL)
         self._previous_plugins: dict[str, Plugin] = {
             PLUGIN_KEY_TEMPLATE.format(
                 project_link=plugin["project_link"],
                 module_name=plugin["module_name"],
             ): plugin
-            for plugin in load_json(PREVIOUS_PLUGINS_PATH)
+            for plugin in load_json(REGISTRY_PLUGINS_URL)
         }
 
     def should_skip(self, key: str) -> bool:
@@ -161,7 +164,6 @@ async def run(
         self, key: str | None = None, config: str | None = None, data: str | None = None
     ):
         """测试商店内插件情况"""
-
         results, plugins = await self.test_plugins(key, config, data)
 
         # 保存测试结果与生成的列表

src/utils/store_test/utils.py

@@ -6,13 +6,12 @@
 import httpx
 
 
-def load_json(path: Path) -> dict:
-    """加载 JSON 文件"""
-    if not path.exists():
-        raise Exception(f"文件 {path} 不存在")
-
-    with open(path, encoding="utf8") as f:
-        return json.load(f)
+def load_json(url: str):
+    """从网络加载 JSON 文件"""
+    r = httpx.get(url)
+    if r.status_code != 200:
+        raise ValueError(f"下载文件失败：{r.text}")
+    return r.json()
 
 
 def dump_json(path: Path, data: dict | list):
@@ -32,9 +31,9 @@ def get_pypi_data(project_link: str) -> dict[str, Any]:
     }
     url = f"https://pypi.org/pypi/{project_link}/json"
     r = httpx.get(url, headers=headers)
-    if r.status_code == 200:
-        return r.json()
-    raise ValueError(f"获取 PyPI 数据失败：{r.text}")
+    if r.status_code != 200:
+        raise ValueError(f"获取 PyPI 数据失败：{r.text}")
+    return r.json()
 
 
 def get_latest_version(project_link: str) -> str:

src/utils/validation/constants.py

@@ -11,17 +11,12 @@
     re.IGNORECASE,
 )
 
-MAX_NAME_LENGTH = 50
+NAME_MAX_LENGTH = 50
 """名称最大长度"""
 
 PLUGIN_VALID_TYPE = ["application", "library"]
 """插件类型当前只支持 application 和 library"""
 
-# NoneBot Store
-STORE_ADAPTERS_URL = (
-    "https://raw.githubusercontent.com/nonebot/nonebot2/master/assets/adapters.json"
-)
-
 # Pydantic 错误信息翻译
 MESSAGE_TRANSLATIONS = {
     "model_type": "值不是合法的字典",

src/utils/validation/models.py

@@ -15,7 +15,7 @@
 from pydantic_extra_types.color import Color
 
 from .constants import (
-    MAX_NAME_LENGTH,
+    NAME_MAX_LENGTH,
     PLUGIN_VALID_TYPE,
     PYPI_PACKAGE_NAME_PATTERN,
     PYTHON_MODULE_NAME_REGEX,
@@ -108,7 +108,7 @@ class Tag(BaseModel):
 class PublishInfo(abc.ABC, BaseModel):
     """发布信息"""
 
-    name: str = Field(max_length=MAX_NAME_LENGTH)
+    name: str = Field(max_length=NAME_MAX_LENGTH)
     desc: str
     author: str
     homepage: str

src/utils/validation/utils.py

@@ -2,10 +2,11 @@
 from typing import TYPE_CHECKING
 
 import httpx
-from nonebot import logger
 from pydantic_extra_types.color import Color, float_to_255
 
-from .constants import MESSAGE_TRANSLATIONS, STORE_ADAPTERS_URL
+from src.utils.constants import STORE_ADAPTERS_URL
+
+from .constants import MESSAGE_TRANSLATIONS
 
 if TYPE_CHECKING:
     from pydantic_core import ErrorDetails
@@ -24,7 +25,6 @@ def check_url(url: str) -> tuple[int, str]:
 
     返回状态码，如果报错则返回 -1
     """
-    logger.info(f"检查网址 {url}")
     try:
         r = httpx.get(url, follow_redirects=True)
         return r.status_code, ""

tests/conftest.py

@@ -11,6 +11,8 @@
 from pytest_mock import MockerFixture
 from respx import MockRouter
 
+from src.utils.constants import STORE_ADAPTERS_URL
+
 if TYPE_CHECKING:
     from nonebot.plugin import Plugin
 
@@ -175,7 +177,7 @@ def mocked_api(respx_mock: MockRouter):
     respx_mock.get("https://www.baidu.com", name="homepage_failed").respond(404)
     respx_mock.get("https://nonebot.dev/", name="homepage").respond()
     respx_mock.get(
-        "https://raw.githubusercontent.com/nonebot/nonebot2/master/assets/adapters.json",
+        STORE_ADAPTERS_URL,
         name="store_adapters",
     ).respond(
         json=[

tests/utils/store_test/test_store_test.py

@@ -1,28 +1,41 @@
-import shutil
+import json
 from pathlib import Path
 
 import pytest
 from pytest_mock import MockerFixture
 from respx import MockRouter
 
+from src.utils.constants import (
+    REGISTRY_PLUGINS_URL,
+    REGISTRY_RESULTS_URL,
+    STORE_ADAPTERS_URL,
+    STORE_BOTS_URL,
+    STORE_DRIVERS_URL,
+    STORE_PLUGINS_URL,
+)
+
+
+def load_json(name: str) -> dict:
+    path = Path(__file__).parent / "store" / f"{name}.json"
+    with path.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
 
 @pytest.fixture
-def mocked_store_data(tmp_path: Path, mocker: MockerFixture) -> dict[str, Path]:
+def mocked_store_data(
+    tmp_path: Path,
+    mocker: MockerFixture,
+    mocked_api: MockRouter,
+) -> dict[str, Path]:
     plugin_test_path = tmp_path / "plugin_test"
-    store_path = plugin_test_path / "store"
+    plugin_test_path.mkdir()
 
     paths = {
         "results": plugin_test_path / "results.json",
         "adapters": plugin_test_path / "adapters.json",
         "bots": plugin_test_path / "bots.json",
         "drivers": plugin_test_path / "drivers.json",
         "plugins": plugin_test_path / "plugins.json",
-        "store_adapters": store_path / "adapters.json",
-        "store_bots": store_path / "bots.json",
-        "store_drivers": store_path / "drivers.json",
-        "store_plugins": store_path / "plugins.json",
-        "previous_results": store_path / "previous_results.json",
-        "previous_plugins": store_path / "previous_plugins.json",
     }
 
     mocker.patch(
@@ -43,29 +56,13 @@ def mocked_store_data(tmp_path: Path, mocker: MockerFixture) -> dict[str, Path]:
         paths["plugins"],
     )
 
-    mocker.patch(
-        "src.utils.store_test.store.STORE_ADAPTERS_PATH",
-        paths["store_adapters"],
-    )
-    mocker.patch("src.utils.store_test.store.STORE_BOTS_PATH", paths["store_bots"])
-    mocker.patch(
-        "src.utils.store_test.store.STORE_DRIVERS_PATH",
-        paths["store_drivers"],
-    )
-    mocker.patch(
-        "src.utils.store_test.store.STORE_PLUGINS_PATH",
-        paths["store_plugins"],
-    )
-    mocker.patch(
-        "src.utils.store_test.store.PREVIOUS_RESULTS_PATH",
-        paths["previous_results"],
-    )
-    mocker.patch(
-        "src.utils.store_test.store.PREVIOUS_PLUGINS_PATH",
-        paths["previous_plugins"],
-    )
+    mocked_api.get(REGISTRY_RESULTS_URL).respond(json=load_json("registry_results"))
+    mocked_api.get(REGISTRY_PLUGINS_URL).respond(json=load_json("registry_plugins"))
+    mocked_api.get(STORE_ADAPTERS_URL).respond(json=load_json("store_adapters"))
+    mocked_api.get(STORE_BOTS_URL).respond(json=load_json("store_bots"))
+    mocked_api.get(STORE_DRIVERS_URL).respond(json=load_json("store_drivers"))
+    mocked_api.get(STORE_PLUGINS_URL).respond(json=load_json("store_plugins"))
 
-    shutil.copytree(Path(__file__).parent / "store", store_path)
     return paths
 
 

tests/utils/store_test/test_utils.py

@@ -0,0 +1,26 @@
+import pytest
+from respx import MockRouter
+
+from src.utils.constants import STORE_ADAPTERS_URL
+
+
+async def test_load_json_failed(mocked_api: MockRouter):
+    """测试加载 json 失败"""
+    from src.utils.store_test.utils import load_json
+
+    mocked_api.get(STORE_ADAPTERS_URL).respond(404)
+
+    with pytest.raises(ValueError) as e:
+        load_json(STORE_ADAPTERS_URL)
+
+    assert str(e.value) == "下载文件失败："
+
+
+async def test_get_pypi_data_failed(mocked_api: MockRouter):
+    """获取 PyPI 数据失败"""
+    from src.utils.store_test.utils import get_pypi_data
+
+    with pytest.raises(ValueError) as e:
+        get_pypi_data("project_link_failed")
+
+    assert str(e.value) == "获取 PyPI 数据失败："