mirror of
https://github.com/suyiiyii/nonebot-bison.git
synced 2026-05-13 20:38:39 +08:00
🚚 rename config_manager to sub_manager
This commit is contained in:
felinae98
@@ -0,0 +1,318 @@
|
||||
import pytest
|
||||
import respx
|
||||
from httpx import Response
|
||||
from nonebug.app import App
|
||||
from nonebug_saa import should_send_saa
|
||||
|
||||
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, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
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, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.weibo import Weibo
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
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,
|
||||
BotReply.add_reply_on_id(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_cats(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.weibo import Weibo
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
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,
|
||||
BotReply.add_reply_on_id(Weibo),
|
||||
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,
|
||||
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, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.weibo import Weibo
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
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,
|
||||
BotReply.add_reply_on_id(Weibo),
|
||||
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,
|
||||
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, 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()
|
||||
|
||||
|
||||
# 删除订阅阶段中止
|
||||
@pytest.mark.asyncio
|
||||
async def test_abort_del_sub(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.bot import Bot
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import MessageFactory, TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.sub_manager import del_sub_matcher
|
||||
from nonebot_bison.types import Target as T_Target
|
||||
|
||||
await config.add_subscribe(
|
||||
TargetQQGroup(group_id=10000),
|
||||
T_Target("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()
|
||||
should_send_saa(
|
||||
ctx,
|
||||
MessageFactory(
|
||||
"订阅的帐号为：\n1 weibo 明日方舟Arknights 6279793937\n [图文] 明日方舟\n请输入要删除的订阅的序号\n输入'取消'中止"
|
||||
),
|
||||
bot,
|
||||
event=event,
|
||||
)
|
||||
event_abort = fake_group_message_event(
|
||||
message=Message("取消"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_abort)
|
||||
ctx.should_call_send(event_abort, "删除中止", True)
|
||||
ctx.should_finished()
|
||||
subs = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert subs
|
||||
@@ -0,0 +1,722 @@
|
||||
import pytest
|
||||
import respx
|
||||
from httpx import Response
|
||||
from nonebug.app import App
|
||||
from nonebug_saa import should_send_saa
|
||||
from pytest_mock import MockerFixture
|
||||
|
||||
from ..platforms.utils import get_json
|
||||
from ..utils import (
|
||||
BotReply,
|
||||
add_reply_on_id_input_search,
|
||||
fake_admin_user,
|
||||
fake_group_message_event,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_configurable_at_me_true_failed(app: App):
|
||||
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.sub_manager import add_sub_matcher
|
||||
|
||||
plugin_config.bison_to_me = True
|
||||
async with app.test_matcher(add_sub_matcher) as ctx:
|
||||
bot = ctx.create_bot(base=Bot)
|
||||
event = fake_group_message_event(
|
||||
message=Message("添加订阅"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event)
|
||||
ctx.should_not_pass_rule()
|
||||
ctx.should_pass_permission()
|
||||
|
||||
async with app.test_matcher(add_sub_matcher) as ctx:
|
||||
bot = ctx.create_bot(base=Bot)
|
||||
event = fake_group_message_event(message=Message("添加订阅"), to_me=True)
|
||||
ctx.receive_event(bot, event)
|
||||
ctx.should_pass_rule()
|
||||
ctx.should_not_pass_permission()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_configurable_at_me_false(app: App):
|
||||
from nonebot.adapters.onebot.v11.bot import Bot
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.plugin_config import plugin_config
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
plugin_config.bison_to_me = False
|
||||
async with app.test_matcher(add_sub_matcher) as ctx:
|
||||
bot = ctx.create_bot(base=Bot)
|
||||
event = fake_group_message_event(
|
||||
message=Message("添加订阅"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event)
|
||||
ctx.should_call_send(
|
||||
event,
|
||||
BotReply.add_reply_on_platform(platform_manager, common_platform),
|
||||
True,
|
||||
)
|
||||
ctx.should_pass_rule()
|
||||
ctx.should_pass_permission()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_add_with_target(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.weibo import Weibo
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
BotReply.add_reply_on_platform(
|
||||
platform_manager=platform_manager, common_platform=common_platform
|
||||
),
|
||||
True,
|
||||
)
|
||||
event_2 = fake_group_message_event(
|
||||
message=Message("全部"), sender=Sender(card="", nickname="test", role="admin")
|
||||
)
|
||||
ctx.receive_event(bot, event_2)
|
||||
ctx.should_rejected()
|
||||
ctx.should_call_send(
|
||||
event_2,
|
||||
BotReply.add_reply_on_platform_input_allplatform(platform_manager),
|
||||
True,
|
||||
)
|
||||
event_3 = fake_group_message_event(
|
||||
message=Message("weibo"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_3)
|
||||
ctx.should_call_send(
|
||||
event_3,
|
||||
BotReply.add_reply_on_id(Weibo),
|
||||
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, 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
|
||||
)
|
||||
ctx.receive_event(bot, event_4_ok)
|
||||
ctx.should_call_send(
|
||||
event_4_ok,
|
||||
BotReply.add_reply_on_target_confirm(
|
||||
"weibo", "明日方舟Arknights", "6279793937"
|
||||
),
|
||||
True,
|
||||
)
|
||||
ctx.should_call_send(
|
||||
event_4_ok,
|
||||
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, 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, BotReply.add_reply_on_tags, True)
|
||||
event_6_more_info = fake_group_message_event(
|
||||
message=Message("详情"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_6_more_info)
|
||||
ctx.should_call_send(
|
||||
event_6_more_info, BotReply.add_reply_on_tags_need_more_info, True
|
||||
)
|
||||
ctx.should_rejected()
|
||||
event_6_ok = fake_group_message_event(
|
||||
message=Message("全部标签"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_6_ok)
|
||||
ctx.should_call_send(
|
||||
event_6_ok, BotReply.add_reply_subscribe_success("明日方舟Arknights"), True
|
||||
)
|
||||
ctx.should_finished()
|
||||
subs = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 1
|
||||
sub = subs[0]
|
||||
assert sub.target.target == "6279793937"
|
||||
assert sub.tags == []
|
||||
assert sub.categories == [platform_manager["weibo"].reverse_category["图文"]] + [
|
||||
platform_manager["weibo"].reverse_category["文字"]
|
||||
]
|
||||
assert sub.target.platform_name == "weibo"
|
||||
assert sub.target.target_name == "明日方舟Arknights"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_add_with_target_no_cat(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.ncm import NcmArtist
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
ncm_router = respx.get("https://music.163.com/api/artist/albums/32540734")
|
||||
ncm_router.mock(return_value=Response(200, json=get_json("ncm_siren.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,
|
||||
BotReply.add_reply_on_platform(platform_manager, common_platform),
|
||||
True,
|
||||
)
|
||||
event_3 = fake_group_message_event(
|
||||
message=Message("ncm-artist"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_3)
|
||||
ctx.should_call_send(
|
||||
event_3,
|
||||
BotReply.add_reply_on_id(NcmArtist),
|
||||
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,
|
||||
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 = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 1
|
||||
sub = subs[0]
|
||||
assert sub.target.target == "32540734"
|
||||
assert sub.tags == []
|
||||
assert sub.categories == []
|
||||
assert sub.target.platform_name == "ncm-artist"
|
||||
assert sub.target.target_name == "塞壬唱片-MSR"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_add_no_target(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
BotReply.add_reply_on_platform(platform_manager, common_platform),
|
||||
True,
|
||||
)
|
||||
event_3 = fake_group_message_event(
|
||||
message=Message("arknights"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_3)
|
||||
ctx.should_call_send(
|
||||
event_3,
|
||||
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, BotReply.add_reply_subscribe_success("明日方舟游戏信息"), True
|
||||
)
|
||||
ctx.should_finished()
|
||||
subs = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 1
|
||||
sub = subs[0]
|
||||
assert sub.target.target == "default"
|
||||
assert sub.tags == []
|
||||
assert sub.categories == [platform_manager["arknights"].reverse_category["游戏公告"]]
|
||||
assert sub.target.platform_name == "arknights"
|
||||
assert sub.target.target_name == "明日方舟游戏信息"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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.platform import platform_manager
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
BotReply.add_reply_on_platform(platform_manager, common_platform),
|
||||
True,
|
||||
)
|
||||
event_2 = fake_group_message_event(
|
||||
message=Message("error"),
|
||||
sender=Sender(card="", nickname="test", role="admin"),
|
||||
)
|
||||
ctx.receive_event(bot, event_2)
|
||||
ctx.should_rejected()
|
||||
ctx.should_call_send(
|
||||
event_2,
|
||||
BotReply.add_reply_on_platform_input_error,
|
||||
True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@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_plugin_saa import (
|
||||
Custom,
|
||||
MessageFactory,
|
||||
SupportedAdapters,
|
||||
TargetQQGroup,
|
||||
Text,
|
||||
)
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.weibo import Weibo
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
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,
|
||||
BotReply.add_reply_on_platform(
|
||||
platform_manager=platform_manager, common_platform=common_platform
|
||||
),
|
||||
True,
|
||||
)
|
||||
event_3 = fake_group_message_event(
|
||||
message=Message("weibo"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_3)
|
||||
ctx.should_call_send(
|
||||
event_3,
|
||||
BotReply.add_reply_on_id(Weibo),
|
||||
True,
|
||||
)
|
||||
event_4_query = fake_group_message_event(
|
||||
message=Message("查询"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_4_query)
|
||||
should_send_saa(
|
||||
ctx,
|
||||
MessageFactory(
|
||||
Text(add_reply_on_id_input_search).overwrite(
|
||||
SupportedAdapters.onebot_v11,
|
||||
BotReply.add_reply_on_id_input_search_ob11(),
|
||||
)
|
||||
),
|
||||
bot,
|
||||
event=event_4_query,
|
||||
)
|
||||
ctx.should_rejected()
|
||||
|
||||
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()
|
||||
await ctx.run()
|
||||
subs = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_add_with_bilibili_target_parser(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.bilibili import Bilibili
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
ak_list_router = respx.get(
|
||||
"https://api.bilibili.com/x/web-interface/card?mid=161775300"
|
||||
)
|
||||
ak_list_router.mock(
|
||||
return_value=Response(200, json=get_json("bilibili_arknights_profile.json"))
|
||||
)
|
||||
|
||||
bilibili_main_page_router = respx.get("https://www.bilibili.com/")
|
||||
bilibili_main_page_router.mock(return_value=Response(200))
|
||||
|
||||
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,
|
||||
BotReply.add_reply_on_platform(
|
||||
platform_manager=platform_manager, common_platform=common_platform
|
||||
),
|
||||
True,
|
||||
)
|
||||
event_2 = fake_group_message_event(
|
||||
message=Message("全部"), sender=Sender(card="", nickname="test", role="admin")
|
||||
)
|
||||
ctx.receive_event(bot, event_2)
|
||||
ctx.should_rejected()
|
||||
ctx.should_call_send(
|
||||
event_2,
|
||||
BotReply.add_reply_on_platform_input_allplatform(platform_manager),
|
||||
True,
|
||||
)
|
||||
event_3 = fake_group_message_event(
|
||||
message=Message("bilibili"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_3)
|
||||
assert Bilibili.parse_target_promot
|
||||
ctx.should_call_send(
|
||||
event_3,
|
||||
BotReply.add_reply_on_id(Bilibili),
|
||||
True,
|
||||
)
|
||||
event_4_err1 = fake_group_message_event(
|
||||
message=Message(
|
||||
"https://live.bilibili.com/5555734?broadcast_type=0&is_room_feed=1&spm_id_from=333.999.0.0"
|
||||
),
|
||||
sender=fake_admin_user,
|
||||
)
|
||||
ctx.receive_event(bot, event_4_err1)
|
||||
ctx.should_call_send(
|
||||
event_4_err1, BotReply.add_reply_on_target_parse_input_error, True
|
||||
)
|
||||
ctx.should_rejected()
|
||||
|
||||
event_4_err1 = fake_group_message_event(
|
||||
message=Message(
|
||||
"https://space.bilibili.com/ark161775300?from=search&seid=13051774060625135297&spm_id_from=333.337.0.0"
|
||||
),
|
||||
sender=fake_admin_user,
|
||||
)
|
||||
ctx.receive_event(bot, event_4_err1)
|
||||
ctx.should_call_send(
|
||||
event_4_err1, BotReply.add_reply_on_target_parse_input_error, True
|
||||
)
|
||||
ctx.should_rejected()
|
||||
|
||||
event_4_ok = fake_group_message_event(
|
||||
message=Message(
|
||||
"https://space.bilibili.com/161775300?from=search&seid=13051774060625135297&spm_id_from=333.337.0.0"
|
||||
),
|
||||
sender=fake_admin_user,
|
||||
)
|
||||
ctx.receive_event(bot, event_4_ok)
|
||||
ctx.should_call_send(
|
||||
event_4_ok,
|
||||
BotReply.add_reply_on_target_confirm("bilibili", "明日方舟", "161775300"),
|
||||
True,
|
||||
)
|
||||
ctx.should_call_send(
|
||||
event_4_ok,
|
||||
BotReply.add_reply_on_cats(platform_manager, "bilibili"),
|
||||
True,
|
||||
)
|
||||
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, 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, BotReply.add_reply_subscribe_success("明日方舟"), True
|
||||
)
|
||||
ctx.should_finished()
|
||||
subs = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 1
|
||||
sub = subs[0]
|
||||
assert sub.target.target == "161775300"
|
||||
assert sub.tags == []
|
||||
assert sub.categories == [platform_manager["bilibili"].reverse_category["视频"]]
|
||||
assert sub.target.platform_name == "bilibili"
|
||||
assert sub.target.target_name == "明日方舟"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_add_with_bilibili_live_target_parser(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.bilibili import Bilibililive
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
ak_list_router = respx.get(
|
||||
"https://api.bilibili.com/x/web-interface/card?mid=161775300"
|
||||
)
|
||||
ak_list_router.mock(
|
||||
return_value=Response(200, json=get_json("bilibili_arknights_profile.json"))
|
||||
)
|
||||
|
||||
bilibili_main_page_router = respx.get("https://www.bilibili.com/")
|
||||
bilibili_main_page_router.mock(return_value=Response(200))
|
||||
|
||||
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,
|
||||
BotReply.add_reply_on_platform(
|
||||
platform_manager=platform_manager, common_platform=common_platform
|
||||
),
|
||||
True,
|
||||
)
|
||||
event_2 = fake_group_message_event(
|
||||
message=Message("全部"), sender=Sender(card="", nickname="test", role="admin")
|
||||
)
|
||||
ctx.receive_event(bot, event_2)
|
||||
ctx.should_rejected()
|
||||
ctx.should_call_send(
|
||||
event_2,
|
||||
BotReply.add_reply_on_platform_input_allplatform(platform_manager),
|
||||
True,
|
||||
)
|
||||
event_3 = fake_group_message_event(
|
||||
message=Message("bilibili-live"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_3)
|
||||
ctx.should_call_send(
|
||||
event_3,
|
||||
BotReply.add_reply_on_id(Bilibililive),
|
||||
True,
|
||||
)
|
||||
|
||||
event_4_ok = fake_group_message_event(
|
||||
message=Message("161775300"),
|
||||
sender=fake_admin_user,
|
||||
)
|
||||
ctx.receive_event(bot, event_4_ok)
|
||||
ctx.should_call_send(
|
||||
event_4_ok,
|
||||
BotReply.add_reply_on_target_confirm("bilibili-live", "明日方舟", "161775300"),
|
||||
True,
|
||||
)
|
||||
ctx.should_call_send(
|
||||
event_4_ok,
|
||||
BotReply.add_reply_on_cats(platform_manager, "bilibili-live"),
|
||||
True,
|
||||
)
|
||||
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, BotReply.add_reply_subscribe_success("明日方舟"), True
|
||||
)
|
||||
ctx.should_finished()
|
||||
subs = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 1
|
||||
sub = subs[0]
|
||||
assert sub.target.target == "161775300"
|
||||
assert sub.tags == []
|
||||
assert sub.categories == [
|
||||
platform_manager["bilibili-live"].reverse_category["开播提醒"]
|
||||
]
|
||||
assert sub.target.platform_name == "bilibili-live"
|
||||
assert sub.target.target_name == "明日方舟"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_add_with_bilibili_bangumi_target_parser(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.event import Sender
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.platform.bilibili import BilibiliBangumi
|
||||
from nonebot_bison.sub_manager import add_sub_matcher, common_platform
|
||||
|
||||
ak_list_router = respx.get(
|
||||
"https://api.bilibili.com/pgc/review/user?media_id=28235413"
|
||||
)
|
||||
ak_list_router.mock(
|
||||
return_value=Response(200, json=get_json("bilibili-gangumi-hanhua1.json"))
|
||||
)
|
||||
|
||||
bilibili_main_page_router = respx.get("https://www.bilibili.com/")
|
||||
bilibili_main_page_router.mock(return_value=Response(200))
|
||||
|
||||
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,
|
||||
BotReply.add_reply_on_platform(
|
||||
platform_manager=platform_manager, common_platform=common_platform
|
||||
),
|
||||
True,
|
||||
)
|
||||
event_2 = fake_group_message_event(
|
||||
message=Message("全部"), sender=Sender(card="", nickname="test", role="admin")
|
||||
)
|
||||
ctx.receive_event(bot, event_2)
|
||||
ctx.should_rejected()
|
||||
ctx.should_call_send(
|
||||
event_2,
|
||||
BotReply.add_reply_on_platform_input_allplatform(platform_manager),
|
||||
True,
|
||||
)
|
||||
event_3 = fake_group_message_event(
|
||||
message=Message("bilibili-bangumi"), sender=fake_admin_user
|
||||
)
|
||||
ctx.receive_event(bot, event_3)
|
||||
ctx.should_call_send(
|
||||
event_3,
|
||||
BotReply.add_reply_on_id(BilibiliBangumi),
|
||||
True,
|
||||
)
|
||||
|
||||
event_4_ok = fake_group_message_event(
|
||||
message=Message("md28235413"),
|
||||
sender=fake_admin_user,
|
||||
)
|
||||
ctx.receive_event(bot, event_4_ok)
|
||||
ctx.should_call_send(
|
||||
event_4_ok,
|
||||
BotReply.add_reply_on_target_confirm(
|
||||
"bilibili-bangumi", "汉化日记 第三季", "28235413"
|
||||
),
|
||||
True,
|
||||
)
|
||||
ctx.should_call_send(
|
||||
event_4_ok, BotReply.add_reply_subscribe_success("汉化日记 第三季"), True
|
||||
)
|
||||
ctx.should_finished()
|
||||
subs = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 1
|
||||
sub = subs[0]
|
||||
assert sub.target.target == "28235413"
|
||||
assert sub.tags == []
|
||||
assert sub.target.platform_name == "bilibili-bangumi"
|
||||
assert sub.target.target_name == "汉化日记 第三季"
|
||||
@@ -0,0 +1,169 @@
|
||||
import pytest
|
||||
from nonebug import App
|
||||
|
||||
from ..utils import fake_group_message_event, fake_private_message_event, fake_superuser
|
||||
|
||||
|
||||
@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.sub_manager import group_manage_matcher
|
||||
|
||||
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,
|
||||
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_api(
|
||||
"get_group_list", {}, [{"group_id": 101, "group_name": "test group"}]
|
||||
)
|
||||
ctx.should_call_send(
|
||||
event,
|
||||
"请选择需要管理的群：\n1. 101 - test group\n请输入左侧序号\n中止操作请输入'取消'",
|
||||
True,
|
||||
)
|
||||
event_1_err = fake_private_message_event(
|
||||
message=Message("0"),
|
||||
sender=fake_superuser,
|
||||
to_me=True,
|
||||
user_id=fake_superuser.user_id,
|
||||
)
|
||||
ctx.receive_event(bot, event_1_err)
|
||||
ctx.should_call_send(event_1_err, "请输入正确序号", True)
|
||||
ctx.should_rejected()
|
||||
event_1_ok = fake_private_message_event(
|
||||
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,
|
||||
to_me=True,
|
||||
user_id=fake_superuser.user_id,
|
||||
)
|
||||
ctx.receive_event(bot, event_2_err)
|
||||
ctx.should_call_send(event_2_err, "请输入正确的命令", True)
|
||||
ctx.should_rejected()
|
||||
event_2_ok = fake_private_message_event(
|
||||
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_abort_on_idx(app: App):
|
||||
from nonebot.adapters.onebot.v11.bot import Bot
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
|
||||
from nonebot_bison.sub_manager import group_manage_matcher
|
||||
|
||||
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,
|
||||
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_api(
|
||||
"get_group_list", {}, [{"group_id": 101, "group_name": "test group"}]
|
||||
)
|
||||
ctx.should_call_send(
|
||||
event,
|
||||
"请选择需要管理的群：\n1. 101 - test group\n请输入左侧序号\n中止操作请输入'取消'",
|
||||
True,
|
||||
)
|
||||
event_abort = fake_private_message_event(
|
||||
message=Message("取消"),
|
||||
sender=fake_superuser,
|
||||
to_me=True,
|
||||
user_id=fake_superuser.user_id,
|
||||
)
|
||||
ctx.receive_event(bot, event_abort)
|
||||
ctx.should_call_send(event_abort, "已取消", True)
|
||||
ctx.should_finished()
|
||||
|
||||
|
||||
async def test_query_with_abort_on_command(app: App):
|
||||
from nonebot.adapters.onebot.v11.bot import Bot
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
|
||||
from nonebot_bison.sub_manager import group_manage_matcher
|
||||
|
||||
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,
|
||||
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_api(
|
||||
"get_group_list", {}, [{"group_id": 101, "group_name": "test group"}]
|
||||
)
|
||||
ctx.should_call_send(
|
||||
event,
|
||||
"请选择需要管理的群：\n1. 101 - test group\n请输入左侧序号\n中止操作请输入'取消'",
|
||||
True,
|
||||
)
|
||||
event_1_ok = fake_private_message_event(
|
||||
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_abort = fake_private_message_event(
|
||||
message=Message("取消"),
|
||||
sender=fake_superuser,
|
||||
to_me=True,
|
||||
user_id=fake_superuser.user_id,
|
||||
)
|
||||
ctx.receive_event(bot, event_abort)
|
||||
ctx.should_call_send(event_abort, "已取消", True)
|
||||
ctx.should_finished()
|
||||
|
||||
|
||||
@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.sub_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)
|
||||
@@ -0,0 +1,118 @@
|
||||
import pytest
|
||||
from nonebug.app import App
|
||||
from nonebug_saa import should_send_saa
|
||||
|
||||
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, init_scheduler):
|
||||
from nonebot import get_driver
|
||||
from nonebot.adapters.onebot.v11 import Bot, Message
|
||||
from nonebot_plugin_saa import MessageFactory, SupportedAdapters, TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.sub_manager import query_sub_matcher
|
||||
from nonebot_bison.types import Target
|
||||
|
||||
await config.add_subscribe(
|
||||
TargetQQGroup(group_id=10000),
|
||||
Target("6279793937"),
|
||||
"明日方舟Arknights",
|
||||
"weibo",
|
||||
[platform_manager["weibo"].reverse_category["图文"]],
|
||||
["明日方舟"],
|
||||
)
|
||||
async with app.test_matcher(query_sub_matcher) as ctx:
|
||||
bot = ctx.create_bot(base=Bot)
|
||||
|
||||
event = fake_group_message_event(message=Message("查询订阅"), to_me=True)
|
||||
ctx.receive_event(bot, event)
|
||||
ctx.should_pass_rule()
|
||||
ctx.should_pass_permission()
|
||||
should_send_saa(
|
||||
ctx,
|
||||
MessageFactory("订阅的帐号为：\nweibo 明日方舟Arknights 6279793937 [图文] 明日方舟\n"),
|
||||
bot,
|
||||
event=event,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_del_sub(app: App, init_scheduler):
|
||||
from nonebot.adapters.onebot.v11.bot import Bot
|
||||
from nonebot.adapters.onebot.v11.message import Message
|
||||
from nonebot_plugin_saa import MessageFactory, TargetQQGroup
|
||||
|
||||
from nonebot_bison.config import config
|
||||
from nonebot_bison.platform import platform_manager
|
||||
from nonebot_bison.sub_manager import del_sub_matcher
|
||||
from nonebot_bison.types import Target
|
||||
|
||||
await config.add_subscribe(
|
||||
TargetQQGroup(group_id=10000),
|
||||
Target("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()
|
||||
should_send_saa(
|
||||
ctx,
|
||||
MessageFactory(
|
||||
"订阅的帐号为：\n1 weibo 明日方舟Arknights 6279793937\n [图文] 明日方舟\n请输入要删除的订阅的序号\n输入'取消'中止"
|
||||
),
|
||||
bot,
|
||||
event=event,
|
||||
)
|
||||
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 = await config.list_subscribe(TargetQQGroup(group_id=10000))
|
||||
assert len(subs) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_del_empty_sub(app: App, init_scheduler):
|
||||
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.platform import platform_manager
|
||||
from nonebot_bison.sub_manager import del_sub_matcher
|
||||
|
||||
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_finished()
|
||||
ctx.should_call_send(
|
||||
event,
|
||||
"暂无已订阅账号\n请使用“添加订阅”命令添加订阅",
|
||||
True,
|
||||
)
|
||||
Reference in New Issue
Block a user