DanDingNoneBot/tests/test_danding_points_http_api.py

"""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 == []