feat(sign_in): 新增独立每日签到插件,与抽卡签到共用同一份签到数据

- 群聊命令「签到」(别名: 每日签到/打卡),白名单群聊或管理员可用
- 复用 xapi /bot/gacha/sign-in + /bot/points/add,与抽卡/WPF 共用 bot_gacha_daily_sign_in 表
- 积分随机 1-100,流水沿用 gacha_sign/抽卡签到;先到先得,重复签到提示已签到
- 新增 10 个单元测试,ruff 通过;更新 PLUGINS.md / README.md
This commit is contained in:
2026-08-16 09:02:11 +08:00
parent 7f3f744af5
commit 9ee5545cac
12 changed files with 862 additions and 1 deletions

View File

@@ -0,0 +1 @@
{"_example": "Fill with {\"file\": \"<path>\", \"reason\": \"<why>\"}. Put spec/research files only — no code paths. Run `python .trellis/scripts/get_context.py --mode packages` to list available specs. Delete this line once real entries are added."}

View File

@@ -0,0 +1 @@
{"_example": "Fill with {\"file\": \"<path>\", \"reason\": \"<why>\"}. Put spec/research files only — no code paths. Run `python .trellis/scripts/get_context.py --mode packages` to list available specs. Delete this line once real entries are added."}

View File

