Mr.Xia
260651363a
- rules.py 改为模块顶层导入 nonebot（NoneBot Rule 解析依赖参数注解在全局命名空间求值） - 权限纯函数 is_allowed 拆至 permission.py（无 nonebot 依赖，保持可测性） - 测试加载模块列表同步调整（config/api/service/permission）
376 lines
12 KiB
Python
376 lines
12 KiB
Python
"""独立签到插件（danding_sign_in）测试。
|
||
|
||
直接加载 config/api/service/permission 子模块，避免执行 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 插件入口。
|
||
|
||
rules.py 顶层导入 nonebot（Rule 解析要求注解全局可见），无 nonebot 的
|
||
测试环境不加载它；权限纯函数在 permission.py 中测试。
|
||
"""
|
||
|
||
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", "permission"):
|
||
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}.permission"],
|
||
)
|
||
|
||
|
||
config_module, api_module, service_module, permission_module = load_modules()
|
||
Config = config_module.Config
|
||
SignInAPI = api_module.SignInAPI
|
||
perform_daily_sign_in = service_module.perform_daily_sign_in
|
||
is_allowed = permission_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"
|