🐛 测试更新

tests/builtin_plugins/auto_update/test_check_update.py

@@ -314,14 +314,14 @@ async def test_check_update_release(
         ctx.should_call_api(
             "send_msg",
             _v11_private_message_send(
-                message="检测真寻已更新，版本更新：v0.2.2 -> v0.2.2\n" "开始更新...",
+                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" "请重新启动真寻以完成更新!"
+                "版本更新完成\n版本: v0.2.2 -> v0.2.2\n请重新启动真寻以完成更新!".strip()
             ),
             result=None,
             bot=bot,
@@ -347,97 +347,6 @@ async def test_check_update_release(
         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 (
-        PYPROJECT_FILE_STRING,
-        PYPROJECT_LOCK_FILE_STRING,
-        REPLACE_FOLDERS,
-        REQ_TXT_FILE_STRING,
-    )
-
-    init_mocked_api(mocked_api=mocked_api)
-
-    (
-        mock_tmp_path,
-        mock_base_path,
-        mock_backup_path,
-        mock_download_gz_file,
-        mock_download_zip_file,
-        mock_pyproject_file,
-        mock_pyproject_lock_file,
-        mock_req_txt_file,
-        mock_version_file,
-    ) = init_mocker_path(mocker, tmp_path)
-
-    # 确保目录下有一个子目录，以便 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-e6f17c4\n"
-                "开始更新...",
-                user_id=UserId.SUPERUSER,
-            ),
-        )
-        ctx.should_call_send(
-            event=event,
-            message=Message(
-                "版本更新完成\n"
-                "版本: v0.2.2 -> v0.2.2-e6f17c4\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,
@@ -482,7 +391,7 @@ async def test_check_update_main(
     async with app.test_matcher(_matcher) as ctx:
         bot = create_bot(ctx)
         bot = cast(Bot, bot)
-        raw_message = "检查更新 main"
+        raw_message = "检查更新 main -r"
         event = _v11_group_message_event(
             raw_message,
             self_id=BotId.QQ_BOT,
@@ -505,7 +414,8 @@ async def test_check_update_main(
             message=Message(
                 "版本更新完成\n"
                 "版本: v0.2.2 -> v0.2.2-e6f17c4\n"
-                "请重新启动真寻以完成更新!"
+                "请重新启动真寻以完成更新!\n"
+                "资源文件更新成功！"
             ),
             result=None,
             bot=bot,

zhenxun/builtin_plugins/__init__.py

@@ -37,9 +37,15 @@ async def _(bot: Bot):
 @driver.on_bot_disconnect
 async def _(bot: Bot):
     logger.debug(f"Bot: {bot.self_id} 断开连接...")
-    await BotConnectLog.create(
+    try:
-        bot_id=bot.self_id, platform=bot.adapter, connect_time=datetime.now(), type=0
+        await BotConnectLog.create(
-    )
+            bot_id=bot.self_id,
+            platform=bot.adapter,
+            connect_time=datetime.now(),
+            type=0,
+        )
+    except Exception as e:
+        logger.warning(f"记录bot: {bot.self_id} 断开连接失败", e=e)
 
 
 SIGN_SQL = """

zhenxun/builtin_plugins/auto_update/__init__.py

@@ -11,7 +11,7 @@ from nonebot_plugin_alconna import (
     on_alconna,
     store_true,
 )
-from nonebot_plugin_uninfo import Uninfo
+from nonebot_plugin_session import EventSession
 
 from zhenxun.configs.utils import PluginExtraData
 from zhenxun.services.log import logger
@@ -32,12 +32,16 @@ __plugin_meta__ = PluginMetadata(
         检查更新真寻最新版本，包括了自动更新
         资源文件大小一般在130mb左右，除非必须更新一般仅更新代码文件
         指令：
-            检查更新 [main|release] ?[-r]
+            检查更新 [main|release|resource] ?[-r]
-            -r: 下载资源文件
+                main: main分支
+                release: 最新release
+                resource: 资源文件
+                -r: 下载资源文件，一般在更新main或release时使用
             示例:
             检查更新 main
             检查更新 main -r
             检查更新 release -r
+            检查更新 resource
     """.strip(),
     extra=PluginExtraData(
         author="HibiKier",
@@ -49,7 +53,7 @@ __plugin_meta__ = PluginMetadata(
 _matcher = on_alconna(
     Alconna(
         "检查更新",
-        Args["ver_type?", ["main", "release"]],
+        Args["ver_type?", ["main", "release", "resource"]],
         Option("-r|--resource", action=store_true, help_text="下载资源文件"),
     ),
     priority=1,
@@ -62,28 +66,32 @@ _matcher = on_alconna(
 @_matcher.handle()
 async def _(
     bot: Bot,
-    session: Uninfo,
+    session: EventSession,
     ver_type: Match[str],
     resource: Query[bool] = Query("resource", False),
 ):
+    if not session.id1:
+        await MessageUtils.build_message("用户id为空...").finish()
     result = ""
-    if not ver_type.available:
+    if ver_type.result in {"main", "release"}:
-        result = await UpdateManage.check_version()
+        if not ver_type.available:
-        logger.info("查看当前版本...", "检查更新", session=session)
+            result = await UpdateManage.check_version()
-        await MessageUtils.build_message(result).finish()
+            logger.info("查看当前版本...", "检查更新", session=session)
-    try:
+            await MessageUtils.build_message(result).finish()
-        result = await UpdateManage.update(bot, session.user.id, ver_type.result)
+        try:
-    except Exception as e:
+            result = await UpdateManage.update(bot, session.id1, ver_type.result)
-        logger.error("版本更新失败...", "检查更新", session=session, e=e)
+        except Exception as e:
-        await MessageUtils.build_message(f"更新版本失败...e: {e}").finish()
+            logger.error("版本更新失败...", "检查更新", session=session, e=e)
-    if resource.result:
+            await MessageUtils.build_message(f"更新版本失败...e: {e}").finish()
+    if resource.result or ver_type.result == "resource":
         try:
             await ResourceManager.init_resources(True)
+            result += "\n资源文件更新成功！"
         except DownloadResourceException:
             result += "\n资源更新下载失败..."
         except Exception as e:
             logger.error("资源更新下载失败...", "检查更新", session=session, e=e)
             result += "\n资源更新未知错误..."
     if result:
-        await MessageUtils.build_message(result).finish()
+        await MessageUtils.build_message(result.strip()).finish()
     await MessageUtils.build_message("更新版本失败...").finish()

zhenxun/builtin_plugins/auto_update/_data_source.py

@@ -203,7 +203,7 @@ class UpdateManage:
         if await AsyncHttpx.download_file(url, download_file, stream=True):
             logger.debug("下载真寻最新版文件完成...", "检查更新")
             await _file_handle(new_version)
-            result = "版本更新完成\n"
+            result = "版本更新完成"
             return (
                 f"{result}\n"
                 f"版本: {cur_version} -> {new_version}\n"