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

✨ 添加 Cookie 组件 (#633)

Browse Source 
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
suyiiyii
2024-10-31 12:56:15 +08:00
committed by GitHub
parent 3bdc79162e
commit 97a0f04808
63 changed files with 6119 additions and 806 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/config/__init__.py
View File
tests/config/test_cookie.py
+124
View File
@@ -0,0 +1,124 @@
import json
from typing import cast
from datetime import datetime
import pytest
from nonebug import App
@pytest.mark.usefixtures("_patch_weibo_get_cookie_name")
async def test_cookie(app: App, init_scheduler):
from nonebot_plugin_saa import TargetQQGroup
from nonebot_bison.config.db_config import config
from nonebot_bison.scheduler import scheduler_dict
from nonebot_bison.types import Target as T_Target
from nonebot_bison.config.utils import DuplicateCookieTargetException
from nonebot_bison.utils.site import CookieClientManager, site_manager
target = T_Target("weibo_id")
platform_name = "weibo"
await config.add_subscribe(
TargetQQGroup(group_id=123),
target=target,
target_name="weibo_name",
platform_name=platform_name,
cats=[],
tags=[],
)
site = site_manager["weibo.com"]
client_mgr = cast(CookieClientManager, scheduler_dict[site].client_mgr)
# 刷新匿名cookie
await client_mgr.refresh_client()
cookies = await config.get_cookie(site_name=site.name)
assert len(cookies) == 1
# 添加用户cookie
await client_mgr.add_user_cookie(json.dumps({"test_cookie": "1"}))
await client_mgr.add_user_cookie(json.dumps({"test_cookie": "2"}))
cookies = await config.get_cookie(site_name=site.name)
assert len(cookies) == 3
cookies = await config.get_cookie(site_name=site.name, is_anonymous=False)
assert len(cookies) == 2
# 单个target，多个cookie
await config.add_cookie_target(target, platform_name, cookies[0].id)
await config.add_cookie_target(target, platform_name, cookies[1].id)
cookies = await config.get_cookie(site_name=site.name, target=target)
assert len(cookies) == 3
cookies = await config.get_cookie(site_name=site.name, target=target, is_anonymous=False)
assert len(cookies) == 2
cookies = await config.get_cookie(site_name=site.name, target=target, is_universal=False)
assert len(cookies) == 2
# 测试不同的target
target2 = T_Target("weibo_id2")
await config.add_subscribe(
TargetQQGroup(group_id=123),
target=target2,
target_name="weibo_name2",
platform_name=platform_name,
cats=[],
tags=[],
)
await client_mgr.add_user_cookie(json.dumps({"test_cookie": "3"}))
cookies = await config.get_cookie(site_name=site.name, is_anonymous=False)
# 多个target，多个cookie
await config.add_cookie_target(target2, platform_name, cookies[0].id)
await config.add_cookie_target(target2, platform_name, cookies[2].id)
cookies = await config.get_cookie(site_name=site.name, target=target2)
assert len(cookies) == 3
# 重复关联 target
with pytest.raises(DuplicateCookieTargetException) as e:
await config.add_cookie_target(target2, platform_name, cookies[2].id)
assert isinstance(e.value, DuplicateCookieTargetException)
cookies = await config.get_cookie(site_name=site.name, target=target2, is_anonymous=False)
assert len(cookies) == 2
# 有关联的cookie不能删除
with pytest.raises(Exception, match="cookie") as e:
await config.delete_cookie_by_id(cookies[1].id)
cookies = await config.get_cookie(site_name=site.name, target=target2, is_anonymous=False)
assert len(cookies) == 2
await config.delete_cookie_target(target2, platform_name, cookies[1].id)
await config.delete_cookie_by_id(cookies[1].id)
cookies = await config.get_cookie(site_name=site.name, target=target2, is_anonymous=False)
assert len(cookies) == 1
cookie = cookies[0]
cookie_id = cookie.id
cookie.last_usage = datetime(2024, 9, 13)
cookie.status = "test"
await config.update_cookie(cookie)
cookies = await config.get_cookie(site_name=site.name, target=target2, is_anonymous=False)
assert len(cookies) == 1
assert cookies[0].id == cookie_id
assert cookies[0].last_usage == datetime(2024, 9, 13)
assert cookies[0].status == "test"
# 不存在的 cookie_id
cookie.id = 114514
with pytest.raises(ValueError, match="cookie") as e:
await config.update_cookie(cookie)
# 获取所有关联对象
cookie_targets = await config.get_cookie_target()
assert len(cookie_targets) == 3
# 删除关联对象
await config.delete_cookie_target_by_id(cookie_targets[0].id)
cookie_targets = await config.get_cookie_target()
assert len(cookie_targets) == 2
tests/conftest.py
+33
View File
@@ -29,6 +29,16 @@ def load_adapters(nonebug_init: None):
return driver
def patch_refresh_bilibili_anonymous_cookie(mocker: MockerFixture):
# patch 掉bilibili的匿名cookie生成函数，避免真实请求
from nonebot_bison.platform.bilibili.scheduler import BilibiliClientManager
mocker.patch.object(
BilibiliClientManager, "_get_cookies", return_value=[{"name": "test anonymous", "content": "test"}]
)
@pytest.fixture
async def app(tmp_path: Path, request: pytest.FixtureRequest, mocker: MockerFixture):
sys.path.append(str(Path(__file__).parent.parent / "src" / "plugins"))
@@ -51,6 +61,10 @@ async def app(tmp_path: Path, request: pytest.FixtureRequest, mocker: MockerFixt
param: AppReq = getattr(request, "param", AppReq())
# 如果在 app 前调用会报错“无法找到调用者”
# 而在后面调用又来不及mock，所以只能在中间mock
patch_refresh_bilibili_anonymous_cookie(mocker)
if not param.get("no_init_db"):
await init_db()
# if not param.get("refresh_bot"):
@@ -123,3 +137,22 @@ async def _no_browser(app: App, mocker: MockerFixture):
mocker.patch.object(plugin_config, "bison_use_browser", False)
mocker.patch("nonebot_bison.platform.unavailable_paltforms", _get_unavailable_platforms())
@pytest.fixture
async def _clear_db(app: App):
from nonebot_bison.config import config
await config.clear_db()
yield
await config.clear_db()
return
@pytest.fixture
def _patch_weibo_get_cookie_name(app: App, mocker: MockerFixture):
from nonebot_bison.platform import weibo
mocker.patch.object(weibo.WeiboClientManager, "_get_current_user_name", return_value="test_name")
yield
mocker.stopall()
tests/platforms/static/weibo_get-cookie-name.json
Vendored
+138
View File
@@ -0,0 +1,138 @@
{
"ok": 1,
"data": {
"config": {
"code": 11000,
"text": "<p>非微博会员不可多次修改昵称。自2024年1月1日至今，您已成功修改1次，目前无法继续修改。如需继续改名可开通微博会员。</p>",
"guide": {
"title": "开通微博会员",
"desc": "<p>本年度<span>增加最多8次</span>改名机会</p>",
"button_text": "开通会员",
"button_url": "https://m.weibo.cn/c/upgrade"
}
},
"data": {
"config": {
"title": "修改昵称次数扣除明细"
}
},
"user": {
"id": 114514,
"idstr": "114514",
"class": 1,
"screen_name": "suyiiyii",
"name": "suyiiyii",
"province": "44",
"city": "1",
"location": "广东 广州",
"description": "",
"url": "",
"profile_image_url": "https://tvax1.sinaimg.cn/default/images/default_avatar_male_50.gif?KID=imgbed,tva&Expires=1728833531&ssig=jURhYal3%2BR",
"light_ring": false,
"profile_url": "u/114514",
"domain": "",
"weihao": "",
"gender": "m",
"followers_count": 1,
"followers_count_str": "1",
"friends_count": 6,
"pagefriends_count": 0,
"statuses_count": 1,
"video_status_count": 0,
"video_play_count": 0,
"super_topic_not_syn_count": 0,
"favourites_count": 0,
"created_at": "Tue Sep 03 00:07:59 +0800 2024",
"following": false,
"allow_all_act_msg": false,
"geo_enabled": true,
"verified": false,
"verified_type": -1,
"remark": "",
"insecurity": {
"sexual_content": false
},
"ptype": 0,
"allow_all_comment": true,
"avatar_large": "https://tvax1.sinaimg.cn/default/images/default_avatar_male_180.gif?KID=imgbed,tva&Expires=1728833531&ssig=cornnikInk",
"avatar_hd": "https://tvax1.sinaimg.cn/default/images/default_avatar_male_180.gif?KID=imgbed,tva&Expires=1728833531&ssig=cornnikInk",
"verified_reason": "",
"verified_trade": "",
"verified_reason_url": "",
"verified_source": "",
"verified_source_url": "",
"follow_me": false,
"like": false,
"like_me": false,
"online_status": 0,
"bi_followers_count": 0,
"lang": "zh-cn",
"star": 0,
"mbtype": 0,
"mbrank": 0,
"svip": 0,
"vvip": 0,
"mb_expire_time": 0,
"block_word": 0,
"block_app": 0,
"chaohua_ability": 0,
"brand_ability": 0,
"nft_ability": 0,
"vplus_ability": 0,
"wenda_ability": 0,
"live_ability": 0,
"gongyi_ability": 0,
"paycolumn_ability": 0,
"newbrand_ability": 0,
"ecommerce_ability": 0,
"hardfan_ability": 0,
"wbcolumn_ability": 0,
"interaction_user": 0,
"audio_ability": 0,
"place_ability": 0,
"credit_score": 80,
"user_ability": 0,
"urank": 0,
"story_read_state": -1,
"vclub_member": 0,
"is_teenager": 0,
"is_guardian": 0,
"is_teenager_list": 0,
"pc_new": 0,
"special_follow": false,
"planet_video": 0,
"video_mark": 0,
"live_status": 0,
"user_ability_extend": 0,
"status_total_counter": {
"total_cnt": 0,
"repost_cnt": 0,
"comment_cnt": 0,
"like_cnt": 0,
"comment_like_cnt": 0
},
"video_total_counter": {
"play_cnt": -1
},
"brand_account": 0,
"hongbaofei": 0,
"green_mode": 0,
"urisk": 524288,
"unfollowing_recom_switch": 1,
"block": 0,
"block_me": 0,
"avatar_type": 0,
"is_big": 0,
"auth_status": 1,
"auth_realname": null,
"auth_career": null,
"auth_career_name": null,
"show_auth": 0,
"is_auth": 0,
"is_punish": 0,
"like_display": 0
},
"submit": true,
"having_count": 0
}
}
tests/platforms/test_weibo.py
+11
View File
@@ -217,3 +217,14 @@ async def test_parse_target(weibo: "Weibo"):
assert res == "6441489862"
with pytest.raises(Platform.ParseTargetException):
await weibo.parse_target("https://weibo.com/arknights")
@respx.mock
async def test_get_cookie_name(weibo: "Weibo"):
from nonebot_bison.platform.weibo import WeiboClientManager
router = respx.get("https://m.weibo.cn/setup/nick/detail")
router.mock(return_value=Response(200, json=get_json("weibo_get-cookie-name.json")))
weibo_client_mgr = WeiboClientManager()
name = await weibo_client_mgr.get_cookie_name("{}")
assert name == "weibo: [suyiiyii]"
tests/scheduler/test_site.py
+4
View File
@@ -5,6 +5,8 @@ from unittest.mock import AsyncMock
from nonebug import App
from pytest_mock import MockerFixture
from tests.conftest import patch_refresh_bilibili_anonymous_cookie
if typing.TYPE_CHECKING:
from nonebot_bison.utils import Site
@@ -199,6 +201,7 @@ async def test_scheduler_skip_browser(mocker: MockerFixture):
site = MockSite
mocker.patch.dict(platform_manager, {"mock_platform": MockPlatform})
patch_refresh_bilibili_anonymous_cookie(mocker)
await init_scheduler()
@@ -229,6 +232,7 @@ async def test_scheduler_no_skip_not_require_browser(mocker: MockerFixture):
site = MockSite
mocker.patch.dict(platform_manager, {"mock_platform": MockPlatform})
patch_refresh_bilibili_anonymous_cookie(mocker)
await init_scheduler()
tests/sub_manager/test_add_cookie.py
+219
View File
@@ -0,0 +1,219 @@
import json
import pytest
from nonebug.app import App
from pytest_mock import MockerFixture
from ..utils import BotReply, fake_superuser, fake_admin_user, fake_private_message_event
async def test_add_cookie_rule(app: App, mocker: MockerFixture):
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_cookie_matcher
mocker.patch.object(plugin_config, "bison_to_me", True)
async with app.test_matcher(add_cookie_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event = fake_private_message_event(message=Message("添加cookie"), sender=fake_superuser)
ctx.receive_event(bot, event)
ctx.should_pass_rule()
ctx.should_pass_permission()
async with app.test_matcher(add_cookie_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event = fake_private_message_event(message=Message("添加cookie"), sender=fake_admin_user)
ctx.receive_event(bot, event)
ctx.should_not_pass_rule()
ctx.should_pass_permission()
@pytest.mark.usefixtures("_clear_db")
async def test_add_cookie_target_no_cookie(app: App):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.sub_manager import add_cookie_target_matcher
async with app.test_matcher(add_cookie_target_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
from nonebug_saa import should_send_saa
from nonebot_plugin_saa import TargetQQGroup, MessageFactory
from nonebot_bison.config import config
from nonebot_bison.types import Target as T_Target
target = T_Target("weibo_id")
platform_name = "weibo"
await config.add_subscribe(
TargetQQGroup(group_id=123),
target=target,
target_name="weibo_name",
platform_name=platform_name,
cats=[],
tags=[],
)
event_1 = fake_private_message_event(
message=Message("关联cookie"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
should_send_saa(
ctx,
MessageFactory(
"订阅的帐号为：\n1 weibo weibo_name weibo_id\n []\n请输入要关联 cookie 的订阅的序号\n输入'取消'中止"
),
bot,
event=event_1,
)
event_2 = fake_private_message_event(
message=Message("1"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_2)
ctx.should_pass_rule()
ctx.should_call_send(
event_2,
"当前平台暂无可关联的 Cookie，请使用“添加cookie”命令添加或检查已关联的 Cookie",
True,
)
@pytest.mark.usefixtures("_clear_db")
@pytest.mark.usefixtures("_patch_weibo_get_cookie_name")
async def test_add_cookie(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.sub_manager import common_platform, add_cookie_matcher, add_cookie_target_matcher
async with app.test_matcher(add_cookie_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event_1 = fake_private_message_event(
message=Message("添加cookie"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
BotReply.add_reply_on_add_cookie(platform_manager, common_platform),
True,
)
event_2 = fake_private_message_event(
message=Message("全部"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_2)
ctx.should_pass_rule()
ctx.should_rejected()
ctx.should_call_send(
event_2,
BotReply.add_reply_on_add_cookie_input_allplatform(platform_manager),
True,
)
event_3 = fake_private_message_event(
message=Message("weibo"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_3)
ctx.should_pass_rule()
ctx.should_call_send(event_3, BotReply.add_reply_on_input_cookie)
event_4_err = fake_private_message_event(
message=Message("test"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_4_err)
ctx.should_call_send(
event_4_err,
"无效的 Cookie，请检查后重新输入，详情见https://nonebot-bison.netlify.app/usage/cookie.html",
True,
)
ctx.should_rejected()
event_4_ok = fake_private_message_event(
message=Message(json.dumps({"cookie": "test"})),
sender=fake_superuser,
to_me=True,
user_id=fake_superuser.user_id,
)
ctx.receive_event(bot, event_4_ok)
ctx.should_pass_rule()
ctx.should_call_send(
event_4_ok, "已添加 Cookie: weibo: [test_name] 到平台 weibo\n请使用“关联cookie”为 Cookie 关联订阅", True
)
async with app.test_matcher(add_cookie_target_matcher) as ctx:
from nonebug_saa import should_send_saa
from nonebot_plugin_saa import TargetQQGroup, MessageFactory
from nonebot_bison.config import config
from nonebot_bison.types import Target as T_Target
target = T_Target("weibo_id")
platform_name = "weibo"
await config.add_subscribe(
TargetQQGroup(group_id=123),
target=target,
target_name="weibo_name",
platform_name=platform_name,
cats=[],
tags=[],
)
bot = ctx.create_bot(base=Bot)
event_1 = fake_private_message_event(
message=Message("关联cookie"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
should_send_saa(
ctx,
MessageFactory(
"订阅的帐号为：\n1 weibo weibo_name weibo_id\n []\n请输入要关联 cookie 的订阅的序号\n输入'取消'中止"
),
bot,
event=event_1,
)
event_2_err = fake_private_message_event(
message=Message("2"), 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("1"), 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_call_send(event_2_ok, "请选择一个 Cookie，已关联的 Cookie 不会显示\n1. weibo: [test_name]", True)
event_3_err = fake_private_message_event(
message=Message("2"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_3_err)
ctx.should_call_send(event_3_err, "序号错误", True)
ctx.should_rejected()
event_3_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_3_ok)
ctx.should_pass_rule()
ctx.should_call_send(event_3_ok, "已关联 Cookie: weibo: [test_name] 到订阅 weibo.com weibo_id", True)
async def test_add_cookie_target_no_target(app: App, mocker: MockerFixture):
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_bison.sub_manager import add_cookie_target_matcher
async with app.test_matcher(add_cookie_target_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event_1 = fake_private_message_event(
message=Message("关联cookie"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_call_send(
event_1,
"暂无已订阅账号\n请使用“添加订阅”命令添加订阅",
True,
)
tests/sub_manager/test_delete_cookie.py
+136
View File
@@ -0,0 +1,136 @@
import json
import pytest
from nonebug.app import App
from ..utils import fake_superuser, fake_private_message_event
@pytest.mark.usefixtures("_clear_db")
async def test_del_cookie(app: App):
from nonebug_saa import should_send_saa
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_plugin_saa import TargetQQGroup, MessageFactory
from nonebot_bison.config import config
from nonebot_bison.config.db_model import Cookie
from nonebot_bison.types import Target as T_Target
from nonebot_bison.sub_manager import del_cookie_matcher
async with app.test_matcher(del_cookie_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event = fake_private_message_event(
message=Message("删除cookie"), 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, "暂无已添加的 Cookie\n请使用“添加cookie”命令添加", True)
async with app.test_matcher(del_cookie_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
target = T_Target("weibo_id")
platform_name = "weibo"
await config.add_subscribe(
TargetQQGroup(group_id=123),
target=target,
target_name="weibo_name",
platform_name=platform_name,
cats=[],
tags=[],
)
await config.add_cookie(Cookie(content=json.dumps({"cookie": "test"}), site_name="weibo.com"))
event_1 = fake_private_message_event(
message=Message("删除cookie"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_pass_permission()
should_send_saa(
ctx,
MessageFactory(
"已添加的 Cookie 为：\n1 weibo.com unnamed cookie"
" 0个关联\n请输入要删除的 Cookie 的序号\n输入'取消'中止"
),
bot,
event=event_1,
)
event_2 = fake_private_message_event(
message=Message("1"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_2)
ctx.should_pass_rule()
ctx.should_pass_permission()
ctx.should_call_send(event_2, "删除成功", True)
@pytest.mark.usefixtures("_clear_db")
@pytest.mark.usefixtures("_patch_weibo_get_cookie_name")
async def test_del_cookie_err(app: App):
from nonebug_saa import should_send_saa
from nonebot.adapters.onebot.v11.bot import Bot
from nonebot.adapters.onebot.v11.message import Message
from nonebot_plugin_saa import TargetQQGroup, MessageFactory
from nonebot_bison.config import config
from nonebot_bison.config.db_model import Cookie
from nonebot_bison.types import Target as T_Target
from nonebot_bison.sub_manager import del_cookie_matcher
async with app.test_matcher(del_cookie_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
event = fake_private_message_event(
message=Message("删除cookie"), 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, "暂无已添加的 Cookie\n请使用“添加cookie”命令添加", True)
async with app.test_matcher(del_cookie_matcher) as ctx:
bot = ctx.create_bot(base=Bot)
target = T_Target("weibo_id")
platform_name = "weibo"
await config.add_subscribe(
TargetQQGroup(group_id=123),
target=target,
target_name="weibo_name",
platform_name=platform_name,
cats=[],
tags=[],
)
await config.add_cookie(Cookie(content=json.dumps({"cookie": "test"}), site_name="weibo.com"))
cookies = await config.get_cookie(is_anonymous=False)
await config.add_cookie_target(target, platform_name, cookies[0].id)
event_1 = fake_private_message_event(
message=Message("删除cookie"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_1)
ctx.should_pass_rule()
ctx.should_pass_permission()
should_send_saa(
ctx,
MessageFactory(
"已添加的 Cookie 为：\n1 weibo.com unnamed cookie 1个关联\n请输入要删除的 Cookie 的序号\n输入'取消'中止"
),
bot,
event=event_1,
)
event_2_err = fake_private_message_event(
message=Message("2"), 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 = fake_private_message_event(
message=Message("1"), sender=fake_superuser, to_me=True, user_id=fake_superuser.user_id
)
ctx.receive_event(bot, event_2)
ctx.should_pass_rule()
ctx.should_call_send(event_2, "只能删除未关联的 Cookie，请使用“取消关联cookie”命令取消关联", True)
ctx.should_call_send(event_2, "删除错误", True)
ctx.should_rejected()
tests/subs_io/static/v3/subs_export.json
+69
View File
@@ -0,0 +1,69 @@
{
"version": 3,
"groups": [
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 1232
},
"subs": [
{
"categories": [],
"tags": [],
"target": {
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
}
]
},
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 2342
},
"subs": [
{
"categories": [],
"tags": ["kaltsit", "amiya"],
"target": {
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
},
{
"categories": [1, 2],
"tags": [],
"target": {
"target_name": "bilibili_name",
"target": "bilibili_id",
"platform_name": "bilibili",
"default_schedule_weight": 10
}
}
]
}
],
"cookies": [
{
"site_name": "weibo.com",
"content": "{\"cookie\": \"test\"}",
"cookie_name": "test cookie",
"cd_milliseconds": 0,
"is_universal": false,
"tags": {},
"targets": [
{
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
]
}
]
}
tests/subs_io/static/v3/subs_export.yaml
+48
View File
@@ -0,0 +1,48 @@
version: 3
groups:
- subs:
- categories: []
tags: []
target:
default_schedule_weight: 10
platform_name: weibo
target: weibo_id
target_name: weibo_name
user_target:
platform_type: QQ Group
group_id: 123552
- subs:
- categories: []
tags:
- kaltsit
- amiya
target:
default_schedule_weight: 10
platform_name: weibo
target: weibo_id
target_name: weibo_name
- categories:
- 1
- 2
tags: []
target:
default_schedule_weight: 10
platform_name: bilibili
target: bilibili_id
target_name: bilibili_name
user_target:
platform_type: QQ Group
group_id: 234662
cookies:
- site_name: weibo.com
content: '{"cookie": "test"}'
cookie_name: test cookie
cd_milliseconds: 0
is_universal: false
tags: {}
targets:
- target_name: weibo_name
target: weibo_id
platform_name: weibo
default_schedule_weight: 10
tests/subs_io/static/v3/subs_export_all_illegal.json
+103
View File
@@ -0,0 +1,103 @@
{
"version": 2,
"groups": [
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 123
},
"subs": [
{
"categories": [],
"tags": [],
"target": {
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
}
]
},
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 234
},
"subs": [
{
"tags": ["kaltsit", "amiya"],
"target": {
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
},
{
"categories": [1, 2],
"tags": [],
"target": [
{
"target_name": "bilibili_name",
"target": "bilibili_id",
"platform_name": "bilibili",
"default_schedule_weight": 10
}
]
}
]
},
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 123
},
"subs": {
"categories": [],
"tags": [],
"target": {
"target_name": "weibo_name2",
"target": "weibo_id2",
"platform_name": "weibo",
"default_schedule_weight": 10
}
}
},
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 123
},
"subs": [
{
"categories": [],
"tags": [],
"target": {
"target_name": "weibo_name2",
"platform_name": "weibo",
"default_schedule_weight": 10
}
}
]
}
],
"cookies": [
{
"site_name": "weibo.com1111",
"content": "{\"cookie\": 111}",
"cookie_name": "test cookie1",
"cd_milliseconds": -1,
"is_universal": false,
"tags": {},
"targets": [
{
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
]
}
]
}
tests/subs_io/static/v3/subs_export_has_subdup_err.json
+97
View File
@@ -0,0 +1,97 @@
{
"version": 3,
"groups": [
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 1232
},
"subs": [
{
"categories": [],
"tags": [],
"target": {
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
}
]
},
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 2342
},
"subs": [
{
"categories": [],
"tags": ["kaltsit", "amiya"],
"target": {
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
},
{
"categories": [1, 2],
"tags": [],
"target": {
"target_name": "bilibili_name",
"target": "bilibili_id",
"platform_name": "bilibili",
"default_schedule_weight": 10
}
}
]
},
{
"user_target": {
"platform_type": "QQ Group",
"group_id": 1232
},
"subs": [
{
"categories": [],
"tags": [],
"target": {
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
},
{
"categories": [2, 6],
"tags": ["poca"],
"target": {
"target_name": "weibo_name2",
"target": "weibo_id2",
"platform_name": "weibo",
"default_schedule_weight": 10
}
}
]
}
],
"cookies": [
{
"site_name": "weibo.com",
"content": "{\"cookie\": \"test\"}",
"cookie_name": "test cookie",
"cd_milliseconds": 0,
"is_universal": false,
"tags": {},
"targets": [
{
"target_name": "weibo_name",
"target": "weibo_id",
"platform_name": "weibo",
"default_schedule_weight": 10
}
]
}
]
}
tests/subs_io/test_cli.py
+15 -3
View File
@@ -40,6 +40,7 @@ def test_cli_help(app: App):
async def test_subs_export(app: App, tmp_path: Path):
from nonebot_plugin_saa import TargetQQGroup
from nonebot_bison.config.db_model import Cookie
from nonebot_bison.config.db_config import config
from nonebot_bison.types import Target as TTarget
from nonebot_bison.script.cli import cli, run_sync
@@ -70,6 +71,14 @@ async def test_subs_export(app: App, tmp_path: Path):
cats=[1, 2],
tags=[],
)
cookie_id = await config.add_cookie(
Cookie(
site_name="weibo.com",
content='{"cookie": "test"}',
cookie_name="test cookie",
)
)
await config.add_cookie_target("weibo_id", "weibo", cookie_id)
assert len(await config.list_subs_with_all_info()) == 3
@@ -84,8 +93,9 @@ async def test_subs_export(app: App, tmp_path: Path):
assert result.exit_code == 0
file_path = Path.cwd() / "bison_subscribes_export_1.json"
assert file_path.exists()
assert '"version": 2' in file_path.read_text()
assert '"version": 3' in file_path.read_text()
assert '"group_id": 123' in file_path.read_text()
assert '"content": "{\\"cookie\\": \\"test\\"}",\n' in file_path.read_text()
# 是否导出到指定已存在文件夹
data_dir = tmp_path / "data"
@@ -94,8 +104,9 @@ async def test_subs_export(app: App, tmp_path: Path):
assert result.exit_code == 0
file_path2 = data_dir / "bison_subscribes_export_1.json"
assert file_path2.exists()
assert '"version": 2' in file_path2.read_text()
assert '"version": 3' in file_path2.read_text()
assert '"group_id": 123' in file_path2.read_text()
assert '"content": "{\\"cookie\\": \\"test\\"}",\n' in file_path.read_text()
# 是否拒绝导出到不存在的文件夹
result = await run_sync(runner.invoke)(cli, ["export", "-p", str(tmp_path / "data2")])
@@ -106,9 +117,10 @@ async def test_subs_export(app: App, tmp_path: Path):
assert result.exit_code == 0
file_path3 = tmp_path / "bison_subscribes_export_1.yaml"
assert file_path3.exists()
assert "version: 2" in file_path3.read_text()
assert "version: 3" in file_path3.read_text()
assert "group_id: 123" in file_path3.read_text()
assert "platform_type: QQ Group" in file_path3.read_text()
assert '"content": "{\\"cookie\\": \\"test\\"}",\n' in file_path.read_text()
# 是否允许以未支持的格式导出
result = await run_sync(runner.invoke)(cli, ["export", "-p", str(tmp_path), "--format", "toml"])
tests/subs_io/test_subs_io.py
+34 -8
View File
@@ -5,12 +5,13 @@ from nonebot.compat import model_dump
from .utils import get_json
@pytest.mark.usefixtures("_clear_db")
async def test_subs_export(app: App, init_scheduler):
from nonebot_plugin_saa import TargetQQGroup
from nonebot_bison.config.db_model import User
from nonebot_bison.config.db_config import config
from nonebot_bison.types import Target as TTarget
from nonebot_bison.config.db_model import User, Cookie
from nonebot_bison.config.subs_io import subscribes_export
await config.add_subscribe(
@@ -37,12 +38,20 @@ async def test_subs_export(app: App, init_scheduler):
cats=[1, 2],
tags=[],
)
cookie_id = await config.add_cookie(
Cookie(
site_name="weibo.com",
content='{"cookie": "test"}',
cookie_name="test cookie",
)
)
await config.add_cookie_target("weibo_id", "weibo", cookie_id)
data = await config.list_subs_with_all_info()
assert len(data) == 3
nbesf_data = await subscribes_export(lambda x: x)
assert model_dump(nbesf_data) == get_json("v2/subs_export.json")
assert model_dump(nbesf_data) == get_json("v3/subs_export.json")
nbesf_data_user_234 = await subscribes_export(
lambda stmt: stmt.where(User.user_target == {"platform_type": "QQ Group", "group_id": 2342})
@@ -102,16 +111,30 @@ async def test_subs_import_dup_err(app: App, init_scheduler):
async def test_subs_import_version_disorder(app: App, init_scheduler):
from nonebot_bison.config.subs_io import subscribes_import
from nonebot_bison.config.subs_io.nbesf_model import v1, v2
from nonebot_bison.config.subs_io.utils import NBESFParseErr
# use v2 parse v1
with pytest.raises(NBESFParseErr):
v1.nbesf_parser(get_json("v2/subs_export_has_subdup_err.json"))
from nonebot_bison.config.subs_io.nbesf_model import v1, v2, v3
# use v1 parse v2
with pytest.raises(NBESFParseErr):
v1.nbesf_parser(get_json("v2/subs_export_has_subdup_err.json"))
# use v1 parse v3
with pytest.raises(NBESFParseErr):
v1.nbesf_parser(get_json("v3/subs_export_has_subdup_err.json"))
# use v2 parse v1
with pytest.raises(NBESFParseErr):
v2.nbesf_parser(get_json("v1/subs_export_has_subdup_err.json"))
# # use v2 parse v3
# with pytest.raises(NBESFParseErr):
# v2.nbesf_parser(get_json("v3/subs_export_has_subdup_err.json"))
# use v3 parse v1
with pytest.raises(NBESFParseErr):
v3.nbesf_parser(get_json("v1/subs_export_has_subdup_err.json"))
# # use v3 parse v2
# with pytest.raises(NBESFParseErr):
# v3.nbesf_parser(get_json("v2/subs_export_has_subdup_err.json"))
# TODO: v3 parse v2 不会报错，但是v3 parse v1 会报错，似乎是有问题 (
with pytest.raises(AssertionError): # noqa: PT012
nbesf_data = v2.nbesf_parser(get_json("v2/subs_export_has_subdup_err.json"))
@@ -121,7 +144,7 @@ async def test_subs_import_version_disorder(app: App, init_scheduler):
async def test_subs_import_all_fail(app: App, init_scheduler):
"""只要文件格式有任何一个错误， 都不会进行订阅"""
from nonebot_bison.config.subs_io.nbesf_model import v1, v2
from nonebot_bison.config.subs_io.nbesf_model import v1, v2, v3
from nonebot_bison.config.subs_io.nbesf_model.v1 import NBESFParseErr
with pytest.raises(NBESFParseErr):
@@ -129,3 +152,6 @@ async def test_subs_import_all_fail(app: App, init_scheduler):
with pytest.raises(NBESFParseErr):
v2.nbesf_parser(get_json("v2/subs_export_all_illegal.json"))
with pytest.raises(NBESFParseErr):
v3.nbesf_parser(get_json("v3/subs_export_all_illegal.json"))
tests/utils.py
+31
View File
@@ -89,6 +89,7 @@ add_reply_on_id_input_search = (
class BotReply:
@staticmethod
def add_reply_on_platform(platform_manager, common_platform):
return (
@@ -159,3 +160,33 @@ class BotReply:
add_reply_on_tags_need_more_info = "订阅标签直接输入标签内容\n屏蔽标签请在标签名称前添加~号\n详见https://nonebot-bison.netlify.app/usage/#%E5%B9%B3%E5%8F%B0%E8%AE%A2%E9%98%85%E6%A0%87%E7%AD%BE-tag"
add_reply_abort = "已中止订阅"
no_permission = "您没有权限进行此操作，请联系 Bot 管理员"
@staticmethod
def add_reply_on_add_cookie(platform_manager, common_platform):
from nonebot_bison.utils.site import is_cookie_client_manager
return (
"请输入想要添加 Cookie 的平台，目前支持，请输入冒号左边的名称：\n"
+ "".join(
[
f"{platform_name}: {platform_manager[platform_name].name}\n"
for platform_name in common_platform
if is_cookie_client_manager(platform_manager[platform_name].site.client_mgr)
]
)
+ "要查看全部平台请输入：“全部”\n中止添加cookie过程请输入：“取消”"
)
@staticmethod
def add_reply_on_add_cookie_input_allplatform(platform_manager):
from nonebot_bison.utils.site import is_cookie_client_manager
return "全部平台\n" + "\n".join(
[
f"{platform_name}: {platform.name}"
for platform_name, platform in platform_manager.items()
if is_cookie_client_manager(platform.site.client_mgr)
]
)
add_reply_on_input_cookie = "请输入 Cookie"