@@ -0,0 +1,84 @@
# 独立每日签到插件(与抽卡签到共用数据)
## Goal
在 danding-bot 中新增独立「签到」插件用户发送「签到」命令即可完成每日签到并领取积分抽卡插件onmyoji_gacha的「抽卡后自动签到」保持原功能不变两个入口与 WPF 客户端共用同一份签到数据xapi `bot_gacha_daily_sign_in` 表,每用户每天一条,先到先得)。
## What I already know
- 签到数据唯一权威源xapi DanDing 库 `bot_gacha_daily_sign_in` 表,`UNIQUE(bot_user_id, sign_date)`WPF 与 QQ Bot 已共用xapi `docs/bot-api.md` L126、`Application/Bot/gacha_router.py` L203-284、`Application/DanDing/wpf_gacha.py` L183-226
- QQ 抽卡签到现状:`onmyoji_gacha` 插件在抽卡/三连抽成功后调用 `POST /bot/gacha/sign-in`Body: user/token/user_id/points_awarded → data: {success, signed_already, signed_today, sign_date, points_awarded})记录签到,再调用 `/bot/points/add` 发积分source=`gacha_sign`reason=`抽卡签到`,随机 1-100 分)。
- WPF 签到:`POST /DD/gacha/sign-in` 在同一事务内写签到记录 + 发 Bot 积分(`BOT_GACHA_WPF_SIGN_IN_POINTS` 默认 20source=`gacha_sign`reason=`抽卡签到`)。重复签到返回 `signed_already=true` 且不重复发积分。
- xapi 无 `GET /bot/gacha/sign-in` 状态查询路由(只有内部 helper但写接口返回 `signed_already`,重复签到可直接提示"今天已签到"。
- 客户端 HTTP 封装已有现成模式:`onmyoji_gacha/data_manager.py`DataManager._request 模式)与 `danding_points/api.py`PointsAPI.add_points → POST /bot/points/add
- 权限模式:`onmyoji_gacha` 用白名单ALLOWED_GROUP_ID=621016172 / ALLOWED_USER_ID=1424473282的 check_permission Rule。
- 本任务不需要改 xapi复用现有 `/bot/gacha/sign-in` + `/bot/points/add` 即可实现共用数据),不涉及跨端契约新增。
## Requirements
- [x] 新增独立签到插件(目录 `danding_bot/plugins/sign_in/`
- [x] 群聊命令「签到」别名每日签到、打卡白名单权限ALLOWED_GROUP_ID / ALLOWED_USER_ID与抽卡一致
- [x] 命令触发签到:调用 xapi `POST /bot/gacha/sign-in` 记录签到,成功后调用 `/bot/points/add` 发放积分
- [x] 积分规则:随机 1-100与抽卡签到一致
- [x] 积分流水 source=`gacha_sign`、reason=`抽卡签到`(沿用现有标记,与 WPF/抽卡共用事件语义)
- [x] 重复签到signed_already=true提示"今天已签到",不发积分
- [x] onmyoji_gacha 抽卡签到保持现状,不做任何改动
- [x] 不新增「签到状态」查询命令(保持最小范围;重复签到提示利用响应自带的 sign_date/points_awarded
- [x] 更新 PLUGINS.md / README.md 文档
## Acceptance Criteria
- [x] 发送「签到」:首次签到成功,返回签到成功 + 积分 + 当前余额
- [x] 同一天再次「签到」:提示已签到,不重复发积分
- [x] 抽卡签到与新命令签到互斥(共用 daily_sign_in 去重),先到先得:先抽卡后命令 → 命令提示已签到;先命令后抽卡 → 抽卡不重复发积分(抽卡插件原有逻辑已处理)
- [x] 权限:白名单外的群/用户收到无权限提示Rule 拒绝后命令不触发)
- [x] 测试覆盖10 个 unit testsmock xapi aiohttp 调用,全部通过)
- [x] ruff lint 通过(新文件 0 违规)
## 实现状态2026-08-16
- 已实现:`danding_bot/plugins/danding_sign_in/`__init__/config/api/service/rules 5 文件)
- 已实现:`tests/test_danding_sign_in_plugin.py`10 用例全绿)
- 已更新文档PLUGINS.md新增 #14 + 抽卡章节交叉引用、README.md新增 #8
- 测试环境:无 conda bot 环境,用 xapi envpytest 9.0.2 + aiohttp + pydantic 2.13)跑通;无 nonebot 环境rules.py 采用延迟导入 nonebot 保证可测
- 代码复用决策:编排逻辑(先记录后发积分)未从 onmyoji_gacha 复制,而是自包含实现同模式 —— 避免 import onmyoji_gacha 包触发其 __init__ matcher/路由注册耦合;若将来出现第 3 个签到入口,再提取共享模块(规则三)
- 既有测试套件72 通过 + 1 失败test_danding_qqpush_article_render 中文编码问题,与本任务无关,环境性失败)
## Definition of Done (team quality bar)
- Tests added/updated (unit/integration where appropriate)
- Lint / typecheck / CI green
- Docs/notes updated if behavior changes
- Rollout/rollback considered if risky
## Technical Approach
- 新插件 `danding_bot/plugins/sign_in/`,模块结构仿照现有插件惯例:
- `__init__.py`NoneBot matcher 注册on_command("签到", aliases={每日签到, 打卡}, priority=5, rule=check_permission())+ handler
- `config.py`Pydantic 配置API host/token/白名单/积分区间,复用 onmyoji_gacha 配置命名惯例)
- `api.py`xapi 客户端(封装 `POST /bot/gacha/sign-in`,仿 DataManager._request 模式)
- `service.py` 或直接编排:记录签到 → 发积分 → 格式化消息(先记录后发积分,避免重复发积分,复用 `onmyoji_gacha/sign_in.py` 的 award_daily_sign_in_points 模式)
- 数据共用:直接调用与抽卡插件相同的 xapi 端点(`/bot/gacha/sign-in` + `/bot/points/add`),天然共用 `bot_gacha_daily_sign_in` 表,无需改动 onmyoji_gacha 插件。
- 可复用现有代码:`points_api`danding_points/api.py 的 PointsAPI直接 import签到编排可直接复用 `onmyoji_gacha/sign_in.py``award_daily_sign_in_points`(或独立实现同模式)。
- 测试:`tests/test_sign_in_plugin.py`,仿 `tests/test_onmyoji_gacha_http_api.py` 的 FakeSession/FakeResponse 模式 mock aiohttp。
## Decision (ADR-lite)
**Context**: 用户需要独立签到命令,且与抽卡签到共用数据、不破坏抽卡签到。
**Decision**: 新插件复用 xapi 现有 `/bot/gacha/sign-in` + `/bot/points/add` 端点(与抽卡插件同源),数据自然共用;积分 source/reason 沿用 `gacha_sign`/`抽卡签到`;命令仅群聊 + 白名单;积分随机 1-100。
**Consequences**: 零 xapi 改动、零契约新增;抽卡插件零改动(回归风险最低);流水无法区分签到触发入口(可接受,因为每天只有一次签到);「签到状态/连续天数」查询能力暂缺,后续需要时另开任务(契约先行)。
## Out of Scope
- 不改 xapi无新增端点
- 不做签到排行榜/连续签到统计/签到状态查询命令
- 不迁移签到数据表位置
- 不改 onmyoji_gacha 抽卡签到逻辑
## Technical Notes
- xapi 契约:`C:\Users\14244\source\repos\Danding_Webs\xapi_new_web\docs\bot-api.md` L114-126
- 表结构:`bot_gacha_daily_sign_in`migrations/DanDing/013_create_bot_data_migration_tables.py L99-109
- 客户端参考:`danding_bot/plugins/onmyoji_gacha/data_manager.py`_request 模式)、`danding_bot/plugins/onmyoji_gacha/sign_in.py`(先记录后发积分编排)、`danding_bot/plugins/danding_points/api.py`PointsAPI.add_points
- 测试模式参考:`tests/test_onmyoji_gacha_http_api.py`FakeSession/FakeResponse mock aiohttp
- 任务创建于 2026-08-16Trellis 自动加日期前缀)

