添加测试:更新与添加插件 (#1594)

*  测试更新与添加插件

*  Sourcery建议

* 👷 添加pytest

* 🎨 优化代码
This commit is contained in:
AkashiCoin 2024-09-01 14:53:41 +08:00 committed by GitHub
parent a3681216bd
commit 7e7436f433
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 624 additions and 17 deletions

View File

@ -28,7 +28,14 @@ jobs:
uses: actions/cache@v3 uses: actions/cache@v3
with: with:
path: ~/.cache/pypoetry path: ~/.cache/pypoetry
key: poetry-cache-${{ runner.os }}-${{ steps.setup_python.outputs.python-version }} key: poetry-cache-${{ runner.os }}-${{ steps.setup_python.outputs.python-version }}-${{ hashFiles('pyproject.toml') }}
- name: Cache playwright cache
id: cache-playwright
uses: actions/cache@v3
with:
path: ~/.cache/ms-playwright
key: playwright-cache-${{ runner.os }}-${{ steps.setup_python.outputs.python-version }}
- name: Cache Data cache - name: Cache Data cache
uses: actions/cache@v3 uses: actions/cache@v3
@ -42,7 +49,9 @@ 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
run: poetry run pytest --cov=zhenxun --cov-report xml
- name: Check bot run - name: Check bot run
id: bot_check_run id: bot_check_run

View File

@ -134,6 +134,9 @@ typeCheckingMode = "standard"
reportShadowedImports = false reportShadowedImports = false
disableBytesTypePromotions = true disableBytesTypePromotions = true
[tool.pytest.ini_options]
asyncio_mode = "auto"
[build-system] [build-system]
requires = ["poetry-core>=1.0.0"] requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api" build-backend = "poetry.core.masonry.api"

View File

@ -0,0 +1,152 @@
from typing import cast
from pathlib import Path
from collections.abc import Callable
from nonebug import App
from respx import MockRouter
from pytest_mock import MockerFixture
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,
)
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"))
mocked_api.get(
url="https://raw.githubusercontent.com/HibiKier/zhenxun_bot/dev/__version__",
name="dev_branch_version",
).respond(text="__version__: v0.2.2")
mocked_api.get(
url="https://raw.githubusercontent.com/HibiKier/zhenxun_bot/main/__version__",
name="main_branch_version",
).respond(text="__version__: v0.2.2")
mocked_api.get(
url="https://api.github.com/repos/HibiKier/zhenxun_bot/tarball/v0.2.2",
name="release_download_url",
).respond(
status_code=302,
headers={
"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(
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(),
)
# 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(
app: App,
mocker: MockerFixture,
mocked_api: MockRouter,
create_bot: Callable,
tmp_path: Path,
) -> None:
"""
测试检查更新
"""
from zhenxun.builtin_plugins.auto_update import _matcher
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,
)
async with app.test_matcher(_matcher) as ctx:
bot = create_bot(ctx)
bot = cast(Bot, bot)
raw_message = "检查更新 release"
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["release_latest"].called
assert mocked_api["release_download_url"].called
assert mocked_api["release_download_url_redirect"].called

View File

