🐛修复添加插件返回403的问题 (#1595)

* 完善测试方法 * vscode测试配置 * 重构插件安装过程
2026-10-03 02:30:01 +08:00 · 2024-09-02 01:50:39 +08:00
parent 8fe061738a
commit 29c48765ae
18 changed files with 1090 additions and 452 deletions
@@ -1,3 +1,7 @@
+import io
+import os
+import tarfile
+import zipfile
 from typing import cast
 from pathlib import Path
 from collections.abc import Callable
@@ -9,18 +13,19 @@ from nonebot.adapters.onebot.v11 import Bot
 from nonebot.adapters.onebot.v11.message import Message

 from tests.config import BotId, UserId, GroupId, MessageId
-from tests.utils import (
-    get_response_json,
-    _v11_group_message_event,
-    _v11_private_message_send,
-)
+from tests.utils import get_response_json as _get_response_json
+from tests.utils import _v11_group_message_event, _v11_private_message_send
+
+
+def get_response_json(file: str) -> dict:
+    return _get_response_json(Path() / "auto_update", file)


 def init_mocked_api(mocked_api: MockRouter) -> None:
    mocked_api.get(
        url="https://api.github.com/repos/HibiKier/zhenxun_bot/releases/latest",
        name="release_latest",
-    ).respond(json=get_response_json(path="release_latest.json"))
+    ).respond(json=get_response_json("release_latest.json"))
    mocked_api.get(
        url="https://raw.githubusercontent.com/HibiKier/zhenxun_bot/dev/__version__",
        name="dev_branch_version",
@@ -38,62 +43,105 @@ def init_mocked_api(mocked_api: MockRouter) -> None:
            "Location": "https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2"
        },
    )
-    import io
-    import tarfile

    tar_buffer = io.BytesIO()
+    zip_bytes = io.BytesIO()

    from zhenxun.builtin_plugins.auto_update.config import (
-        REQ_TXT_FILE,
-        PYPROJECT_FILE,
-        PYPROJECT_LOCK_FILE,
+        REPLACE_FOLDERS,
+        REQ_TXT_FILE_STRING,
+        PYPROJECT_FILE_STRING,
+        PYPROJECT_LOCK_FILE_STRING,
    )

    # 指定要添加到压缩文件中的文件路径列表