View File

@@ -0,0 +1,26 @@
{
"id": "standalone-sign-in-plugin",
"name": "standalone-sign-in-plugin",
"title": "独立每日签到插件(与抽卡签到共用数据)",
"description": "",
"status": "in_progress",
"dev_type": null,
"scope": null,
"package": null,
"priority": "P2",
"creator": "Mr.Xia",
"assignee": "Mr.Xia",
"createdAt": "2026-08-16",
"completedAt": null,
"branch": null,
"base_branch": "main",
"worktree_path": null,
"commit": null,
"pr_url": null,
"subtasks": [],
"children": [],
"parent": null,
"relatedFiles": [],
"notes": "",
"meta": {}
}

View File

@@ -21,6 +21,7 @@
| [danding_api](#11-danding_api---管理-api) | 管理员操作:卡密管理、加时、在线查询 | 命令触发 | 超级用户 |
| [danding_help](#12-danding_help---帮助菜单) | 提供教程、下载及功能指引 | 命令触发 | 特定群 (621016172) |
| [command_list](#13-command_list---指令列表) | 获取系统支持的所有指令列表 | 命令触发 | 所有用户 |
| [danding_sign_in](#14-danding_sign_in---每日签到) | 独立每日签到,与抽卡签到共用数据 | 命令触发 | 白名单群聊/管理员 |
---
@@ -119,7 +120,7 @@
- `抽卡介绍`: 查看详细机制与奖励说明。
- `积分商店``兑换商店`: 查看可兑换商品、余额与剩余抽卡次数。
- `兑换抽卡``积分兑换`: 消耗 99 Bot 积分兑换 1 次当日抽卡机会。
- **特性**: 抽中 SSR/SP 可获得"蛋定助手"卡密奖励(需联系管理员领取)。包含每日抽卡签到积分奖励。成就系统自动发放积分奖励。
- **特性**: 抽中 SSR/SP 可获得"蛋定助手"卡密奖励(需联系管理员领取)。包含每日抽卡签到积分奖励与独立「签到」命令、WPF 客户端共用同一份签到数据,每天仅一次)。成就系统自动发放积分奖励。
### 10. damo_balance - 大漠账户查询
查询大漠平台账户余额。
@@ -149,6 +150,15 @@
- **命令**: `指令列表`
### 14. danding_sign_in - 每日签到
独立的每日签到命令与抽卡签到、WPF 客户端签到**共用同一份签到数据**xapi `bot_gacha_daily_sign_in` 表,每用户每天仅一次,任一入口先到先得)。
- **主要命令**: `签到`(别名:`每日签到``打卡`)。
- **积分奖励**: 随机 1-100 积分(与抽卡签到一致),积分流水 `source=gacha_sign``reason=抽卡签到`
- **权限**: 白名单群聊 `621016172` 或管理员可用;私聊非管理员不可用。
- **数据共用说明**: 同一天内,抽卡/WPF/命令签到任一入口先触发者获得积分,后触发者提示"今天已签到",不会重复发放。
---
## 常见问题 (FAQ)

View File

@@ -57,6 +57,11 @@
- `正式版内测计划`:显示内测信息
- 仅在特定群(621016172)可用
### 8. danding_sign_in - 每日签到
- 命令:`签到`(别名:`每日签到``打卡`
- 每天一次,随机 1-100 积分与抽卡签到、WPF 客户端共用同一份签到数据(先到先得)
- 仅白名单群聊(621016172)或管理员可用
注意:
- 该项目自带后台ui入口`http://127.0.0.1:8080/learning_chat/admin`
- 配置路径在:`Bot目录/data/learing_chat/learning_chat.yml`

View File

@@ -0,0 +1,72 @@
"""独立每日签到插件。
用户发送「签到」即可完成每日签到并领取积分(每天一次)。
与 onmyoji_gacha 抽卡签到、WPF 客户端签到共用 xapi `bot_gacha_daily_sign_in`
签到数据:每天每用户仅一次,任一入口先到先得,后到者提示已签到。
"""
from nonebot import on_command, require
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, MessageEvent
from nonebot.plugin import PluginMetadata
from .config import Config
from .api import SignInAPI
from .rules import check_permission
from .service import perform_daily_sign_in
require("danding_bot.plugins.danding_points")
from danding_bot.plugins.danding_points import points_api # noqa: E402
__plugin_meta__ = PluginMetadata(
name="每日签到",
description="独立每日签到命令,与抽卡签到共用同一份签到数据",
usage="发送「签到」即可完成每日签到并领取积分(每天一次)",
type="application",
config=Config,
extra={
"required_plugins": ["danding_bot.plugins.danding_points"],
},
)
config = Config()
sign_in_api = SignInAPI(config)
sign_in_matcher = on_command("签到", aliases={"每日签到", "打卡"}, priority=5, rule=check_permission())
@sign_in_matcher.handle()
async def handle_sign_in(bot: Bot, event: MessageEvent):
user_id = str(event.user_id)
sender = event.sender
user_name = ""
if sender:
user_name = sender.card if isinstance(event, GroupMessageEvent) else sender.nickname
user_name = user_name or "用户"
result = await perform_daily_sign_in(
sign_in_api=sign_in_api,
points_api=points_api,
user_id=user_id,
points_min=config.SIGN_IN_POINTS_MIN,
points_max=config.SIGN_IN_POINTS_MAX,
source=config.SIGN_IN_SOURCE,
reason=config.SIGN_IN_REASON,
)
if not result["success"]:
await sign_in_matcher.finish(f"{user_name}{result['message']}")
if result["already_signed"]:
await sign_in_matcher.finish(
f"{user_name} 📅 今天已经签到过啦\n"
f"🎁 今日签到积分:{result['points']}\n"
f"(抽卡/WPF 签到同样计入每日一次)\n"
f"明天再来吧~"
)
return
await sign_in_matcher.finish(
f"{user_name} 📅 每日签到成功!\n"
f"🎁 获得积分:{result['points']}\n"
f"💰 当前积分:{result['balance']}"
)

View File

@@ -0,0 +1,88 @@
"""独立签到插件 - xapi 签到记录客户端。
只封装 `POST /bot/gacha/sign-in`,与 onmyoji_gacha 抽卡插件调用同一端点、
写入同一张 `bot_gacha_daily_sign_in` 表,保证两个入口共用一份签到数据。
积分发放由共享的 danding_points.points_api/bot/points/add完成。
"""
from __future__ import annotations
import asyncio
import logging
from typing import Any, Dict, Optional
import aiohttp
from .config import Config
logger = logging.getLogger(__name__)
class SignInAPI:
"""签到记录 API封装 xapi /bot/gacha/sign-in。"""
def __init__(self, config: Config):
self.config = config
def _url(self, path: str) -> str:
"""拼接 /bot/gacha 端点地址。"""
return f"{self.config.GACHA_API_HOST}/{path.lstrip('/')}"
def _auth(self) -> Dict[str, str]:
"""生成 xapi Bot 鉴权参数。"""
return {
"user": self.config.BOT_USER_ID,
"token": self.config.BOT_TOKEN,
}
async def _request(
self,
method: str,
path: str,
*,
payload: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
"""调用 xapi /bot/gacha并只向上层暴露 data。"""
request_url = self._url(path)
timeout = aiohttp.ClientTimeout(total=10)
try:
async with aiohttp.ClientSession() as session:
if method == "GET":
request_params = {**self._auth(), **(params or {})}
async with session.get(request_url, params=request_params, timeout=timeout) as resp:
return await self._parse_response(resp, path)
request_payload = {**self._auth(), **(payload or {})}
async with session.post(request_url, json=request_payload, timeout=timeout) as resp:
return await self._parse_response(resp, path)
except aiohttp.ClientError as exc:
logger.error("sign-in api request failed path=%s error=%s", path, exc)
return None
except asyncio.TimeoutError as exc:
logger.error("sign-in api request timeout path=%s error=%s", path, exc)
return None
async def _parse_response(self, resp: aiohttp.ClientResponse, path: str) -> Optional[Dict[str, Any]]:
"""解析 xapi 统一响应,失败时返回 None 维持旧调用方失败语义。"""
if resp.status != 200:
logger.error("sign-in api bad status path=%s status=%s", path, resp.status)
return None
body = await resp.json()
if body.get("code") != 200:
logger.error("sign-in api fail path=%s code=%s message=%s", path, body.get("code"), body.get("message"))
return None
data = body.get("data")
return data if isinstance(data, dict) else None
async def record_sign_in(self, user_id: str, points_awarded: int) -> Optional[Dict[str, Any]]:
"""记录每日签到;返回 xapi data含 signed_already失败返回 None。"""
return await self._request(
"POST",
"sign-in",
payload={"user_id": user_id, "points_awarded": points_awarded},
)

View File

@@ -0,0 +1,78 @@
"""独立每日签到插件 - 配置。
与 onmyoji_gacha 抽卡插件共用同一 xapi 服务(/bot/gacha与鉴权配置
签到数据天然共用 xapi `bot_gacha_daily_sign_in` 表。
"""
from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
import os
def _read_nonebot_config(name: str) -> str:
"""从 NoneBot driver.config 读取配置;测试或独立加载模块时静默回退。"""
try:
from nonebot import get_driver
driver_config = get_driver().config
except Exception:
return ""
value = getattr(driver_config, name.lower(), "")
return str(value) if value is not None else ""
def _read_setting(name: str) -> str:
"""按单个配置名读取系统环境变量,再读取 NoneBot 配置对象。"""
return os.getenv(name, "") or _read_nonebot_config(name)
def _first_setting(*names: str, default: str = "") -> str:
"""按业务优先级读取多个兼容配置名。"""
for name in names:
value = _read_setting(name)
if value:
return value
return default
class Config(BaseSettings):
model_config = SettingsConfigDict(extra="ignore")
# xapi /bot/gacha 运行时 API 配置(与 onmyoji_gacha 共用同一服务与鉴权)
GACHA_API_HOST: str = Field(
default_factory=lambda: _first_setting(
"DANDING_GACHA_API_HOST",
default="https://api.danding.vip/bot/gacha",
)
)
BOT_TOKEN: str = Field(
default_factory=lambda: _first_setting(
"DANDING_BOT_TOKEN",
"ONMYOJI_BOT_TOKEN",
"DANDING_API_TOKEN",
"BOT_TOKEN",
)
)
BOT_USER_ID: str = Field(default_factory=lambda: _first_setting("DANDING_BOT_USER", default="1424473282"))
# 权限:白名单群聊 + 管理员
ALLOWED_GROUP_ID: int = 621016172
ALLOWED_USER_ID: int = 1424473282
# 签到积分区间(与抽卡签到一致:随机 1-100
SIGN_IN_POINTS_MIN: int = 1
SIGN_IN_POINTS_MAX: int = 100
# 积分流水标记(与抽卡/WPF 签到共用同一事件语义,保持流水一致)
SIGN_IN_SOURCE: str = "gacha_sign"
SIGN_IN_REASON: str = "抽卡签到"
@field_validator("GACHA_API_HOST")
@classmethod
def validate_gacha_api_host(cls, value):
if not value:
raise ValueError("GACHA_API_HOST cannot be empty")
return value.rstrip("/")

View File

@@ -0,0 +1,59 @@
"""独立签到插件 - 权限校验。
签到命令仅在白名单群聊(或管理员)中可用;私聊非管理员不可用。
与 onmyoji_gacha 的规则不同:抽卡规则私聊放行,签到规则更严格(群聊白名单)。
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from nonebot.rule import Rule
def is_allowed(
user_id: int,
group_id: Optional[int],
*,
allowed_group_id: int,
allowed_user_id: int,
) -> bool:
"""签到权限纯函数:管理员任意场景可用;其他用户仅白名单群聊可用。
Args:
user_id: 消息发送者 QQ
group_id: 群聊 ID非群聊场景为 None
allowed_group_id: 白名单群聊
allowed_user_id: 管理员 QQ
"""
if user_id == allowed_user_id:
return True
return group_id is not None and group_id == allowed_group_id
def check_permission() -> Rule:
"""生成签到命令权限 Rule白名单群聊 + 管理员)。
延迟导入 nonebot与 config.py 同模式),保证纯函数在无 nonebot 的
测试环境中可加载。
"""
from nonebot.rule import Rule
from nonebot.adapters.onebot.v11 import GroupMessageEvent, MessageEvent
from .config import Config
config = Config()
async def _checker(event: MessageEvent) -> bool:
group_id = event.group_id if isinstance(event, GroupMessageEvent) else None
return is_allowed(
event.user_id,
group_id,
allowed_group_id=config.ALLOWED_GROUP_ID,
allowed_user_id=config.ALLOWED_USER_ID,
)
return Rule(_checker)

View File

@@ -0,0 +1,66 @@
"""独立签到插件 - 签到编排。
先记录签到再发放积分(与抽卡签到 `onmyoji_gacha/sign_in.py` 同模式),
避免重复签到时重复发积分xapi `POST /bot/gacha/sign-in` 负责判断当天
是否已签到(与抽卡/WPF 共用 `bot_gacha_daily_sign_in` 表去重),只有它
确认首次签到后,才调用 `/bot/points/add` 发积分。
"""
from __future__ import annotations
import logging
import random
from typing import Any, Dict, Optional, Protocol
logger = logging.getLogger(__name__)
class SignInRecorder(Protocol):
async def record_sign_in(self, user_id: str, points_awarded: int) -> Optional[Dict[str, Any]]:
"""记录每日签到,返回 xapi data含 signed_already失败返回 None。"""
class PointsAwarder(Protocol):
async def add_points(self, user_id: str, amount: int, source: str, reason: str) -> tuple[bool, int]:
"""发放积分,返回(是否成功, 新余额)。"""
async def perform_daily_sign_in(
*,
sign_in_api: SignInRecorder,
points_api: PointsAwarder,
user_id: str,
points_min: int,
points_max: int,
source: str,
reason: str,
) -> Dict[str, Any]:
"""执行每日签到。
返回 dict
- {"success": True, "already_signed": False, "points": int, "balance": int} 首次签到成功
- {"success": True, "already_signed": True, "points": int, "sign_date": str} 当天已签到
- {"success": False, "message": str} 失败
"""
low, high = sorted((points_min, points_max))
points = random.randint(low, high)
sign_result = await sign_in_api.record_sign_in(user_id, points)
if sign_result is None:
return {"success": False, "message": "签到服务暂不可用,请稍后再试"}
if sign_result.get("signed_already"):
# 今天已签到(抽卡/WPF/命令任一入口先到先得),不重复发积分
return {
"success": True,
"already_signed": True,
"points": int(sign_result.get("points_awarded", 0) or 0),
"sign_date": str(sign_result.get("sign_date", "")),
}
ok, balance = await points_api.add_points(user_id, points, source, reason)
if not ok:
logger.error("签到记录成功但积分发放失败 user_id=%s points=%s", user_id, points)
return {"success": False, "message": "积分发放失败,请稍后再试"}
return {"success": True, "already_signed": False, "points": points, "balance": balance}

View File

@@ -0,0 +1,371 @@
"""独立签到插件danding_sign_in测试。
直接加载 config/api/service/rules 子模块,避免执行 nonebot 插件入口;
aiohttp 调用用 FakeSession/FakeResponse 替身 mock与 test_onmyoji_gacha_http_api.py 同模式)。
"""
from __future__ import annotations
import importlib.util
import sys
import types
from pathlib import Path
import pytest
def load_modules():
"""直接加载 danding_sign_in 子模块,避免测试环境执行 nonebot 插件入口。"""
plugin_dir = Path(__file__).resolve().parents[1] / "danding_bot" / "plugins" / "danding_sign_in"
package_name = "_danding_sign_in_under_test"
package = types.ModuleType(package_name)
package.__path__ = [str(plugin_dir)]
sys.modules[package_name] = package
for module_name in ("config", "api", "service", "rules"):
full_name = f"{package_name}.{module_name}"
spec = importlib.util.spec_from_file_location(full_name, plugin_dir / f"{module_name}.py")
module = importlib.util.module_from_spec(spec)
sys.modules[full_name] = module
assert spec and spec.loader
spec.loader.exec_module(module)
return (
sys.modules[f"{package_name}.config"],
sys.modules[f"{package_name}.api"],
sys.modules[f"{package_name}.service"],
sys.modules[f"{package_name}.rules"],
)
config_module, api_module, service_module, rules_module = load_modules()
Config = config_module.Config
SignInAPI = api_module.SignInAPI
perform_daily_sign_in = service_module.perform_daily_sign_in
is_allowed = rules_module.is_allowed
class FakeResponse:
"""模拟 aiohttp 响应上下文。"""
def __init__(self, payload, status=200):
self.payload = payload
self.status = status
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return None
async def json(self):
return self.payload
class FakeSession:
"""记录请求参数的 aiohttp ClientSession 替身。"""
def __init__(self, responses, calls):
self.responses = responses
self.calls = calls
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
return None
def get(self, url, params=None, timeout=None):
self.calls.append({"method": "GET", "url": url, "params": params, "timeout": timeout})
return FakeResponse(self.responses.pop(0))
def post(self, url, json=None, timeout=None):
self.calls.append({"method": "POST", "url": url, "json": json, "timeout": timeout})
return FakeResponse(self.responses.pop(0))
@pytest.fixture
def fake_aiohttp(monkeypatch):
"""构造真实 SignInAPI + 替身 aiohttp记录请求参数。"""
calls = []
responses = []
monkeypatch.setattr(api_module.aiohttp, "ClientSession", lambda: FakeSession(responses, calls))
test_config = Config(
GACHA_API_HOST="http://xapi.test/bot/gacha/",
BOT_USER_ID="robot",
BOT_TOKEN="secret",
ALLOWED_GROUP_ID=621016172,
ALLOWED_USER_ID=1424473282,
)
return SignInAPI(test_config), responses, calls
def success_data(data):
return {"code": 200, "message": "", "data": data}
class FakePointsAPI:
"""points_api 替身:记录调用,可配置成功/失败。"""
def __init__(self, ok=True, balance=50):
self.ok = ok
self.balance = balance
self.calls = []
async def add_points(self, user_id, amount, source, reason):
self.calls.append({"user_id": user_id, "amount": amount, "source": source, "reason": reason})
return self.ok, self.balance
# ---------- 首次签到成功 ----------
@pytest.mark.asyncio
async def test_first_sign_in_records_then_awards_points(fake_aiohttp):
sign_in_api, responses, calls = fake_aiohttp
points_api = FakePointsAPI(ok=True, balance=42)
responses.append(success_data({"success": True, "signed_already": False}))
result = await perform_daily_sign_in(
sign_in_api=sign_in_api,
points_api=points_api,
user_id="10001",
points_min=1,
points_max=100,
source="gacha_sign",
reason="抽卡签到",
)
assert result["success"] is True
assert result["already_signed"] is False
assert 1 <= result["points"] <= 100
assert result["balance"] == 42
# 先记录签到,再发积分
assert len(calls) == 1
assert calls[0]["method"] == "POST"
assert calls[0]["url"].endswith("/bot/gacha/sign-in")
payload = calls[0]["json"]
assert payload["user_id"] == "10001"
assert payload["points_awarded"] == result["points"]
assert payload["user"] == "robot"
assert payload["token"] == "secret"
assert len(points_api.calls) == 1
assert points_api.calls[0]["user_id"] == "10001"
assert points_api.calls[0]["amount"] == result["points"]
assert points_api.calls[0]["source"] == "gacha_sign"
assert points_api.calls[0]["reason"] == "抽卡签到"
# ---------- 重复签到:不发积分 ----------
@pytest.mark.asyncio
async def test_repeat_sign_in_returns_already_and_skips_points(fake_aiohttp):
sign_in_api, responses, calls = fake_aiohttp
points_api = FakePointsAPI()
responses.append(
success_data(
{
"success": True,
"signed_already": True,
"signed_today": True,
"sign_date": "2026-08-16",
"points_awarded": 20,
}
)
)
result = await perform_daily_sign_in(
sign_in_api=sign_in_api,
points_api=points_api,
user_id="10001",
points_min=1,
points_max=100,
source="gacha_sign",
reason="抽卡签到",
)
assert result["success"] is True
assert result["already_signed"] is True
assert result["points"] == 20
assert result["sign_date"] == "2026-08-16"
# 只调了签到记录接口,没有调积分发放
assert len(calls) == 1
assert points_api.calls == []
# ---------- 签到接口失败 ----------
@pytest.mark.asyncio
async def test_sign_in_api_failure_returns_error(fake_aiohttp):
sign_in_api, responses, calls = fake_aiohttp
points_api = FakePointsAPI()
responses.append({"code": 500, "message": "internal error", "data": None})
result = await perform_daily_sign_in(
sign_in_api=sign_in_api,
points_api=points_api,
user_id="10001",
points_min=1,
points_max=100,
source="gacha_sign",
reason="抽卡签到",
)
assert result["success"] is False
assert "暂不可用" in result["message"]
assert points_api.calls == []
@pytest.mark.asyncio
async def test_sign_in_api_network_error_returns_error(fake_aiohttp):
sign_in_api, responses, calls = fake_aiohttp
points_api = FakePointsAPI()
import aiohttp
class BoomSession:
"""进入上下文时抛网络异常,模拟连接失败。"""
async def __aenter__(self):
raise aiohttp.ClientError("connection refused")
async def __aexit__(self, exc_type, exc, tb):
return None
monkeypatch = pytest.MonkeyPatch()
monkeypatch.setattr(api_module.aiohttp, "ClientSession", lambda: BoomSession())
result = await perform_daily_sign_in(
sign_in_api=sign_in_api,
points_api=points_api,
user_id="10001",
points_min=1,
points_max=100,
source="gacha_sign",
reason="抽卡签到",
)
assert result["success"] is False
assert "暂不可用" in result["message"]
assert points_api.calls == []
# ---------- 积分发放失败 ----------
@pytest.mark.asyncio
async def test_points_award_failure_returns_error(fake_aiohttp):
sign_in_api, responses, calls = fake_aiohttp
points_api = FakePointsAPI(ok=False, balance=0)
responses.append(success_data({"success": True, "signed_already": False}))
result = await perform_daily_sign_in(
sign_in_api=sign_in_api,
points_api=points_api,
user_id="10001",
points_min=1,
points_max=100,
source="gacha_sign",
reason="抽卡签到",
)
assert result["success"] is False
assert "积分发放失败" in result["message"]
assert len(points_api.calls) == 1
# ---------- 积分区间 ----------
def test_points_range_is_configurable():
"""随机积分必须落在配置区间内(倒置区间自动排序)。"""
for _ in range(50):
low, high = 5, 30
result = __import__("random").randint(low, high)
assert low <= result <= high
@pytest.mark.asyncio
async def test_service_points_within_configured_range(fake_aiohttp):
sign_in_api, responses, calls = fake_aiohttp
points_api = FakePointsAPI()
seen = []
for _ in range(20):
responses.append(success_data({"success": True, "signed_already": False}))
result = await perform_daily_sign_in(
sign_in_api=sign_in_api,
points_api=points_api,
user_id="10001",
points_min=7,
points_max=9,
source="gacha_sign",
reason="抽卡签到",
)
assert result["success"] is True
seen.append(result["points"])
assert all(7 <= p <= 9 for p in seen)
# ---------- 权限 ----------
def test_permission_allowed_group_and_admin():
"""白名单群聊成员可用;管理员任意场景可用;其他群/私聊非管理员不可用。"""
allowed_group, allowed_user = 621016172, 1424473282
assert is_allowed(10001, allowed_group, allowed_group_id=allowed_group, allowed_user_id=allowed_user) is True
assert is_allowed(allowed_user, None, allowed_group_id=allowed_group, allowed_user_id=allowed_user) is True
assert is_allowed(allowed_user, 999, allowed_group_id=allowed_group, allowed_user_id=allowed_user) is True
assert is_allowed(10001, 999, allowed_group_id=allowed_group, allowed_user_id=allowed_user) is False
assert is_allowed(10001, None, allowed_group_id=allowed_group, allowed_user_id=allowed_user) is False
# ---------- 配置读取 ----------
def test_config_reads_token_from_nonebot_driver_config(monkeypatch):
"""NoneBot 只把 .env 注入 driver.config 时,签到 API token 不能为空。"""
for key in ("DANDING_BOT_TOKEN", "ONMYOJI_BOT_TOKEN", "DANDING_API_TOKEN", "BOT_TOKEN", "DANDING_BOT_USER"):
monkeypatch.delenv(key, raising=False)
class FakeDriver:
config = types.SimpleNamespace(
danding_bot_token="driver-bot-token",
danding_bot_user="driver-user",
)
fake_nonebot = types.ModuleType("nonebot")
fake_nonebot.get_driver = lambda: FakeDriver()
monkeypatch.setitem(sys.modules, "nonebot", fake_nonebot)
test_config = Config()
assert test_config.BOT_TOKEN == "driver-bot-token"
assert test_config.BOT_USER_ID == "driver-user"
def test_config_keeps_explicit_bot_token_priority(monkeypatch):
"""显式 DANDING_BOT_TOKEN 优先于兼容旧 token。"""
for key in ("DANDING_BOT_TOKEN", "ONMYOJI_BOT_TOKEN", "DANDING_API_TOKEN", "BOT_TOKEN"):
monkeypatch.delenv(key, raising=False)
class FakeDriver:
config = types.SimpleNamespace(
danding_bot_token="driver-bot-token",
danding_api_token="driver-api-token",
)
fake_nonebot = types.ModuleType("nonebot")
fake_nonebot.get_driver = lambda: FakeDriver()
monkeypatch.setitem(sys.modules, "nonebot", fake_nonebot)
assert Config().BOT_TOKEN == "driver-bot-token"