@ -0,0 +1,147 @@
from typing import cast
from pathlib import Path
from collections.abc import Callable
from nonebug import App
from respx import MockRouter
from pytest_mock import MockerFixture
from nonebot.adapters.onebot.v11 import Bot
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.utils import get_response_json, _v11_group_message_event
def init_mocked_api(mocked_api: MockRouter) -> None:
mocked_api.get(
"https://cdn.jsdelivr.net/gh/zhenxun-org/zhenxun_bot_plugins/plugins.json",
name="basic_plugins",
).respond(200, json=get_response_json("basic_plugins.json"))
mocked_api.get(
"https://raw.githubusercontent.com/zhenxun-org/zhenxun_bot_plugins_index/index/plugins.json",
name="extra_plugins",
).respond(200, json=get_response_json("extra_plugins.json"))
mocked_api.get(
"https://api.github.com/repos/zhenxun-org/zhenxun_bot_plugins/contents/plugins/search_image?ref=main",
name="search_image_plugin_api",
).respond(200, json=get_response_json("search_image_plugin_api.json"))
mocked_api.get(
"https://raw.githubusercontent.com/zhenxun-org/zhenxun_bot_plugins/main/plugins/search_image/__init__.py",
name="search_image_plugin_file_init",
).respond(content=b"")
mocked_api.get(
"https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/",
name="github_sub_plugin_contents",
).respond(json=get_response_json("github_sub_plugin_contents.json"))
mocked_api.get(
"https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/github_sub?ref=main",
name="github_sub_plugin_api",
).respond(json=get_response_json("github_sub_plugin_api.json"))
mocked_api.get(
"https://raw.githubusercontent.com/xuanerwa/zhenxun_github_sub/main/github_sub/__init__.py",
name="github_sub_plugin_file_init",
).respond(content=b"")
async def test_add_plugin_basic(
app: App,
mocker: MockerFixture,
mocked_api: MockRouter,
create_bot: Callable,
tmp_path: Path,
) -> None:
"""
测试添加基础插件
"""
from zhenxun.builtin_plugins.plugin_store import _matcher
init_mocked_api(mocked_api=mocked_api)
mocker.patch(
"zhenxun.builtin_plugins.plugin_store.data_source.BASE_PATH",
return_value=tmp_path / "zhenxun",
)
plugin_id = 1
async with app.test_matcher(_matcher) as ctx:
bot = create_bot(ctx)
bot: Bot = cast(Bot, bot)
raw_message = f"添加插件 {plugin_id}"
event: GroupMessageEvent = _v11_group_message_event(
message=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=bot, event=event)
ctx.should_call_send(
event=event,
message=Message(message=f"正在添加插件 Id: {plugin_id}"),
result=None,
bot=bot,
)
ctx.should_call_send(
event=event,
message=Message(message="插件 识图 安装成功! 重启后生效"),
result=None,
bot=bot,
)
assert mocked_api["basic_plugins"].called
assert mocked_api["extra_plugins"].called
assert mocked_api["search_image_plugin_api"].called
assert mocked_api["search_image_plugin_file_init"].called
async def test_add_plugin_extra(
app: App,
mocker: MockerFixture,
mocked_api: MockRouter,
create_bot: Callable,
tmp_path: Path,
) -> None:
"""
测试添加额外插件
"""
from zhenxun.builtin_plugins.plugin_store import _matcher
init_mocked_api(mocked_api=mocked_api)
mocker.patch(
"zhenxun.builtin_plugins.plugin_store.data_source.BASE_PATH",
return_value=tmp_path / "zhenxun",
)
plugin_id = 3
async with app.test_matcher(_matcher) as ctx:
bot = create_bot(ctx)
bot: Bot = cast(Bot, bot)
raw_message: str = f"添加插件 {plugin_id}"
event: GroupMessageEvent = _v11_group_message_event(
message=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=bot, event=event)
ctx.should_call_send(
event=event,
message=Message(message=f"正在添加插件 Id: {plugin_id}"),
result=None,
bot=bot,
)
ctx.should_call_send(
event=event,
message=Message(message="插件 github订阅 安装成功! 重启后生效"),
result=None,
bot=bot,
)
assert mocked_api["basic_plugins"].called
assert mocked_api["extra_plugins"].called
assert mocked_api["github_sub_plugin_contents"].called
assert mocked_api["github_sub_plugin_api"].called
assert mocked_api["github_sub_plugin_file_init"].called

20
tests/config.py Normal file
View File

@ -0,0 +1,20 @@
class BotId:
QQ_BOT = 12345
class UserId:
SUPERUSER = 10000
SUPERUSER_QQ = 11000
SUPERUSER_DODO = 12000
USER_LEVEL_0 = 10010
USER_LEVEL_5 = 10005
class GroupId:
GROUP_ID_LEVEL_0 = 20000
GROUP_ID_LEVEL_5 = 20005
class MessageId:
MESSAGE_ID = 30001
MESSAGE_ID_2 = 30002