-    file_paths: list[Path] = [
-        PYPROJECT_FILE,
-        PYPROJECT_LOCK_FILE,
-        REQ_TXT_FILE,
+    file_paths: list[str] = [
+        PYPROJECT_FILE_STRING,
+        PYPROJECT_LOCK_FILE_STRING,
+        REQ_TXT_FILE_STRING,
    ]

    # 打开一个tarfile对象，写入到上面创建的BytesIO对象中
    with tarfile.open(mode="w:gz", fileobj=tar_buffer) as tar:
-        _extracted_from_init_mocked_api_43(tarfile, tar, file_paths, io)
+        _extracted_from_init_mocked_api_43(tar, file_paths, folders=REPLACE_FOLDERS)
+
+    with zipfile.ZipFile(zip_bytes, mode="w", compression=zipfile.ZIP_DEFLATED) as zipf:
+        _extracted_from_init_mocked_api_zip(zipf, file_paths, folders=REPLACE_FOLDERS)
+
    mocked_api.get(
        url="https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2",
        name="release_download_url_redirect",
    ).respond(
        content=tar_buffer.getvalue(),
    )
+    mocked_api.get(
+        url="https://ghproxy.cc/https://github.com/HibiKier/zhenxun_bot/archive/refs/heads/dev.zip",
+        name="dev_download_url",
+    ).respond(
+        content=zip_bytes.getvalue(),
+    )
+    mocked_api.get(
+        url="https://ghproxy.cc/https://github.com/HibiKier/zhenxun_bot/archive/refs/heads/main.zip",
+        name="main_download_url",
+    ).respond(
+        content=zip_bytes.getvalue(),
+    )


 # TODO Rename this here and in `init_mocked_api`
-def _extracted_from_init_mocked_api_43(tarfile, tar, file_paths, io):
+def _extracted_from_init_mocked_api_zip(
+    zipf: zipfile.ZipFile, file_paths: list[str], folders: list[str] = []
+):
+    # 假设有一个文件夹名为 folder_name
+    folder_name = "my_folder/"
+
+    # 添加文件夹到 ZIP 中，注意 ZIP 中文件夹路径应以 '/' 结尾
+    zipf.writestr(folder_name, "")  # 空内容表示这是一个文件夹
+
+    for file_path in file_paths:
+        # 将文件添加到 ZIP 中，路径为 folder_name + file_name
+        zipf.writestr(f"{folder_name}{os.path.basename(file_path)}", b"new")
+    base_folder = f"{folder_name}zhenxun/"
+    zipf.writestr(base_folder, "")
+
+    for folder in folders:
+        zipf.writestr(f"{base_folder}{folder}/", "")
+
+
+# TODO Rename this here and in `init_mocked_api`
+def _extracted_from_init_mocked_api_43(
+    tar: tarfile.TarFile, file_paths: list[str], folders: list[str] = []
+):
    folder_name = "my_folder"
    tarinfo = tarfile.TarInfo(folder_name)
+    _extracted_from__extracted_from_init_mocked_api_43_30(tarinfo, tar)
+    # 读取并添加指定的文件
+    for file_path in file_paths:
+        # 创建TarInfo对象
+        tar_buffer = io.BytesIO(b"new")
+        tarinfo = tarfile.TarInfo(
+            f"{folder_name}/{file_path}"
+        )  # 使用文件名作为tar中的名字
+        tarinfo.mode = 0o644  # 设置文件夹权限
+        tarinfo.size = len(tar_buffer.getvalue())  # 设置文件大小
+
+        # 添加文件
+        tar.addfile(tarinfo, fileobj=tar_buffer)
+
+    base_folder = f"{folder_name}/zhenxun"
+    tarinfo = tarfile.TarInfo(base_folder)
+    _extracted_from__extracted_from_init_mocked_api_43_30(tarinfo, tar)
+    for folder in folders:
+        tarinfo = tarfile.TarInfo(f"{base_folder}{folder}")
+        _extracted_from__extracted_from_init_mocked_api_43_30(tarinfo, tar)
+
+
+# TODO Rename this here and in `_extracted_from_init_mocked_api_43`
+def _extracted_from__extracted_from_init_mocked_api_43_30(tarinfo, tar):
    tarinfo.type = tarfile.DIRTYPE
    tarinfo.mode = 0o755
    tar.addfile(tarinfo)

-    # 读取并添加指定的文件
-    for file_path in file_paths:
-        # 读取文件内容
-        with open(file_path, "rb") as file:
-            file_content = file.read()
-
-        # 使用BytesIO创建文件内容
-        file_buffer = io.BytesIO(file_content)
-
-        # 创建TarInfo对象
-        tarinfo = tarfile.TarInfo(
-            f"{folder_name}/{file_path.name}"
-        )  # 使用文件名作为tar中的名字
-        tarinfo.mode = 0o644  # 设置文件夹权限
-        tarinfo.size = len(file_content)
-
-        # 添加文件
-        tar.addfile(tarinfo, fileobj=file_buffer)
-

 async def test_check_update_release(
    app: App,
@@ -103,20 +151,64 @@ async def test_check_update_release(
    tmp_path: Path,
 ) -> None:
    """
-    测试检查更新
+    测试检查更新（release）
    """
    from zhenxun.builtin_plugins.auto_update import _matcher
+    from zhenxun.builtin_plugins.auto_update.config import (
+        REPLACE_FOLDERS,
+        REQ_TXT_FILE_STRING,
+        PYPROJECT_FILE_STRING,
+        PYPROJECT_LOCK_FILE_STRING,
+    )

    init_mocked_api(mocked_api=mocked_api)

-    mocker.patch(
-        "zhenxun.builtin_plugins.auto_update._data_source.REPLACE_FOLDERS",
-        return_value=[],
-    )
    mocker.patch(
        "zhenxun.builtin_plugins.auto_update._data_source.install_requirement",
        return_value=None,
    )
+    mock_tmp_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.TMP_PATH",
+        new=tmp_path / "auto_update",
+    )
+    mock_base_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.BASE_PATH",
+        new=tmp_path / "zhenxun",
+    )
+    mock_backup_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.BACKUP_PATH",
+        new=tmp_path / "backup",
+    )
+    mock_download_gz_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_GZ_FILE",
+        new=mock_tmp_path / "download_latest_file.tar.gz",
+    )
+    mock_download_zip_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_ZIP_FILE",
+        new=mock_tmp_path / "download_latest_file.zip",
+    )
+    mock_pyproject_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_FILE",
+        new=tmp_path / PYPROJECT_FILE_STRING,
+    )
+    mock_pyproject_lock_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_LOCK_FILE",
+        new=tmp_path / PYPROJECT_LOCK_FILE_STRING,
+    )
+    mock_req_txt_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.REQ_TXT_FILE",
+        new=tmp_path / REQ_TXT_FILE_STRING,
+    )
+
+    # 确保目录下有一个子目录，以便 os.listdir() 能返回一个目录名
+    mock_tmp_path.mkdir(parents=True, exist_ok=True)
+
+    for folder in REPLACE_FOLDERS:
+        (mock_base_path / folder).mkdir(parents=True, exist_ok=True)
+
+    mock_pyproject_file.write_bytes(b"")
+    mock_pyproject_lock_file.write_bytes(b"")
+    mock_req_txt_file.write_bytes(b"")

    async with app.test_matcher(_matcher) as ctx:
        bot = create_bot(ctx)
