"""danding_points HTTP PointsAPI 测试。""" from __future__ import annotations import aiohttp import importlib.util import pytest import sys import types from pathlib import Path def load_points_modules(): """直接加载 danding_points 子模块，避免测试环境缺 nonebot 时执行插件元数据。""" plugin_dir = Path(__file__).resolve().parents[1] / "danding_bot" / "plugins" / "danding_points" package_name = "_danding_points_under_test" package = types.ModuleType(package_name) package.__path__ = [str(plugin_dir)] sys.modules[package_name] = package for module_name in ("config", "api"): 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}.api"], sys.modules[f"{package_name}.config"] api_module, config_module = load_points_modules() PointsAPI = api_module.PointsAPI Config = config_module.Config 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)) def make_points_api() -> PointsAPI: return PointsAPI( Config( POINTS_API_HOST="http://xapi.test/bot/points/", BOT_USER="robot", BOT_TOKEN="secret", ) ) @pytest.fixture def fake_aiohttp(monkeypatch): calls = [] responses = [] monkeypatch.setattr(api_module.aiohttp, "ClientSession", lambda: FakeSession(responses, calls)) return responses, calls @pytest.mark.asyncio async def test_get_balance_sends_auth_in_query(fake_aiohttp): responses, calls = fake_aiohttp responses.append({"code": 200, "message": "", "data": {"balance": 88}}) points = make_points_api() balance = await points.get_balance("10001") assert balance == 88 assert not hasattr(points, "db") assert calls[0]["method"] == "GET" assert calls[0]["url"] == "http://xapi.test/bot/points/balance" assert calls[0]["params"] == {"user": "robot", "token": "secret", "user_id": "10001"} @pytest.mark.asyncio async def test_add_spend_set_send_auth_in_post_body(fake_aiohttp): responses, calls = fake_aiohttp responses.extend( [ {"code": 200, "message": "", "data": {"success": True, "balance": 10}}, {"code": 200, "message": "", "data": {"success": False, "balance": 7}}, {"code": 200, "message": "", "data": {"success": True, "balance": 99}}, ] ) points = make_points_api() add_result = await points.add_points("10001", 10, "gacha_sign", "签到") spend_result = await points.spend_points("10001", 5, "horse_race", "下注") set_result = await points.set_points("10001", 99, "admin", "调整") assert add_result == (True, 10) assert spend_result == (False, 7) assert set_result == (True, 99) assert [call["method"] for call in calls] == ["POST", "POST", "POST"] assert calls[0]["json"] == { "user": "robot", "token": "secret", "user_id": "10001", "amount": 10, "source": "gacha_sign", "reason": "签到", } assert calls[1]["url"].endswith("/spend") assert calls[2]["url"].endswith("/set") @pytest.mark.asyncio async def test_transactions_and_ranking_return_items_unchanged(fake_aiohttp): responses, calls = fake_aiohttp tx_item = { "id": 1, "user_id": "10001", "amount": 10, "balance_after": 10, "source": "gacha_sign", "reason": "签到", "created_at": "2026-06-20T12:00:00", } ranking_item = { "rank": 1, "user_id": "10001", "points": 10, "total_earned": 10, "total_spent": 0, } responses.extend( [ {"code": 200, "message": "", "data": {"items": [tx_item]}}, {"code": 200, "message": "", "data": {"items": [ranking_item]}}, ] ) points = make_points_api() transactions = await points.get_transactions("10001", limit=5, offset=2) ranking = await points.get_ranking(limit=3, order_by="unknown") assert transactions == [tx_item] assert ranking == [ranking_item] assert calls[0]["params"] == { "user": "robot", "token": "secret", "user_id": "10001", "limit": 5, "offset": 2, } assert calls[1]["params"] == { "user": "robot", "token": "secret", "limit": 3, "order_by": "points", } @pytest.mark.asyncio async def test_network_error_keeps_old_failure_returns(monkeypatch): class FailingSession: async def __aenter__(self): raise aiohttp.ClientError("offline") async def __aexit__(self, exc_type, exc, tb): return None monkeypatch.setattr(api_module.aiohttp, "ClientSession", lambda: FailingSession()) points = make_points_api() assert await points.get_balance("10001") == 0 assert await points.add_points("10001", 1, "gacha_sign") == (False, 0) assert await points.spend_points("10001", 1, "horse_race") == (False, 0) assert await points.set_points("10001", 1, "admin") == (False, 0) assert await points.get_transactions("10001") == [] assert await points.get_ranking() == [] @pytest.mark.asyncio async def test_invalid_change_request_does_not_call_http(fake_aiohttp): _responses, calls = fake_aiohttp points = make_points_api() assert await points.add_points("10001", 0, "gacha_sign") == (False, 0) assert await points.spend_points("", 1, "horse_race") == (False, 0) assert await points.set_points("10001", -1, "admin") == (False, 0) assert calls == []