View File

@ -1,37 +1,99 @@
import json
from pathlib import Path from pathlib import Path
from collections.abc import Callable
import nonebot
import pytest import pytest
from nonebot.plugin import Plugin import nonebot
from nonebug import NONEBOT_INIT_KWARGS
from nonebug.app import App from nonebug.app import App
from pytest_mock import MockerFixture
from respx import MockRouter from respx import MockRouter
from pytest_mock import MockerFixture
from nonebug import NONEBOT_INIT_KWARGS
from nonebug.mixin.process import MatcherContext
from tests.config import BotId, UserId
nonebot.load_plugin("nonebot_plugin_session")
def get_response_json(path: str) -> dict:
return json.loads(
(Path(__file__).parent / "response" / path).read_text(encoding="utf8")
)
def pytest_configure(config: pytest.Config) -> None: def pytest_configure(config: pytest.Config) -> None:
config.stash[NONEBOT_INIT_KWARGS] = { config.stash[NONEBOT_INIT_KWARGS] = {
"driver": "~fastapi+~httpx+~websockets", "driver": "~fastapi+~httpx+~websockets",
"superusers": ["AkashiCoin"], "superusers": [UserId.SUPERUSER.__str__()],
"command_start": "", "command_start": [""],
"session_running_expression": "别急呀,小真寻要宕机了!QAQ", "session_running_expression": "别急呀,小真寻要宕机了!QAQ",
"image_to_bytes": False, "image_to_bytes": False,
"nickname": ["真寻", "小真寻", "绪山真寻", "小寻子"], "nickname": ["真寻", "小真寻", "绪山真寻", "小寻子"],
"session_expire_timeout": 30, "session_expire_timeout": 30,
"self_nickname": "小真寻", "self_nickname": "小真寻",
"db_url": "sqlite://:memory:", "db_url": "sqlite://:memory:",
"platform_superusers": {"qq": ["qq_su"], "dodo": ["dodo_su"]}, "platform_superusers": {
"qq": [UserId.SUPERUSER_QQ.__str__()],
"dodo": [UserId.SUPERUSER_DODO.__str__()],
},
"host": "127.0.0.1", "host": "127.0.0.1",
"port": 8080, "port": 8080,
"log_level": "DEBUG",
} }
@pytest.fixture(scope="session", autouse=True) @pytest.fixture(scope="session", autouse=True)
def load_plugin(nonebug_init: None) -> set[Plugin]: def _init_bot(nonebug_init: None):
return nonebot.load_plugins("zhenxun.plugins") from nonebot.adapters.onebot.v11 import Adapter as OneBotV11Adapter
driver = nonebot.get_driver()
driver.register_adapter(OneBotV11Adapter)
nonebot.load_plugin("nonebot_plugin_alconna")
nonebot.load_plugin("nonebot_plugin_apscheduler")
nonebot.load_plugin("nonebot_plugin_userinfo")
nonebot.load_plugin("nonebot_plugin_htmlrender")
nonebot.load_plugins("zhenxun/builtin_plugins")
nonebot.load_plugins("zhenxun/plugins")
@pytest.fixture @pytest.fixture
async def app(app: App, tmp_path: Path, mocker: MockerFixture): async def app(app: App, tmp_path: Path, mocker: MockerFixture):
mocker.patch("nonebot.drivers.websockets.connect", return_value=MockRouter()) from zhenxun.services.db_context import init, disconnect
return app
driver = nonebot.get_driver()
# 清除连接钩子,现在 NoneBug 会自动触发 on_bot_connect
driver._bot_connection_hook.clear()
mock_config_path = mocker.MagicMock()
mock_config_path.LOG_PATH = tmp_path / "log"
# mock_config_path.LOG_PATH.mkdir(parents=True, exist_ok=True)
mock_config_path.DATA_PATH = tmp_path / "data"
# mock_config_path.DATA_PATH.mkdir(parents=True, exist_ok=True)
mock_config_path.TEMP_PATH = tmp_path / "resources" / "temp"
# mock_config_path.TEMP_PATH.mkdir(parents=True, exist_ok=True)
mocker.patch("zhenxun.configs.path_config", return_value=mock_config_path)
await init()
yield app
await disconnect()
@pytest.fixture
def create_bot() -> Callable:
from nonebot.adapters.onebot.v11 import Bot, Adapter
def _create_bot(context: MatcherContext):
return context.create_bot(
base=Bot,
adapter=nonebot.get_adapter(Adapter),
self_id=BotId.QQ_BOT.__str__(),
)
return _create_bot
@pytest.fixture
def mocked_api(respx_mock: MockRouter):
return respx_mock

