suyiiyii/nonebot-bison
1
0
Fork 0
mirror of https://github.com/suyiiyii/nonebot-bison.git synced 2026-05-09 18:27:56 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'dev' into feat/db

Browse Source
This commit is contained in:
felinae98
2022-03-22 20:07:23 +08:00
parent 39a0d21e57 1fe04ddb99
commit ad80006844
24 changed files with 1123 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/platforms/static/arknights-detail-807
+24
View File
@@ -0,0 +1,24 @@
<!doctype html>
<html>
<head>
<meta charset="utf-8" />
<meta http-equiv="X-UA-Compatible" content="IE=edge" />
<meta name="viewport" content="width=device-width, initial-scale=1, minimum-scale=1" />
<meta name="keywords" content="明日方舟,明日方舟官网,明日方舟手游,二次元,明日方舟Arknights,魔物娘,战棋,策略,塔防,塔防RPG,Arknights,人外,Monster" />
<meta name="description" content="《明日方舟》是一款魔物主题的策略手游。在游戏中，玩家将管理一艘满载“ 魔物干员”的方舟，为调查来源神秘的矿石灾难而踏上旅途。在这个宽广而危机四伏的世界中，你或许会看到废土中的城市废墟，或许会看到仿若幻境的亚人国度，或许会遭遇无法解读的神秘，或许参与无比残酷的战争。在有关幻想与异种生命的世界中，体验史诗与想象，情感与牵绊！" />
<link rel="icon" href="data:;base64,=" />
<title>公告</title>
<link rel="stylesheet" href="../../assets/css/announcement.v_0_1_2.css" />
</head>
<body>
<div class="main">
<div class="container">
<div class="banner-image-container cover">
<a class="cover-jumper" href="uniwebview://move?target=recruit&amp;param1=NORM_24_0_1">
<img class="banner-image" src="https://ak.hycdn.cn/announce/images/20211209/ff886833a69b48e513ff911f64f6d881.JPG" />
</a>
</div>
</div>
</div>
</body>
</html>
tests/platforms/static/arknights_list_-1.json
+91
View File
@@ -0,0 +1,91 @@
{
"focusAnnounceId": "816",
"announceList": [
{
"announceId": "809",
"title": "冰原信使系列\n新装限时上架",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/809_1640060505.html",
"day": 21,
"month": 12,
"group": "ACTIVITY"
},
{
"announceId": "810",
"title": "寒武纪系列\n限时复刻上架",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/810_1640060511.html",
"day": 21,
"month": 12,
"group": "ACTIVITY"
},
{
"announceId": "806",
"title": "跨年欢庆·回首\n限时寻访说明",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/806_1639379808.html",
"day": 14,
"month": 12,
"group": "ACTIVITY"
},
{
"announceId": "802",
"title": "「制作组通讯」\n#15期",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/802_1638871766.html",
"day": 8,
"month": 12,
"group": "SYSTEM"
},
{
"announceId": "97",
"title": "新人寻访特惠\n必得六星干员",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/97_1606379786.html",
"day": 30,
"month": 4,
"group": "ACTIVITY"
},
{
"announceId": "95",
"title": "通关特定关卡\n赠送专属时装",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/95_1606379781.html",
"day": 30,
"month": 4,
"group": "ACTIVITY"
},
{
"announceId": "192",
"title": "《明日方舟》\n公测开启说明",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/192_1606379744.html",
"day": 30,
"month": 4,
"group": "SYSTEM"
},
{
"announceId": "98",
"title": "《明日方舟》\n公平运营申明",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/98_1638970453.html",
"day": 30,
"month": 4,
"group": "SYSTEM"
},
{
"announceId": "94",
"title": "常驻活动介绍",
"isWebUrl": true,
"webUrl": "https://ak.hycdn.cn/announce/IOS/announcement/94_1606379757.html",
"day": 30,
"month": 4,
"group": "ACTIVITY"
}
],
"extra": {
"enable": false,
"name": "额外活动"
}
}
tests/platforms/test_arknights.py
+60
View File
@@ -18,6 +18,11 @@ def arknights_list_0():
return get_json("arknights_list_0.json")
@pytest.fixture(scope="module")
def arknights_list__1():
return get_json("arknights_list_-1.json")
@pytest.fixture(scope="module")
def arknights_list_1():
return get_json("arknights_list_1.json")
@@ -36,6 +41,61 @@ def monster_siren_list_1():
@pytest.mark.asyncio
@respx.mock
async def test_fetch_new(
arknights,
dummy_user_subinfo,
arknights_list_0,
arknights_list__1,
monster_siren_list_0,
monster_siren_list_1,
):
ak_list_router = respx.get(
"https://ak-conf.hypergryph.com/config/prod/announce_meta/IOS/announcement.meta.json"
)
detail_router = respx.get(
"https://ak.hycdn.cn/announce/IOS/announcement/807_1640060583.html"
)
version_router = respx.get(
"https://ak-conf.hypergryph.com/config/prod/official/IOS/version"
)
preannouncement_router = respx.get(
"https://ak-conf.hypergryph.com/config/prod/announce_meta/IOS/preannouncement.meta.json"
)
monster_siren_router = respx.get("https://monster-siren.hypergryph.com/api/news")
ak_list_router.mock(return_value=Response(200, json=arknights_list__1))
detail_router.mock(
return_value=Response(200, text=get_file("arknights-detail-807"))
)
version_router.mock(
return_value=Response(200, json=get_json("arknights-version-0.json"))
)
preannouncement_router.mock(
return_value=Response(200, json=get_json("arknights-pre-0.json"))
)
monster_siren_router.mock(return_value=Response(200, json=monster_siren_list_0))
target = ""
res = await arknights.fetch_new_post(target, [dummy_user_subinfo])
assert ak_list_router.called
assert len(res) == 0
assert not detail_router.called
mock_data = arknights_list_0
ak_list_router.mock(return_value=Response(200, json=mock_data))
res3 = await arknights.fetch_new_post(target, [dummy_user_subinfo])
assert len(res3[0][1]) == 1
assert detail_router.called
post = res3[0][1][0]
assert post.target_type == "arknights"
assert post.text == ""
assert post.url == ""
assert post.target_name == "明日方舟游戏内公告"
assert len(post.pics) == 1
# assert(post.pics == ['https://ak-fs.hypergryph.com/announce/images/20210623/e6f49aeb9547a2278678368a43b95b07.jpg'])
print(res3[0][1])
r = await post.generate_messages()
@pytest.mark.render
@respx.mock
async def test_send_with_render(
arknights,
dummy_user_subinfo,
arknights_list_0,
tests/test_config_manager_abort.py
+283
View File
@@ -0,0 +1,283 @@
import pytest
import respx
from httpx import Response
from nonebug.app import App
from .platforms.utils import get_json
from .utils import BotReply, fake_admin_user, fake_group_message_event
# 选择platform阶段中止
@pytest.mark.asyncio
@respx.mock
async def test_abort_add_on_platform(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
ak_list_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=1005056279793937"
)
ak_list_router.mock(
return_value=Response(200, json=get_json("weibo_ak_profile.json"))
)
ak_list_bad_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=100505000"
)
ak_list_bad_router.mock(
return_value=Response(200, json=get_json("weibo_err_profile.json"))
)
async with app.test_matcher(add_sub_matcher) as ctx:
bot = ctx.create_bot()
event_1 = fake_group_message_event(
message=Message("添加订阅"),
sender=Sender(card="", nickname="test", role="admin"),
to_me=True,
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
Message(BotReply.add_reply_on_platform(platform_manager, common_platform)),
True,
)
event_abort = fake_group_message_event(
message=Message("取消"), sender=Sender(card="", nickname="test", role="admin")
)
ctx.receive_event(bot, event_abort)
ctx.should_call_send(
event_abort,
BotReply.add_reply_abort,
True,
)
ctx.should_finished()
# 输入id阶段中止
@pytest.mark.asyncio
@respx.mock
async def test_abort_add_on_id(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
ak_list_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=1005056279793937"
)
ak_list_router.mock(
return_value=Response(200, json=get_json("weibo_ak_profile.json"))
)
ak_list_bad_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=100505000"
)
ak_list_bad_router.mock(
return_value=Response(200, json=get_json("weibo_err_profile.json"))
)
async with app.test_matcher(add_sub_matcher) as ctx:
bot = ctx.create_bot()
event_1 = fake_group_message_event(
message=Message("添加订阅"),
sender=Sender(card="", nickname="test", role="admin"),
to_me=True,
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
Message(BotReply.add_reply_on_platform(platform_manager, common_platform)),
True,
)
event_2 = fake_group_message_event(
message=Message("weibo"), sender=fake_admin_user
)
ctx.receive_event(bot, event_2)
ctx.should_call_send(
event_2,
Message(BotReply.add_reply_on_id),
True,
)
event_abort = fake_group_message_event(
message=Message("取消"), sender=Sender(card="", nickname="test", role="admin")
)
ctx.receive_event(bot, event_abort)
ctx.should_call_send(
event_abort,
BotReply.add_reply_abort,
True,
)
ctx.should_finished()
# 输入订阅类别阶段中止
@pytest.mark.asyncio
@respx.mock
async def test_abort_add_on_cats(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
ak_list_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=1005056279793937"
)
ak_list_router.mock(
return_value=Response(200, json=get_json("weibo_ak_profile.json"))
)
ak_list_bad_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=100505000"
)
ak_list_bad_router.mock(
return_value=Response(200, json=get_json("weibo_err_profile.json"))
)
async with app.test_matcher(add_sub_matcher) as ctx:
bot = ctx.create_bot()
event_1 = fake_group_message_event(
message=Message("添加订阅"),
sender=Sender(card="", nickname="test", role="admin"),
to_me=True,
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
Message(
BotReply.add_reply_on_platform(
platform_manager=platform_manager, common_platform=common_platform
)
),
True,
)
event_2 = fake_group_message_event(
message=Message("weibo"), sender=fake_admin_user
)
ctx.receive_event(bot, event_2)
ctx.should_call_send(
event_2,
Message(BotReply.add_reply_on_id),
True,
)
event_3 = fake_group_message_event(
message=Message("6279793937"), sender=fake_admin_user
)
ctx.receive_event(bot, event_3)
ctx.should_call_send(
event_3,
BotReply.add_reply_on_target_confirm(
"weibo", "明日方舟Arknights", "6279793937"
),
True,
)
ctx.should_call_send(
event_3,
Message(BotReply.add_reply_on_cats(platform_manager, "weibo")),
True,
)
event_abort = fake_group_message_event(
message=Message("取消"), sender=Sender(card="", nickname="test", role="admin")
)
ctx.receive_event(bot, event_abort)
ctx.should_call_send(
event_abort,
BotReply.add_reply_abort,
True,
)
ctx.should_finished()
# 输入标签阶段中止
@pytest.mark.asyncio
@respx.mock
async def test_abort_add_on_tag(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
ak_list_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=1005056279793937"
)
ak_list_router.mock(
return_value=Response(200, json=get_json("weibo_ak_profile.json"))
)
ak_list_bad_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=100505000"
)
ak_list_bad_router.mock(
return_value=Response(200, json=get_json("weibo_err_profile.json"))
)
async with app.test_matcher(add_sub_matcher) as ctx:
bot = ctx.create_bot()
event_1 = fake_group_message_event(
message=Message("添加订阅"),
sender=Sender(card="", nickname="test", role="admin"),
to_me=True,
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
Message(
BotReply.add_reply_on_platform(
platform_manager=platform_manager, common_platform=common_platform
)
),
True,
)
event_2 = fake_group_message_event(
message=Message("weibo"), sender=fake_admin_user
)
ctx.receive_event(bot, event_2)
ctx.should_call_send(
event_2,
Message(BotReply.add_reply_on_id),
True,
)
event_3 = fake_group_message_event(
message=Message("6279793937"), sender=fake_admin_user
)
ctx.receive_event(bot, event_3)
ctx.should_call_send(
event_3,
BotReply.add_reply_on_target_confirm(
"weibo", "明日方舟Arknights", "6279793937"
),
True,
)
ctx.should_call_send(
event_3,
Message(BotReply.add_reply_on_cats(platform_manager, "weibo")),
True,
)
event_4 = fake_group_message_event(
message=Message("图文 文字"), sender=fake_admin_user
)
ctx.receive_event(bot, event_4)
ctx.should_call_send(event_4, Message(BotReply.add_reply_on_tags), True)
event_abort = fake_group_message_event(
message=Message("取消"), sender=Sender(card="", nickname="test", role="admin")
)
ctx.receive_event(bot, event_abort)
ctx.should_call_send(
event_abort,
BotReply.add_reply_abort,
True,
)
ctx.should_finished()
tests/test_config_manager.py → tests/test_config_manager_add.py
+108 -164
View File
@@ -4,7 +4,7 @@ from httpx import Response
from nonebug.app import App
from .platforms.utils import get_json
from .utils import fake_admin_user, fake_group_message_event
from .utils import BotReply, fake_admin_user, fake_group_message_event
@pytest.mark.asyncio
@@ -49,18 +49,7 @@ async def test_configurable_at_me_false(app: App):
ctx.receive_event(bot, event)
ctx.should_call_send(
event,
Message(
"请输入想要订阅的平台，目前支持，请输入冒号左边的名称：\n"
+ "".join(
[
"{}：{}\n".format(
platform_name, platform_manager[platform_name].name
)
for platform_name in common_platform
]
)
+ "要查看全部平台请输入：“全部”"
),
Message(BotReply.add_reply_on_platform(platform_manager, common_platform)),
True,
)
ctx.should_pass_rule()
@@ -72,10 +61,11 @@ async def test_configurable_at_me_false(app: App):
async def test_add_with_target(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import config
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
ak_list_router = respx.get(
@@ -103,16 +93,9 @@ async def test_add_with_target(app: App):
ctx.should_call_send(
event_1,
Message(
"请输入想要订阅的平台，目前支持，请输入冒号左边的名称：\n"
+ "".join(
[
"{}：{}\n".format(
platform_name, platform_manager[platform_name].name
)
for platform_name in common_platform
]
BotReply.add_reply_on_platform(
platform_manager=platform_manager, common_platform=common_platform
)
+ "要查看全部平台请输入：“全部”"
),
True,
)
@@ -123,15 +106,7 @@ async def test_add_with_target(app: App):
ctx.should_rejected()
ctx.should_call_send(
event_2,
(
"全部平台\n"
+ "\n".join(
[
"{}：{}".format(platform_name, platform.name)
for platform_name, platform in platform_manager.items()
]
)
),
BotReply.add_reply_on_platform_input_allplatform(platform_manager),
True,
)
event_3 = fake_group_message_event(
@@ -140,16 +115,14 @@ async def test_add_with_target(app: App):
ctx.receive_event(bot, event_3)
ctx.should_call_send(
event_3,
Message(
"请输入订阅用户的id，详情查阅https://nonebot-bison.vercel.app/usage/#%E6%89%80%E6%94%AF%E6%8C%81%E5%B9%B3%E5%8F%B0%E7%9A%84uid"
),
Message(BotReply.add_reply_on_id),
True,
)
event_4_err = fake_group_message_event(
message=Message("000"), sender=fake_admin_user
)
ctx.receive_event(bot, event_4_err)
ctx.should_call_send(event_4_err, "id输入错误", True)
ctx.should_call_send(event_4_err, BotReply.add_reply_on_id_input_error, True)
ctx.should_rejected()
event_4_ok = fake_group_message_event(
message=Message("6279793937"), sender=fake_admin_user
@@ -157,29 +130,36 @@ async def test_add_with_target(app: App):
ctx.receive_event(bot, event_4_ok)
ctx.should_call_send(
event_4_ok,
Message(
"请输入要订阅的类别，以空格分隔，支持的类别有：{}".format(
" ".join(list(platform_manager["weibo"].categories.values()))
)
BotReply.add_reply_on_target_confirm(
"weibo", "明日方舟Arknights", "6279793937"
),
True,
)
ctx.should_call_send(
event_4_ok,
Message(BotReply.add_reply_on_cats(platform_manager, "weibo")),
True,
)
event_5_err = fake_group_message_event(
message=Message("图文 文字 err"), sender=fake_admin_user
)
ctx.receive_event(bot, event_5_err)
ctx.should_call_send(event_5_err, "不支持 err", True)
ctx.should_call_send(
event_5_err, BotReply.add_reply_on_cats_input_error("err"), True
)
ctx.should_rejected()
event_5_ok = fake_group_message_event(
message=Message("图文 文字"), sender=fake_admin_user
)
ctx.receive_event(bot, event_5_ok)
ctx.should_call_send(event_5_ok, Message('请输入要订阅的tag，订阅所有tag输入"全部标签"'), True)
ctx.should_call_send(event_5_ok, Message(BotReply.add_reply_on_tags), True)
event_6 = fake_group_message_event(
message=Message("全部标签"), sender=fake_admin_user
)
ctx.receive_event(bot, event_6)
ctx.should_call_send(event_6, ("添加 明日方舟Arknights 成功"), True)
ctx.should_call_send(
event_6, BotReply.add_reply_subscribe_success("明日方舟Arknights"), True
)
ctx.should_finished()
subs = config.list_subscribe(10000, "group")
assert len(subs) == 1
@@ -198,10 +178,11 @@ async def test_add_with_target(app: App):
async def test_add_with_target_no_cat(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import config
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
ncm_router = respx.get("https://music.163.com/api/artist/albums/32540734")
@@ -218,18 +199,7 @@ async def test_add_with_target_no_cat(app: App):
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
Message(
"请输入想要订阅的平台，目前支持，请输入冒号左边的名称：\n"
+ "".join(
[
"{}：{}\n".format(
platform_name, platform_manager[platform_name].name
)
for platform_name in common_platform
]
)
+ "要查看全部平台请输入：“全部”"
),
Message(BotReply.add_reply_on_platform(platform_manager, common_platform)),
True,
)
event_3 = fake_group_message_event(
@@ -238,16 +208,21 @@ async def test_add_with_target_no_cat(app: App):
ctx.receive_event(bot, event_3)
ctx.should_call_send(
event_3,
Message(
"请输入订阅用户的id，详情查阅https://nonebot-bison.vercel.app/usage/#%E6%89%80%E6%94%AF%E6%8C%81%E5%B9%B3%E5%8F%B0%E7%9A%84uid"
),
Message(BotReply.add_reply_on_id),
True,
)
event_4_ok = fake_group_message_event(
message=Message("32540734"), sender=fake_admin_user
)
ctx.receive_event(bot, event_4_ok)
ctx.should_call_send(event_4_ok, ("添加 塞壬唱片-MSR 成功"), True)
ctx.should_call_send(
event_4_ok,
BotReply.add_reply_on_target_confirm("ncm-artist", "塞壬唱片-MSR", "32540734"),
True,
)
ctx.should_call_send(
event_4_ok, BotReply.add_reply_subscribe_success("塞壬唱片-MSR"), True
)
ctx.should_finished()
subs = config.list_subscribe(10000, "group")
assert len(subs) == 1
@@ -264,10 +239,11 @@ async def test_add_with_target_no_cat(app: App):
async def test_add_no_target(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import config
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
async with app.test_matcher(add_sub_matcher) as ctx:
@@ -281,18 +257,7 @@ async def test_add_no_target(app: App):
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
Message(
"请输入想要订阅的平台，目前支持，请输入冒号左边的名称：\n"
+ "".join(
[
"{}：{}\n".format(
platform_name, platform_manager[platform_name].name
)
for platform_name in common_platform
]
)
+ "要查看全部平台请输入：“全部”"
),
Message(BotReply.add_reply_on_platform(platform_manager, common_platform)),
True,
)
event_3 = fake_group_message_event(
@@ -301,18 +266,16 @@ async def test_add_no_target(app: App):
ctx.receive_event(bot, event_3)
ctx.should_call_send(
event_3,
Message(
"请输入要订阅的类别，以空格分隔，支持的类别有：{}".format(
" ".join(list(platform_manager["arknights"].categories.values()))
)
),
Message(BotReply.add_reply_on_cats(platform_manager, "arknights")),
True,
)
event_4 = fake_group_message_event(
message=Message("游戏公告"), sender=fake_admin_user
)
ctx.receive_event(bot, event_4)
ctx.should_call_send(event_4, ("添加 明日方舟游戏信息 成功"), True)
ctx.should_call_send(
event_4, BotReply.add_reply_subscribe_success("明日方舟游戏信息"), True
)
ctx.should_finished()
subs = config.list_subscribe(10000, "group")
assert len(subs) == 1
@@ -328,10 +291,11 @@ async def test_add_no_target(app: App):
async def test_platform_name_err(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import config
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
async with app.test_matcher(add_sub_matcher) as ctx:
bot = ctx.create_bot()
@@ -344,18 +308,7 @@ async def test_platform_name_err(app: App):
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
Message(
"请输入想要订阅的平台，目前支持，请输入冒号左边的名称：\n"
+ "".join(
[
"{}：{}\n".format(
platform_name, platform_manager[platform_name].name
)
for platform_name in common_platform
]
)
+ "要查看全部平台请输入：“全部”"
),
Message(BotReply.add_reply_on_platform(platform_manager, common_platform)),
True,
)
event_2 = fake_group_message_event(
@@ -366,98 +319,89 @@ async def test_platform_name_err(app: App):
ctx.should_rejected()
ctx.should_call_send(
event_2,
"平台输入错误",
BotReply.add_reply_on_platform_input_error,
True,
)
@pytest.mark.asyncio
async def test_query_sub(app: App):
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import config
from nonebot_bison.config_manager import query_sub_matcher
@respx.mock
async def test_add_with_get_id(app: App):
from nonebot.adapters.onebot.v11.event import Sender
from nonebot.adapters.onebot.v11.message import Message, MessageSegment
from nonebot_bison.config import Config
from nonebot_bison.config_manager import add_sub_matcher, common_platform
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
config.add_subscribe(
10000,
"group",
"6279793937",
"明日方舟Arknights",
"weibo",
[platform_manager["weibo"].reverse_category["图文"]],
["明日方舟"],
ak_list_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=1005056279793937"
)
async with app.test_matcher(query_sub_matcher) as ctx:
ak_list_router.mock(
return_value=Response(200, json=get_json("weibo_ak_profile.json"))
)
ak_list_bad_router = respx.get(
"https://m.weibo.cn/api/container/getIndex?containerid=100505000"
)
ak_list_bad_router.mock(
return_value=Response(200, json=get_json("weibo_err_profile.json"))
)
async with app.test_matcher(add_sub_matcher) as ctx:
bot = ctx.create_bot()
event = fake_group_message_event(message=Message("查询订阅"), to_me=True)
ctx.receive_event(bot, event)
ctx.should_pass_rule()
ctx.should_pass_permission()
ctx.should_call_send(
event, Message("订阅的帐号为：\nweibo 明日方舟Arknights 6279793937 [图文] 明日方舟\n"), True
event_1 = fake_group_message_event(
message=Message("添加订阅"),
sender=Sender(card="", nickname="test", role="admin"),
to_me=True,
)
@pytest.mark.asyncio
async def test_del_sub(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import config
from nonebot_bison.config_manager import del_sub_matcher
from nonebot_bison.platform import platform_manager
config.user_target.truncate()
config.add_subscribe(
10000,
"group",
"6279793937",
"明日方舟Arknights",
"weibo",
[platform_manager["weibo"].reverse_category["图文"]],
["明日方舟"],
)
async with app.test_matcher(del_sub_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
assert isinstance(bot, Bot)
event = fake_group_message_event(
message=Message("删除订阅"), to_me=True, sender=fake_admin_user
)
ctx.receive_event(bot, event)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_pass_permission()
ctx.should_call_send(
event,
event_1,
Message(
"订阅的帐号为：\n1 weibo 明日方舟Arknights 6279793937\n [图文] 明日方舟\n请输入要删除的订阅的序号"
BotReply.add_reply_on_platform(
platform_manager=platform_manager, common_platform=common_platform
)
),
True,
)
event_1_err = fake_group_message_event(
message=Message("2"), sender=fake_admin_user
event_3 = fake_group_message_event(
message=Message("weibo"), sender=fake_admin_user
)
ctx.receive_event(bot, event_1_err)
ctx.should_call_send(event_1_err, "删除错误", True)
ctx.receive_event(bot, event_3)
ctx.should_call_send(
event_3,
Message(BotReply.add_reply_on_id),
True,
)
event_4_query = fake_group_message_event(
message=Message("查询"), sender=fake_admin_user
)
ctx.receive_event(bot, event_4_query)
ctx.should_rejected()
event_1_ok = fake_group_message_event(
message=Message("1"), sender=fake_admin_user
ctx.should_call_send(
event_4_query,
Message([MessageSegment(*BotReply.add_reply_on_id_input_search())]),
True,
)
"""
line 362:
鬼知道为什么要在这里这样写，
没有[]的话assert不了(should_call_send使用[MessageSegment(...)]的格式进行比较)
不在这里MessageSegment()的话也assert不了(指不能让add_reply_on_id_input_search直接返回一个MessageSegment对象)
amen
"""
event_abort = fake_group_message_event(
message=Message("取消"), sender=Sender(card="", nickname="test", role="admin")
)
ctx.receive_event(bot, event_abort)
ctx.should_call_send(
event_abort,
BotReply.add_reply_abort,
True,
)
ctx.receive_event(bot, event_1_ok)
ctx.should_call_send(event_1_ok, "删除成功", True)
ctx.should_finished()
subs = config.list_subscribe(10000, "group")
assert len(subs) == 0
async def test_test(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config_manager import test_matcher
async with app.test_matcher(test_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event = fake_group_message_event(message=Message("testtt"))
ctx.receive_event(bot, event)
ctx.should_pass_permission()
ctx.should_pass_rule()
ctx.should_call_send(event, "666", True)
tests/test_config_manager_admin.py
+48 -9
View File
@@ -1,9 +1,11 @@
import pytest
from nonebug import App
from .utils import fake_admin_user, fake_private_message_event, fake_superuser
from .utils import fake_group_message_event, fake_private_message_event, fake_superuser
async def test_query(app: App):
@pytest.mark.asyncio
async def test_query_with_superuser_private(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config_manager import group_manage_matcher
@@ -11,7 +13,10 @@ async def test_query(app: App):
async with app.test_matcher(group_manage_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event = fake_private_message_event(
message=Message("群管理"), sender=fake_superuser
message=Message("群管理"),
sender=fake_superuser,
to_me=True,
user_id=fake_superuser.user_id,
)
ctx.receive_event(bot, event)
ctx.should_pass_rule()
@@ -23,23 +28,57 @@ async def test_query(app: App):
event, Message("请选择需要管理的群：\n1. 101 - test group\n请输入左侧序号"), True
)
event_1_err = fake_private_message_event(
message=Message("0"), sender=fake_superuser
message=Message("0"),
sender=fake_superuser,
to_me=True,
user_id=fake_superuser.user_id,
)
ctx.receive_event(bot, event_1_err)
ctx.should_rejected()
ctx.should_call_send(event_1_err, "请输入正确序号", True)
ctx.should_rejected()
event_1_ok = fake_private_message_event(
message=Message("1"), sender=fake_superuser
message=Message("1"),
sender=fake_superuser,
to_me=True,
user_id=fake_superuser.user_id,
)
ctx.receive_event(bot, event_1_ok)
ctx.should_call_send(event_1_ok, "请输入需要使用的命令：添加订阅，查询订阅，删除订阅", True)
event_2_err = fake_private_message_event(
message=Message("222"), sender=fake_superuser
message=Message("222"),
sender=fake_superuser,
to_me=True,
user_id=fake_superuser.user_id,
)
ctx.receive_event(bot, event_2_err)
ctx.should_rejected()
ctx.should_call_send(event_2_err, "请输入正确的命令", True)
ctx.should_rejected()
event_2_ok = fake_private_message_event(
message=Message("查询订阅"), sender=fake_superuser
message=Message("查询订阅"),
sender=fake_superuser,
to_me=True,
user_id=fake_superuser.user_id,
)
ctx.receive_event(bot, event_2_ok)
ctx.should_pass_rule()
ctx.should_pass_permission()
@pytest.mark.asyncio
async def test_query_with_superuser_group_tome(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config_manager import group_manage_matcher
async with app.test_matcher(group_manage_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event = fake_group_message_event(
message=Message("群管理"),
sender=fake_superuser,
to_me=True,
user_id=fake_superuser.user_id,
)
ctx.receive_event(bot, event)
ctx.should_pass_rule()
ctx.should_pass_permission()
ctx.should_call_send(event, Message("该功能只支持私聊使用，请私聊Bot"), True)
tests/test_config_manager_query_del.py
+87
View File
@@ -0,0 +1,87 @@
import pytest
import respx
from httpx import Response
from nonebug.app import App
from .platforms.utils import get_json
from .utils import fake_admin_user, fake_group_message_event
@pytest.mark.asyncio
async def test_query_sub(app: App):
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import Config
from nonebot_bison.config_manager import query_sub_matcher
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
config.add_subscribe(
10000,
"group",
"6279793937",
"明日方舟Arknights",
"weibo",
[platform_manager["weibo"].reverse_category["图文"]],
["明日方舟"],
)
async with app.test_matcher(query_sub_matcher) as ctx:
bot = ctx.create_bot()
event = fake_group_message_event(message=Message("查询订阅"), to_me=True)
ctx.receive_event(bot, event)
ctx.should_pass_rule()
ctx.should_pass_permission()
ctx.should_call_send(
event, Message("订阅的帐号为：\nweibo 明日方舟Arknights 6279793937 [图文] 明日方舟\n"), True
)
@pytest.mark.asyncio
async def test_del_sub(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.config import Config
from nonebot_bison.config_manager import del_sub_matcher
from nonebot_bison.platform import platform_manager
config = Config()
config.user_target.truncate()
config.add_subscribe(
10000,
"group",
"6279793937",
"明日方舟Arknights",
"weibo",
[platform_manager["weibo"].reverse_category["图文"]],
["明日方舟"],
)
async with app.test_matcher(del_sub_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
assert isinstance(bot, Bot)
event = fake_group_message_event(
message=Message("删除订阅"), to_me=True, sender=fake_admin_user
)
ctx.receive_event(bot, event)
ctx.should_pass_rule()
ctx.should_pass_permission()
ctx.should_call_send(
event,
Message(
"订阅的帐号为：\n1 weibo 明日方舟Arknights 6279793937\n [图文] 明日方舟\n请输入要删除的订阅的序号"
),
True,
)
event_1_err = fake_group_message_event(
message=Message("2"), sender=fake_admin_user
)
ctx.receive_event(bot, event_1_err)
ctx.should_call_send(event_1_err, "删除错误", True)
ctx.should_rejected()
event_1_ok = fake_group_message_event(
message=Message("1"), sender=fake_admin_user
)
ctx.receive_event(bot, event_1_ok)
ctx.should_call_send(event_1_ok, "删除成功", True)
ctx.should_finished()
subs = config.list_subscribe(10000, "group")
assert len(subs) == 0
tests/test_send.py
+161 -11
View File
@@ -1,13 +1,12 @@
import asyncio
import pytest
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebug import App
@pytest.mark.asyncio
async def test_send_no_queue(app: App):
import nonebot
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.send import send_msgs
@@ -16,25 +15,27 @@ async def test_send_no_queue(app: App):
bot = ctx.create_bot(base=Bot)
assert isinstance(bot, Bot)
ctx.should_call_api(
"send_group_msg", {"group_id": "1233", "message": "msg1"}, True
"send_group_msg", {"group_id": "1233", "message": Message("msg1")}, True
)
ctx.should_call_api(
"send_group_msg", {"group_id": "1233", "message": "msg2"}, True
"send_group_msg", {"group_id": "1233", "message": Message("msg2")}, True
)
ctx.should_call_api(
"send_private_msg", {"user_id": "666", "message": "priv"}, True
"send_private_msg", {"user_id": "666", "message": Message("priv")}, True
)
await send_msgs(bot, "1233", "group", ["msg1", "msg2"])
await send_msgs(bot, "666", "private", ["priv"])
await send_msgs(bot, "1233", "group", [Message("msg1"), Message("msg2")])
await send_msgs(bot, "666", "private", [Message("priv")])
assert ctx.wait_list.empty()
@pytest.mark.asyncio
async def test_send_queue(app: App):
import nonebot
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison import send
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.send import LAST_SEND_TIME, do_send_msgs, send_msgs
from nonebot_bison.send import do_send_msgs, send_msgs
async with app.test_api() as ctx:
new_bot = ctx.create_bot(base=Bot)
@@ -47,7 +48,7 @@ async def test_send_queue(app: App):
"send_group_msg", {"group_id": "1233", "message": "test msg"}, True
)
await bot.call_api("send_group_msg", group_id="1233", message="test msg")
await send_msgs(bot, "1233", "group", ["msg"])
await send_msgs(bot, "1233", "group", [Message("msg")])
ctx.should_call_api(
"send_group_msg", {"group_id": "1233", "message": "msg"}, True
)
@@ -56,3 +57,152 @@ async def test_send_queue(app: App):
app.monkeypatch.setattr(send, "LAST_SEND_TIME", 0, True)
await do_send_msgs()
assert ctx.wait_list.empty()
def gen_node(id, name, content: Message):
return {"type": "node", "data": {"name": name, "uin": id, "content": content}}
def _merge_messge(nodes):
return nodes
@pytest.mark.asyncio
async def test_send_merge_no_queue(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message, MessageSegment
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.send import send_msgs
plugin_config.bison_use_pic_merge = 1
plugin_config.bison_use_queue = False
async with app.test_api() as ctx:
bot = ctx.create_bot(base=Bot, self_id="8888")
assert isinstance(bot, Bot)
message = [
Message(MessageSegment.text("test msg")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
]
ctx.should_call_api(
"send_group_msg",
{"group_id": 633, "message": Message(MessageSegment.text("test msg"))},
None,
)
ctx.should_call_api(
"send_group_msg",
{"group_id": 633, "message": message[1]},
None,
)
await send_msgs(bot, 633, "group", message)
message = [
Message(MessageSegment.text("test msg")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
]
ctx.should_call_api(
"send_group_msg",
{"group_id": 633, "message": Message(MessageSegment.text("test msg"))},
None,
)
ctx.should_call_api(
"get_group_member_info",
{"group_id": 633, "user_id": 8888, "no_cache": True},
{"user_id": 8888, "card": "admin", "nickname": "adminuser"},
)
merged_message = _merge_messge(
[gen_node(8888, "admin", message[1]), gen_node(8888, "admin", message[2])]
)
ctx.should_call_api(
"send_group_forward_msg",
{"group_id": 633, "messages": merged_message},
None,
)
await send_msgs(bot, 633, "group", message)
message = [
Message(MessageSegment.text("test msg")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
]
ctx.should_call_api(
"send_group_msg",
{"group_id": 633, "message": Message(MessageSegment.text("test msg"))},
None,
)
ctx.should_call_api(
"get_group_member_info",
{"group_id": 633, "user_id": 8888, "no_cache": True},
{"user_id": 8888, "card": None, "nickname": "adminuser"},
)
merged_message = _merge_messge(
[
gen_node(8888, "adminuser", message[1]),
gen_node(8888, "adminuser", message[2]),
gen_node(8888, "adminuser", message[3]),
]
)
ctx.should_call_api(
"send_group_forward_msg",
{"group_id": 633, "messages": merged_message},
None,
)
await send_msgs(bot, 633, "group", message)
# private user should not send in forward
message = [
Message(MessageSegment.text("test msg")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
]
ctx.should_call_api(
"send_private_msg",
{"user_id": 633, "message": Message(MessageSegment.text("test msg"))},
None,
)
ctx.should_call_api(
"send_private_msg", {"user_id": 633, "message": message[1]}, None
)
ctx.should_call_api(
"send_private_msg", {"user_id": 633, "message": message[2]}, None
)
await send_msgs(bot, 633, "private", message)
async def test_send_merge2_no_queue(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message, MessageSegment
from nonebot_bison.plugin_config import plugin_config
from nonebot_bison.send import send_msgs
plugin_config.bison_use_pic_merge = 2
plugin_config.bison_use_queue = False
async with app.test_api() as ctx:
bot = ctx.create_bot(base=Bot, self_id="8888")
assert isinstance(bot, Bot)
message = [
Message(MessageSegment.text("test msg")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
Message(MessageSegment.image("https://picsum.photos/200/300")),
]
ctx.should_call_api(
"get_group_member_info",
{"group_id": 633, "user_id": 8888, "no_cache": True},
{"user_id": 8888, "card": "admin", "nickname": "adminuser"},
)
merged_message = _merge_messge(
[
gen_node(8888, "admin", message[0]),
gen_node(8888, "admin", message[1]),
gen_node(8888, "admin", message[2]),
]
)
ctx.should_call_api(
"send_group_forward_msg",
{"group_id": 633, "messages": merged_message},
None,
)
await send_msgs(bot, 633, "group", message)
tests/utils.py
+66
View File
@@ -69,3 +69,69 @@ from nonebot.adapters.onebot.v11.event import Sender
fake_admin_user = Sender(nickname="test", role="admin")
fake_superuser = Sender(user_id=10001, nickname="superuser")
class BotReply:
@staticmethod
def add_reply_on_platform(platform_manager, common_platform):
return (
"请输入想要订阅的平台，目前支持，请输入冒号左边的名称：\n"
+ "".join(
[
"{}：{}\n".format(
platform_name, platform_manager[platform_name].name
)
for platform_name in common_platform
]
)
+ "要查看全部平台请输入：“全部”\n中止订阅过程请输入：“取消”"
)
@staticmethod
def add_reply_on_platform_input_allplatform(platform_manager):
return "全部平台\n" + "\n".join(
[
"{}：{}".format(platform_name, platform.name)
for platform_name, platform in platform_manager.items()
]
)
@staticmethod
def add_reply_on_id_input_search():
search_url = "https://nonebot-bison.vercel.app/usage/#%E6%89%80%E6%94%AF%E6%8C%81%E5%B9%B3%E5%8F%B0%E7%9A%84-uid"
search_title = "Bison所支持的平台UID"
search_content = "查询相关平台的uid格式或获取方式"
search_image = "https://s3.bmp.ovh/imgs/2022/03/ab3cc45d83bd3dd3.jpg"
type = "share"
data = {
"url": search_url,
"title": search_title,
"content": search_content,
"image": search_image,
}
msg = [type, data]
return msg
@staticmethod
def add_reply_on_target_confirm(platform, name, id):
return f"即将订阅的用户为:{platform} {name} {id}\n如有错误请输入“取消”重新订阅"
@staticmethod
def add_reply_on_cats(platform_manager, platform: str):
return "请输入要订阅的类别，以空格分隔，支持的类别有：{}".format(
" ".join(list(platform_manager[platform].categories.values()))
)
@staticmethod
def add_reply_on_cats_input_error(cat: str):
return "不支持 {}".format(cat)
@staticmethod
def add_reply_subscribe_success(name):
return "添加 {} 成功".format(name)
add_reply_on_id_input_error = "id输入错误"
add_reply_on_platform_input_error = "平台输入错误"
add_reply_on_id = "请输入订阅用户的id:\n查询id获取方法请回复:“查询”"
add_reply_on_tags = '请输入要订阅的tag，订阅所有tag输入"全部标签"'
add_reply_abort = "已中止订阅"