🎨 优化代码

This commit is contained in:
AkashiCoin 2024-09-01 14:41:01 +08:00
parent df1a84bf77
commit 0348dc53bc
3 changed files with 78 additions and 31 deletions

View File

@ -49,7 +49,6 @@ jobs:
rm -rf poetry.lock rm -rf poetry.lock
poetry source remove ali poetry source remove ali
poetry install --no-root poetry install --no-root
poetry run pip install pydantic==1.10
- name: Run tests - name: Run tests
run: poetry run pytest --cov=zhenxun --cov-report xml run: poetry run pytest --cov=zhenxun --cov-report xml

View File

@ -10,7 +10,6 @@ from nonebot.adapters.onebot.v11.message import Message
from tests.config import BotId, UserId, GroupId, MessageId from tests.config import BotId, UserId, GroupId, MessageId
from tests.utils import ( from tests.utils import (
get_content_bytes,
get_response_json, get_response_json,
_v11_group_message_event, _v11_group_message_event,
_v11_private_message_send, _v11_private_message_send,
@ -19,35 +18,83 @@ from tests.utils import (
def init_mocked_api(mocked_api: MockRouter) -> None: def init_mocked_api(mocked_api: MockRouter) -> None:
mocked_api.get( mocked_api.get(
"https://api.github.com/repos/HibiKier/zhenxun_bot/releases/latest", url="https://api.github.com/repos/HibiKier/zhenxun_bot/releases/latest",
name="release_latest", name="release_latest",
).respond(200, json=get_response_json("release_latest.json")) ).respond(json=get_response_json(path="release_latest.json"))
mocked_api.get( mocked_api.get(
"https://raw.githubusercontent.com/HibiKier/zhenxun_bot/dev/__version__", url="https://raw.githubusercontent.com/HibiKier/zhenxun_bot/dev/__version__",
name="dev_branch_version", name="dev_branch_version",
).respond(200, text="__version__: v0.2.2") ).respond(text="__version__: v0.2.2")
mocked_api.get( mocked_api.get(
"https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/__version__", url="https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/__version__",
name="main_branch_version", name="main_branch_version",
).respond(200, text="__version__: v0.2.2") ).respond(text="__version__: v0.2.2")
mocked_api.get( mocked_api.get(
"https://api.github.com/repos/HibiKier/zhenxun_bot/tarball/v0.2.2", url="https://api.github.com/repos/HibiKier/zhenxun_bot/tarball/v0.2.2",
name="release_download_url", name="release_download_url",
).respond( ).respond(
302, status_code=302,
headers={ headers={
"Location": "https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2" "Location": "https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2"
}, },
) )
import io
import tarfile
tar_buffer = io.BytesIO()
from zhenxun.builtin_plugins.auto_update.config import (
REQ_TXT_FILE,
PYPROJECT_FILE,
PYPROJECT_LOCK_FILE,
)
# 指定要添加到压缩文件中的文件路径列表
file_paths: list[Path] = [
PYPROJECT_FILE,
PYPROJECT_LOCK_FILE,
REQ_TXT_FILE,
]
# 打开一个tarfile对象写入到上面创建的BytesIO对象中
with tarfile.open(mode="w:gz", fileobj=tar_buffer) as tar:
_extracted_from_init_mocked_api_43(tarfile, tar, file_paths, io)
mocked_api.get( mocked_api.get(
"https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2", url="https://codeload.github.com/HibiKier/zhenxun_bot/legacy.tar.gz/refs/tags/v0.2.2",
name="release_download_url_redirect", name="release_download_url_redirect",
).respond( ).respond(
200, content=tar_buffer.getvalue(),
content=get_content_bytes("download_latest_file.tar.gz"),
) )
# TODO Rename this here and in `init_mocked_api`
def _extracted_from_init_mocked_api_43(tarfile, tar, file_paths, io):
folder_name = "my_folder"
tarinfo = tarfile.TarInfo(folder_name)
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( async def test_check_update_release(
app: App, app: App,
mocker: MockerFixture, mocker: MockerFixture,
@ -60,7 +107,7 @@ async def test_check_update_release(
""" """
from zhenxun.builtin_plugins.auto_update import _matcher from zhenxun.builtin_plugins.auto_update import _matcher
init_mocked_api(mocked_api) init_mocked_api(mocked_api=mocked_api)
mocker.patch( mocker.patch(
"zhenxun.builtin_plugins.auto_update._data_source.REPLACE_FOLDERS", "zhenxun.builtin_plugins.auto_update._data_source.REPLACE_FOLDERS",

View File

@ -7,6 +7,7 @@ from respx import MockRouter
from pytest_mock import MockerFixture from pytest_mock import MockerFixture
from nonebot.adapters.onebot.v11 import Bot from nonebot.adapters.onebot.v11 import Bot
from nonebot.adapters.onebot.v11.message import Message from nonebot.adapters.onebot.v11.message import Message
from nonebot.adapters.onebot.v11.event import GroupMessageEvent
from tests.config import BotId, UserId, GroupId, MessageId from tests.config import BotId, UserId, GroupId, MessageId
from tests.utils import get_response_json, _v11_group_message_event from tests.utils import get_response_json, _v11_group_message_event
@ -32,11 +33,11 @@ def init_mocked_api(mocked_api: MockRouter) -> None:
mocked_api.get( mocked_api.get(
"https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/", "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/",
name="github_sub_plugin_contents", name="github_sub_plugin_contents",
).respond(200, json=get_response_json("github_sub_plugin_contents.json")) ).respond(json=get_response_json("github_sub_plugin_contents.json"))
mocked_api.get( mocked_api.get(
"https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/github_sub?ref=main", "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/github_sub?ref=main",
name="github_sub_plugin_api", name="github_sub_plugin_api",
).respond(200, json=get_response_json("github_sub_plugin_api.json")) ).respond(json=get_response_json("github_sub_plugin_api.json"))
mocked_api.get( mocked_api.get(
"https://raw.githubusercontent.com/xuanerwa/zhenxun_github_sub/main/github_sub/__init__.py", "https://raw.githubusercontent.com/xuanerwa/zhenxun_github_sub/main/github_sub/__init__.py",
name="github_sub_plugin_file_init", name="github_sub_plugin_file_init",
@ -55,7 +56,7 @@ async def test_add_plugin_basic(
""" """
from zhenxun.builtin_plugins.plugin_store import _matcher from zhenxun.builtin_plugins.plugin_store import _matcher
init_mocked_api(mocked_api) init_mocked_api(mocked_api=mocked_api)
mocker.patch( mocker.patch(
"zhenxun.builtin_plugins.plugin_store.data_source.BASE_PATH", "zhenxun.builtin_plugins.plugin_store.data_source.BASE_PATH",
return_value=tmp_path / "zhenxun", return_value=tmp_path / "zhenxun",
@ -65,26 +66,26 @@ async def test_add_plugin_basic(
async with app.test_matcher(_matcher) as ctx: async with app.test_matcher(_matcher) as ctx:
bot = create_bot(ctx) bot = create_bot(ctx)
bot = cast(Bot, bot) bot: Bot = cast(Bot, bot)
raw_message = f"添加插件 {plugin_id}" raw_message = f"添加插件 {plugin_id}"
event = _v11_group_message_event( event: GroupMessageEvent = _v11_group_message_event(
raw_message, message=raw_message,
self_id=BotId.QQ_BOT, self_id=BotId.QQ_BOT,
user_id=UserId.SUPERUSER, user_id=UserId.SUPERUSER,
group_id=GroupId.GROUP_ID_LEVEL_5, group_id=GroupId.GROUP_ID_LEVEL_5,
message_id=MessageId.MESSAGE_ID, message_id=MessageId.MESSAGE_ID,
to_me=True, to_me=True,
) )
ctx.receive_event(bot, event) ctx.receive_event(bot=bot, event=event)
ctx.should_call_send( ctx.should_call_send(
event=event, event=event,
message=Message(f"正在添加插件 Id: {plugin_id}"), message=Message(message=f"正在添加插件 Id: {plugin_id}"),
result=None, result=None,
bot=bot, bot=bot,
) )
ctx.should_call_send( ctx.should_call_send(
event=event, event=event,
message=Message("插件 识图 安装成功! 重启后生效"), message=Message(message="插件 识图 安装成功! 重启后生效"),
result=None, result=None,
bot=bot, bot=bot,
) )
@ -106,7 +107,7 @@ async def test_add_plugin_extra(
""" """
from zhenxun.builtin_plugins.plugin_store import _matcher from zhenxun.builtin_plugins.plugin_store import _matcher
init_mocked_api(mocked_api) init_mocked_api(mocked_api=mocked_api)
mocker.patch( mocker.patch(
"zhenxun.builtin_plugins.plugin_store.data_source.BASE_PATH", "zhenxun.builtin_plugins.plugin_store.data_source.BASE_PATH",
return_value=tmp_path / "zhenxun", return_value=tmp_path / "zhenxun",
@ -116,26 +117,26 @@ async def test_add_plugin_extra(
async with app.test_matcher(_matcher) as ctx: async with app.test_matcher(_matcher) as ctx:
bot = create_bot(ctx) bot = create_bot(ctx)
bot = cast(Bot, bot) bot: Bot = cast(Bot, bot)
raw_message = f"添加插件 {plugin_id}" raw_message: str = f"添加插件 {plugin_id}"
event = _v11_group_message_event( event: GroupMessageEvent = _v11_group_message_event(
raw_message, message=raw_message,
self_id=BotId.QQ_BOT, self_id=BotId.QQ_BOT,
user_id=UserId.SUPERUSER, user_id=UserId.SUPERUSER,
group_id=GroupId.GROUP_ID_LEVEL_5, group_id=GroupId.GROUP_ID_LEVEL_5,
message_id=MessageId.MESSAGE_ID, message_id=MessageId.MESSAGE_ID,
to_me=True, to_me=True,
) )
ctx.receive_event(bot, event) ctx.receive_event(bot=bot, event=event)
ctx.should_call_send( ctx.should_call_send(
event=event, event=event,
message=Message(f"正在添加插件 Id: {plugin_id}"), message=Message(message=f"正在添加插件 Id: {plugin_id}"),
result=None, result=None,
bot=bot, bot=bot,
) )
ctx.should_call_send( ctx.should_call_send(
event=event, event=event,
message=Message("插件 github订阅 安装成功! 重启后生效"), message=Message(message="插件 github订阅 安装成功! 重启后生效"),
result=None, result=None,
bot=bot, bot=bot,
) )