Binary file not shown.

View File

@ -0,0 +1,32 @@
{
"鸡汤": {
"module": "jitang",
"module_path": "plugins.alapi.jitang",
"description": "喏,亲手为你煮的鸡汤",
"usage": "不喝点什么感觉有点不舒服\n 指令:\n 鸡汤",
"author": "HibiKier",
"version": "0.1",
"plugin_type": "NORMAL",
"is_dir": false
},
"识图": {
"module": "search_image",
"module_path": "plugins.search_image",
"description": "以图搜图,看破本源",
"usage": "识别图片 [二次元图片]\n 指令:\n 识图 [图片]",
"author": "HibiKier",
"version": "0.1",
"plugin_type": "NORMAL",
"is_dir": true
},
"网易云热评": {
"module": "comments_163",
"module_path": "plugins.alapi.comments_163",
"description": "生了个人,我很抱歉",
"usage": "到点了,还是防不了下塔\n 指令:\n 网易云热评/到点了/12点了",
"author": "HibiKier",
"version": "0.1",
"plugin_type": "NORMAL",
"is_dir": false
}
}

View File

@ -0,0 +1,24 @@
{
"github订阅": {
"module": "github_sub",
"module_path": "github_sub",
"description": "订阅github用户或仓库",
"usage": "usage\n github新CommentPRIssue等提醒\n 指令:\n 添加github ['用户'/'仓库'] [用户名/{owner/repo}]\n 删除github [用户名/{owner/repo}]\n 查看github\n 示例添加github订阅 用户 HibiKier\n 示例添加gb订阅 仓库 HibiKier/zhenxun_bot\n 示例添加github 用户 HibiKier\n 示例删除gb订阅 HibiKier",
"author": "xuanerwa",
"version": "0.7",
"plugin_type": "NORMAL",
"is_dir": true,
"github_url": "https://github.com/xuanerwa/zhenxun_github_sub"
},
"Minecraft查服": {
"module": "mc_check",
"module_path": "mc_check",
"description": "Minecraft服务器状态查询支持IPv6",
"usage": "Minecraft服务器状态查询支持IPv6\n用法\n\t查服 [ip]:[端口] / 查服 [ip]\n\t设置语言 zh-cn\n\t当前语言\n\t语言列表\neg:\t\nmcheck ip:port / mcheck ip\n\tset_lang en\n\tlang_now\n\tlang_list",
"author": "molanp",
"version": "1.13",
"plugin_type": "NORMAL",
"is_dir": true,
"github_url": "https://github.com/molanp/zhenxun_check_Minecraft"
}
}

View File

@ -0,0 +1,18 @@
[
{
"name": "__init__.py",
"path": "github_sub/__init__.py",
"sha": "7d17fd49fe82fa3897afcef61b2c694ed93a4ba3",
"size": 7551,
"url": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/github_sub/__init__.py?ref=main",
"html_url": "https://github.com/xuanerwa/zhenxun_github_sub/blob/main/github_sub/__init__.py",
"git_url": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/git/blobs/7d17fd49fe82fa3897afcef61b2c694ed93a4ba3",
"download_url": "https://raw.githubusercontent.com/xuanerwa/zhenxun_github_sub/main/github_sub/__init__.py",
"type": "file",
"_links": {
"self": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/github_sub/__init__.py?ref=main",
"git": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/git/blobs/7d17fd49fe82fa3897afcef61b2c694ed93a4ba3",
"html": "https://github.com/xuanerwa/zhenxun_github_sub/blob/main/github_sub/__init__.py"
}
}
]

