"""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 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} @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)