@@ -150,3 +242,245 @@ async def test_check_update_release(
    assert mocked_api["release_latest"].called
    assert mocked_api["release_download_url"].called
    assert mocked_api["release_download_url_redirect"].called
+
+    assert (mock_backup_path / PYPROJECT_FILE_STRING).exists()
+    assert (mock_backup_path / PYPROJECT_LOCK_FILE_STRING).exists()
+    assert (mock_backup_path / REQ_TXT_FILE_STRING).exists()
+
+    assert not mock_download_gz_file.exists()
+    assert not mock_download_zip_file.exists()
+
+    assert mock_pyproject_file.read_bytes() == b"new"
+    assert mock_pyproject_lock_file.read_bytes() == b"new"
+    assert mock_req_txt_file.read_bytes() == b"new"
+
+    for folder in REPLACE_FOLDERS:
+        assert not (mock_base_path / folder).exists()
+    for folder in REPLACE_FOLDERS:
+        assert (mock_backup_path / folder).exists()
+
+
+async def test_check_update_dev(
+    app: App,
+    mocker: MockerFixture,
+    mocked_api: MockRouter,
+    create_bot: Callable,
+    tmp_path: Path,
+) -> None:
+    """
+    测试检查更新（开发环境）
+    """
+    from zhenxun.builtin_plugins.auto_update import _matcher
+    from zhenxun.builtin_plugins.auto_update.config import (
+        REPLACE_FOLDERS,
+        REQ_TXT_FILE_STRING,
+        PYPROJECT_FILE_STRING,
+        PYPROJECT_LOCK_FILE_STRING,
+    )
+
+    init_mocked_api(mocked_api=mocked_api)
+
+    mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.install_requirement",
+        return_value=None,
+    )
+    mock_tmp_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.TMP_PATH",
+        new=tmp_path / "auto_update",
+    )
+    mock_base_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.BASE_PATH",
+        new=tmp_path / "zhenxun",
+    )
+    mock_backup_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.BACKUP_PATH",
+        new=tmp_path / "backup",
+    )
+    mock_download_gz_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_GZ_FILE",
+        new=mock_tmp_path / "download_latest_file.tar.gz",
+    )
+    mock_download_zip_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_ZIP_FILE",
+        new=mock_tmp_path / "download_latest_file.zip",
+    )
+    mock_pyproject_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_FILE",
+        new=tmp_path / PYPROJECT_FILE_STRING,
+    )
+    mock_pyproject_lock_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_LOCK_FILE",
+        new=tmp_path / PYPROJECT_LOCK_FILE_STRING,
+    )
+    mock_req_txt_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.REQ_TXT_FILE",
+        new=tmp_path / REQ_TXT_FILE_STRING,
+    )
+
+    # 确保目录下有一个子目录，以便 os.listdir() 能返回一个目录名
+    mock_tmp_path.mkdir(parents=True, exist_ok=True)
+    for folder in REPLACE_FOLDERS:
+        (mock_base_path / folder).mkdir(parents=True, exist_ok=True)
+
+    mock_pyproject_file.write_bytes(b"")
+    mock_pyproject_lock_file.write_bytes(b"")
+    mock_req_txt_file.write_bytes(b"")
+
+    async with app.test_matcher(_matcher) as ctx:
+        bot = create_bot(ctx)
+        bot = cast(Bot, bot)
+        raw_message = "检查更新 dev"
+        event = _v11_group_message_event(
+            raw_message,
+            self_id=BotId.QQ_BOT,
+            user_id=UserId.SUPERUSER,
+            group_id=GroupId.GROUP_ID_LEVEL_5,
+            message_id=MessageId.MESSAGE_ID,
+            to_me=True,
+        )
+        ctx.receive_event(bot, event)
+        ctx.should_call_api(
+            "send_msg",
+            _v11_private_message_send(
+                message="检测真寻已更新，版本更新：v0.2.2 -> v0.2.2\n开始更新...",
+                user_id=UserId.SUPERUSER,
+            ),
+        )
+        ctx.should_call_send(
+            event=event,
+            message=Message(
+                "版本更新完成\n" "版本: v0.2.2 -> v0.2.2\n" "请重新启动真寻以完成更新!"
+            ),
+            result=None,
+            bot=bot,
+        )
+        ctx.should_finished(_matcher)
+    assert mocked_api["dev_download_url"].called
+    assert (mock_backup_path / PYPROJECT_FILE_STRING).exists()
+    assert (mock_backup_path / PYPROJECT_LOCK_FILE_STRING).exists()
+    assert (mock_backup_path / REQ_TXT_FILE_STRING).exists()
+
+    assert not mock_download_gz_file.exists()
+    assert not mock_download_zip_file.exists()
+
+    assert mock_pyproject_file.read_bytes() == b"new"
+    assert mock_pyproject_lock_file.read_bytes() == b"new"
+    assert mock_req_txt_file.read_bytes() == b"new"
+
+    for folder in REPLACE_FOLDERS:
+        assert (mock_base_path / folder).exists()
+    for folder in REPLACE_FOLDERS:
+        assert (mock_backup_path / folder).exists()
+
+
+async def test_check_update_main(
+    app: App,
+    mocker: MockerFixture,
+    mocked_api: MockRouter,
+    create_bot: Callable,
+    tmp_path: Path,
+) -> None:
+    """
+    测试检查更新（正式环境）
+    """
+    from zhenxun.builtin_plugins.auto_update import _matcher
+    from zhenxun.builtin_plugins.auto_update.config import (
+        REPLACE_FOLDERS,
+        REQ_TXT_FILE_STRING,
+        PYPROJECT_FILE_STRING,
+        PYPROJECT_LOCK_FILE_STRING,
+    )
+
+    init_mocked_api(mocked_api=mocked_api)
+
+    mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.install_requirement",
+        return_value=None,
+    )
+    mock_tmp_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.TMP_PATH",
+        new=tmp_path / "auto_update",
+    )
+    mock_base_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.BASE_PATH",
+        new=tmp_path / "zhenxun",
+    )
+    mock_backup_path = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.BACKUP_PATH",
+        new=tmp_path / "backup",
+    )
+    mock_download_gz_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_GZ_FILE",
+        new=mock_tmp_path / "download_latest_file.tar.gz",
+    )
+    mock_download_zip_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.DOWNLOAD_ZIP_FILE",
+        new=mock_tmp_path / "download_latest_file.zip",
+    )
+    mock_pyproject_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_FILE",
+        new=tmp_path / PYPROJECT_FILE_STRING,
+    )
+    mock_pyproject_lock_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.PYPROJECT_LOCK_FILE",
+        new=tmp_path / PYPROJECT_LOCK_FILE_STRING,
+    )
+    mock_req_txt_file = mocker.patch(
+        "zhenxun.builtin_plugins.auto_update._data_source.REQ_TXT_FILE",
+        new=tmp_path / REQ_TXT_FILE_STRING,
+    )
+
+    # 确保目录下有一个子目录，以便 os.listdir() 能返回一个目录名
+    mock_tmp_path.mkdir(parents=True, exist_ok=True)
+    for folder in REPLACE_FOLDERS:
+        (mock_base_path / folder).mkdir(parents=True, exist_ok=True)
+
+    mock_pyproject_file.write_bytes(b"")
+    mock_pyproject_lock_file.write_bytes(b"")
+    mock_req_txt_file.write_bytes(b"")
+
+    async with app.test_matcher(_matcher) as ctx:
+        bot = create_bot(ctx)
+        bot = cast(Bot, bot)
+        raw_message = "检查更新 main"
+        event = _v11_group_message_event(
+            raw_message,
+            self_id=BotId.QQ_BOT,
+            user_id=UserId.SUPERUSER,
+            group_id=GroupId.GROUP_ID_LEVEL_5,
+            message_id=MessageId.MESSAGE_ID,
+            to_me=True,
+        )
+        ctx.receive_event(bot, event)
+        ctx.should_call_api(
+            "send_msg",
+            _v11_private_message_send(
+                message="检测真寻已更新，版本更新：v0.2.2 -> v0.2.2\n开始更新...",
+                user_id=UserId.SUPERUSER,
+            ),
+        )
+        ctx.should_call_send(
+            event=event,
+            message=Message(
+                "版本更新完成\n" "版本: v0.2.2 -> v0.2.2\n" "请重新启动真寻以完成更新!"
+            ),
+            result=None,
+            bot=bot,
+        )
+        ctx.should_finished(_matcher)
+    assert mocked_api["main_download_url"].called
+    assert (mock_backup_path / PYPROJECT_FILE_STRING).exists()
+    assert (mock_backup_path / PYPROJECT_LOCK_FILE_STRING).exists()
+    assert (mock_backup_path / REQ_TXT_FILE_STRING).exists()
+
+    assert not mock_download_gz_file.exists()
+    assert not mock_download_zip_file.exists()
+
+    assert mock_pyproject_file.read_bytes() == b"new"
+    assert mock_pyproject_lock_file.read_bytes() == b"new"
+    assert mock_req_txt_file.read_bytes() == b"new"
+
+    for folder in REPLACE_FOLDERS:
+        assert (mock_base_path / folder).exists()
+    for folder in REPLACE_FOLDERS:
+        assert (mock_backup_path / folder).exists()