View File

@ -0,0 +1,18 @@
[
{
"name": "github_sub",
"path": "github_sub",
"sha": "0f7d76bcf472e2ab0610fa542b067633d6e3ae7e",
"size": 0,
"url": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/github_sub?ref=main",
"html_url": "https://github.com/xuanerwa/zhenxun_github_sub/tree/main/github_sub",
"git_url": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/git/trees/0f7d76bcf472e2ab0610fa542b067633d6e3ae7e",
"download_url": null,
"type": "dir",
"_links": {
"self": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/contents/github_sub?ref=main",
"git": "https://api.github.com/repos/xuanerwa/zhenxun_github_sub/git/trees/0f7d76bcf472e2ab0610fa542b067633d6e3ae7e",
"html": "https://github.com/xuanerwa/zhenxun_github_sub/tree/main/github_sub"
}
}
]

View File

@ -0,0 +1,39 @@
{
"url": "https://api.github.com/repos/HibiKier/zhenxun_bot/releases/172632135",
"assets_url": "https://api.github.com/repos/HibiKier/zhenxun_bot/releases/172632135/assets",
"upload_url": "https://uploads.github.com/repos/HibiKier/zhenxun_bot/releases/172632135/assets{?name,label}",
"html_url": "https://github.com/HibiKier/zhenxun_bot/releases/tag/v0.2.2",
"id": 172632135,
"author": {
"login": "HibiKier",
"id": 45528451,
"node_id": "MDQ6VXNlcjQ1NTI4NDUx",
"avatar_url": "https://avatars.githubusercontent.com/u/45528451?v=4",
"gravatar_id": "",
"url": "https://api.github.com/users/HibiKier",
"html_url": "https://github.com/HibiKier",
"followers_url": "https://api.github.com/users/HibiKier/followers",
"following_url": "https://api.github.com/users/HibiKier/following{/other_user}",
"gists_url": "https://api.github.com/users/HibiKier/gists{/gist_id}",
"starred_url": "https://api.github.com/users/HibiKier/starred{/owner}{/repo}",
"subscriptions_url": "https://api.github.com/users/HibiKier/subscriptions",
"organizations_url": "https://api.github.com/users/HibiKier/orgs",
"repos_url": "https://api.github.com/users/HibiKier/repos",
"events_url": "https://api.github.com/users/HibiKier/events{/privacy}",
"received_events_url": "https://api.github.com/users/HibiKier/received_events",
"type": "User",
"site_admin": false
},
"node_id": "RE_kwDOFe9cjs4KSihH",
"tag_name": "v0.2.2",
"target_commitish": "main",
"name": "v0.2.2",
"draft": false,
"prerelease": false,
"created_at": "2024-08-29T18:48:53Z",
"published_at": "2024-08-29T18:49:34Z",
"assets": [],
"tarball_url": "https://api.github.com/repos/HibiKier/zhenxun_bot/tarball/v0.2.2",
"zipball_url": "https://api.github.com/repos/HibiKier/zhenxun_bot/zipball/v0.2.2",
"body": "* 集成webui\r\n* 移除plugins\r\n* 签到和帮助ui改版\r\n* 修改bug"
}

View File

