374 lines
13 KiB
Python
374 lines
13 KiB
Python
"""onmyoji_gacha HTTP DataManager 测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import aiohttp
|
||
import importlib.util
|
||
import inspect
|
||
import pytest
|
||
import sys
|
||
import types
|
||
from pathlib import Path
|
||
|
||
|
||
def load_gacha_modules():
|
||
"""直接加载 onmyoji_gacha 子模块，避免测试环境执行 nonebot 插件入口。"""
|
||
|
||
plugin_dir = Path(__file__).resolve().parents[1] / "danding_bot" / "plugins" / "onmyoji_gacha"
|
||
package_name = "_onmyoji_gacha_under_test"
|
||
package = types.ModuleType(package_name)
|
||
package.__path__ = [str(plugin_dir)]
|
||
sys.modules[package_name] = package
|
||
|
||
for module_name in ("config", "data_manager", "gacha"):
|
||
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}.data_manager"],
|
||
sys.modules[f"{package_name}.gacha"],
|
||
)
|
||
|
||
|
||
config_module, data_manager_module, gacha_module = load_gacha_modules()
|
||
Config = config_module.Config
|
||
DataManager = data_manager_module.DataManager
|
||
GachaSystem = gacha_module.GachaSystem
|
||
|
||
|
||
def load_sign_in_module():
|
||
"""直接加载签到编排模块，避免导入 NoneBot 插件入口。"""
|
||
|
||
plugin_dir = Path(__file__).resolve().parents[1] / "danding_bot" / "plugins" / "onmyoji_gacha"
|
||
module_name = "_onmyoji_gacha_sign_in_under_test"
|
||
spec = importlib.util.spec_from_file_location(module_name, plugin_dir / "sign_in.py")
|
||
module = importlib.util.module_from_spec(spec)
|
||
sys.modules[module_name] = module
|
||
assert spec and spec.loader
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
sign_in_module = load_sign_in_module()
|
||
|
||
|
||
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):
|
||
calls = []
|
||
responses = []
|
||
monkeypatch.setattr(data_manager_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",
|
||
)
|
||
monkeypatch.setattr(data_manager_module, "config", test_config)
|
||
monkeypatch.setattr(gacha_module, "config", test_config)
|
||
return responses, calls
|
||
|
||
|
||
def success_data(data):
|
||
return {"code": 200, "message": "", "data": data}
|
||
|
||
|
||
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",
|
||
"WEB_ADMIN_TOKEN",
|
||
"DANDING_BOT_USER",
|
||
"DANDING_GACHA_API_HOST",
|
||
):
|
||
monkeypatch.delenv(key, raising=False)
|
||
|
||
class FakeDriver:
|
||
config = types.SimpleNamespace(
|
||
danding_api_token="driver-api-token",
|
||
web_admin_token="driver-admin-token",
|
||
danding_bot_user="driver-user",
|
||
danding_gacha_api_host="http://driver.test/bot/gacha/",
|
||
)
|
||
|
||
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-api-token"
|
||
assert test_config.WEB_ADMIN_TOKEN == "driver-admin-token"
|
||
assert test_config.BOT_USER_ID == "driver-user"
|
||
assert test_config.GACHA_API_HOST == "http://driver.test/bot/gacha"
|
||
|
||
|
||
def test_config_keeps_explicit_bot_token_priority(monkeypatch):
|
||
"""显式抽卡 token 优先于兼容旧 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"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_shikigami_cache_and_draw_send_auth_to_xapi(fake_aiohttp):
|
||
responses, calls = fake_aiohttp
|
||
responses.extend(
|
||
[
|
||
success_data(
|
||
{
|
||
"items": [
|
||
{"id": 1, "name": "灯笼鬼", "rarity": "R", "image_path": "/r/灯笼鬼.png"},
|
||
{"id": 3, "name": "茨木童子", "rarity": "SSR", "image_path": "/ssr/茨木童子.png"},
|
||
]
|
||
}
|
||
),
|
||
success_data(
|
||
{
|
||
"success": True,
|
||
"rarity": "SSR",
|
||
"name": "茨木童子",
|
||
"image_path": "/ssr/茨木童子.png",
|
||
"image_url": "/ssr/茨木童子.png",
|
||
"draws_left": 2,
|
||
"unlocked_achievements": ["no_ssr_60"],
|
||
}
|
||
),
|
||
]
|
||
)
|
||
manager = DataManager()
|
||
|
||
grouped = await manager.refresh_shikigami_data()
|
||
draw_result = await manager.record_draw_result("10001", "SSR", grouped["SSR"][0])
|
||
|
||
assert grouped["SSR"][0]["image_url"] == "/ssr/茨木童子.png"
|
||
assert draw_result["unlocked_achievements"] == ["no_ssr_60"]
|
||
assert calls[0]["method"] == "GET"
|
||
assert calls[0]["url"] == "http://xapi.test/bot/gacha/shikigami"
|
||
assert calls[0]["params"] == {"user": "robot", "token": "secret"}
|
||
assert calls[1]["method"] == "POST"
|
||
assert calls[1]["json"] == {
|
||
"user": "robot",
|
||
"token": "secret",
|
||
"user_id": "10001",
|
||
"shikigami_id": 3,
|
||
"rarity": "SSR",
|
||
"name": "茨木童子",
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_gacha_system_draw_keeps_probability_local_and_uses_xapi(fake_aiohttp, monkeypatch):
|
||
responses, calls = fake_aiohttp
|
||
responses.extend(
|
||
[
|
||
success_data({"draws_left": 3, "daily_limit": 3}),
|
||
success_data({"items": [{"id": 3, "name": "茨木童子", "rarity": "SSR", "image_path": "/ssr/茨木童子.png"}]}),
|
||
success_data(
|
||
{
|
||
"success": True,
|
||
"rarity": "SSR",
|
||
"name": "茨木童子",
|
||
"image_path": "/ssr/茨木童子.png",
|
||
"image_url": "/ssr/茨木童子.png",
|
||
"draws_left": 2,
|
||
"unlocked_achievements": [],
|
||
}
|
||
),
|
||
]
|
||
)
|
||
manager = DataManager()
|
||
monkeypatch.setattr(gacha_module, "data_manager", manager)
|
||
system = GachaSystem()
|
||
monkeypatch.setattr(system, "_draw_rarity", lambda user_id=None: "SSR")
|
||
|
||
result = await system.draw("10001")
|
||
|
||
assert result == {
|
||
"success": True,
|
||
"rarity": "SSR",
|
||
"name": "茨木童子",
|
||
"image_url": "/ssr/茨木童子.png",
|
||
"draws_left": 2,
|
||
"unlocked_achievements": [],
|
||
}
|
||
assert [call["url"].rsplit("/", 1)[-1] for call in calls] == ["draws-left", "shikigami", "draw"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_triple_sign_in_claim_and_query_shapes(fake_aiohttp):
|
||
responses, calls = fake_aiohttp
|
||
responses.extend(
|
||
[
|
||
success_data({"success": True, "results": [], "draws_left": 0, "unlocked_achievements": ["no_ssr_60"]}),
|
||
success_data({"success": True, "signed_already": True}),
|
||
success_data({"success": True, "reward_type": "天卡"}),
|
||
success_data({"success": True, "total_draws": 1, "R_count": 1, "SR_count": 0, "SSR_count": 0, "SP_count": 0, "recent_draws": []}),
|
||
success_data({"success": True, "date": "2026-06-20", "stats": {"total_users": 1}}),
|
||
success_data({"items": [{"user_id": "10001", "total_draws": 1, "R_count": 1, "SR_count": 0, "SSR_count": 0, "SP_count": 0, "ssr_sp_total": 0}]}),
|
||
success_data({"achievements": {"no_ssr_60": {"unlocked_date": "2026-06-20", "reward_claimed": False}}, "progress": {"no_ssr_streak": 60}}),
|
||
success_data({"success": True, "date": "2026-06-20", "records": [], "total_count": 0}),
|
||
]
|
||
)
|
||
manager = DataManager()
|
||
draws = [
|
||
{"id": 1, "name": "灯笼鬼", "rarity": "R", "image_url": "/r/灯笼鬼.png"},
|
||
{"id": 2, "name": "雪女", "rarity": "SR", "image_url": "/sr/雪女.png"},
|
||
{"id": 3, "name": "茨木童子", "rarity": "SSR", "image_url": "/ssr/茨木童子.png"},
|
||
]
|
||
|
||
triple = await manager.record_triple_draw_result("10001", draws)
|
||
signed = await manager.record_sign_in("10001", 20)
|
||
claimed = await manager.claim_achievement_reward("10001", "no_ssr_60")
|
||
stats = await manager.get_user_stats("10001")
|
||
daily = await manager.get_daily_stats("2026-06-20")
|
||
rank = await manager.get_rank()
|
||
achievements = await manager.get_user_achievements("10001")
|
||
records = await manager.get_daily_records("2026-06-20")
|
||
|
||
assert triple["unlocked_achievements"] == ["no_ssr_60"]
|
||
assert signed is False
|
||
assert claimed is True
|
||
assert stats["success"] is True
|
||
assert daily["date"] == "2026-06-20"
|
||
assert rank[0]["user_id"] == "10001"
|
||
assert "unlocked" in achievements and "progress" in achievements
|
||
assert records["total_count"] == 0
|
||
assert calls[0]["json"]["draws"][0] == {"shikigami_id": 1, "rarity": "R", "name": "灯笼鬼"}
|
||
assert calls[1]["json"]["points_awarded"] == 20
|
||
assert calls[2]["json"]["achievement_id"] == "no_ssr_60"
|
||
assert calls[3]["params"]["user_id"] == "10001"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_network_error_keeps_failure_shapes_and_no_sqlite(monkeypatch):
|
||
class FailingSession:
|
||
async def __aenter__(self):
|
||
raise aiohttp.ClientError("offline")
|
||
|
||
async def __aexit__(self, exc_type, exc, tb):
|
||
return None
|
||
|
||
monkeypatch.setattr(data_manager_module.aiohttp, "ClientSession", lambda: FailingSession())
|
||
manager = DataManager()
|
||
|
||
assert await manager.get_draws_left("10001") == 0
|
||
assert await manager.record_sign_in("10001", 20) is False
|
||
assert await manager.get_rank() == []
|
||
assert "sqlite3" not in inspect.getsource(data_manager_module)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_daily_sign_in_awards_points_only_after_sign_recorded():
|
||
calls = []
|
||
|
||
class FakeDataManager:
|
||
async def record_sign_in(self, user_id, points_awarded):
|
||
calls.append(("sign-in", user_id, points_awarded))
|
||
return True
|
||
|
||
class FakePointsAPI:
|
||
async def add_points(self, user_id, amount, source, reason):
|
||
calls.append(("add-points", user_id, amount, source, reason))
|
||
return True, 66
|
||
|
||
success, balance, status = await sign_in_module.award_daily_sign_in_points(
|
||
data_manager=FakeDataManager(),
|
||
points_api=FakePointsAPI(),
|
||
user_id="10001",
|
||
points=20,
|
||
source="gacha_sign",
|
||
reason="抽卡签到",
|
||
)
|
||
|
||
assert (success, balance, status) == (True, 66, "awarded")
|
||
assert calls == [
|
||
("sign-in", "10001", 20),
|
||
("add-points", "10001", 20, "gacha_sign", "抽卡签到"),
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_daily_sign_in_repeat_does_not_award_points():
|
||
calls = []
|
||
|
||
class FakeDataManager:
|
||
async def record_sign_in(self, user_id, points_awarded):
|
||
calls.append(("sign-in", user_id, points_awarded))
|
||
return False
|
||
|
||
class FakePointsAPI:
|
||
async def add_points(self, user_id, amount, source, reason):
|
||
calls.append(("add-points", user_id, amount, source, reason))
|
||
return True, 66
|
||
|
||
success, balance, status = await sign_in_module.award_daily_sign_in_points(
|
||
data_manager=FakeDataManager(),
|
||
points_api=FakePointsAPI(),
|
||
user_id="10001",
|
||
points=20,
|
||
source="gacha_sign",
|
||
reason="抽卡签到",
|
||
)
|
||
|
||
assert (success, balance, status) == (False, 0, "signed_already_or_failed")
|
||
assert calls == [("sign-in", "10001", 20)]
|