@ -0,0 +1,18 @@
[
{
"name": "__init__.py",
"path": "plugins/search_image/__init__.py",
"sha": "38e86de0caafe6c3e88d973fb5c4bc9d1430d213",
"size": 3010,
"url": "https://api.github.com/repos/zhenxun-org/zhenxun_bot_plugins/contents/plugins/search_image/__init__.py?ref=main",
"html_url": "https://github.com/zhenxun-org/zhenxun_bot_plugins/blob/main/plugins/search_image/__init__.py",
"git_url": "https://api.github.com/repos/zhenxun-org/zhenxun_bot_plugins/git/blobs/38e86de0caafe6c3e88d973fb5c4bc9d1430d213",
"download_url": "https://raw.githubusercontent.com/zhenxun-org/zhenxun_bot_plugins/main/plugins/search_image/__init__.py",
"type": "file",
"_links": {
"self": "https://api.github.com/repos/zhenxun-org/zhenxun_bot_plugins/contents/plugins/search_image/__init__.py?ref=main",
"git": "https://api.github.com/repos/zhenxun-org/zhenxun_bot_plugins/git/blobs/38e86de0caafe6c3e88d973fb5c4bc9d1430d213",
"html": "https://github.com/zhenxun-org/zhenxun_bot_plugins/blob/main/plugins/search_image/__init__.py"
}
}
]

65
tests/utils.py Normal file
View File

@ -0,0 +1,65 @@
import json
from pathlib import Path
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11 import Message, MessageSegment, GroupMessageEvent
def get_response_json(path: str) -> dict:
try:
return json.loads(
(Path(__file__).parent / "response" / path).read_text(encoding="utf8")
)
except (FileNotFoundError, json.JSONDecodeError) as e:
raise ValueError(f"Error reading or parsing JSON file: {e}") from e
def get_content_bytes(path: str) -> bytes:
try:
return (Path(__file__).parent / "content" / path).read_bytes()
except FileNotFoundError as e:
raise ValueError(f"Error reading file: {e}") from e
def _v11_group_message_event(
message: str,
self_id: int,
user_id: int,
group_id: int,
message_id: int,
to_me: bool = True,
) -> GroupMessageEvent:
return GroupMessageEvent(
time=1122,
self_id=self_id,
post_type="message",
sub_type="",
user_id=user_id,
message_id=message_id,
message=Message(message),
original_message=Message(message),
message_type="group",
raw_message=message,
font=1,
sender=Sender(user_id=user_id),
to_me=to_me,
group_id=group_id,
)
def _v11_private_message_send(
message: str,
user_id: int,
):
return {
"message_type": "private",
"user_id": user_id,
"message": [
MessageSegment(
type="text",
data={
"text": message,
},
)
],
}

View File

@ -117,14 +117,14 @@ def _file_handle(latest_version: str | None):
shutil.move(src_folder_path, dest_folder_path) shutil.move(src_folder_path, dest_folder_path)
else: else:
logger.debug(f"源文件夹不存在: {src_folder_path}", "检查更新") logger.debug(f"源文件夹不存在: {src_folder_path}", "检查更新")
if tf:
tf.close()
if download_file.exists(): if download_file.exists():
logger.debug(f"删除下载文件: {download_file}", "检查更新") logger.debug(f"删除下载文件: {download_file}", "检查更新")
download_file.unlink() download_file.unlink()
if extract_path.exists(): if extract_path.exists():
logger.debug(f"删除解压文件夹: {extract_path}", "检查更新") logger.debug(f"删除解压文件夹: {extract_path}", "检查更新")
shutil.rmtree(extract_path) shutil.rmtree(extract_path)
if tf:
tf.close()
if TMP_PATH.exists(): if TMP_PATH.exists():
shutil.rmtree(TMP_PATH) shutil.rmtree(TMP_PATH)
if latest_version: if latest_version:

View File

@ -86,8 +86,8 @@ async def download_file(url: str, _is: bool = False, api_url: str | None = None)
for download_url, path in data_list: for download_url, path in data_list:
if download_url and "." in path: if download_url and "." in path:
logger.debug(f"下载文件: {path}", "插件管理") logger.debug(f"下载文件: {path}", "插件管理")
base_path = "zhenxun/plugins/" if _is else "zhenxun/" base_path = BASE_PATH / "plugins" if _is else BASE_PATH
file = Path(f"{base_path}{path}") file = base_path / path
file.parent.mkdir(parents=True, exist_ok=True) file.parent.mkdir(parents=True, exist_ok=True)
r = await AsyncHttpx.get(download_url) r = await AsyncHttpx.get(download_url)
if r.status_code != 200: if r.status_code != 200: