from __future__ import annotations
from typing import Any
import httpx
import pytest
from bigbugai_mcp.auth import configure_rate_limiter_from_env
from bigbugai_mcp.models import TokenAnalysisReq, TrendingReq
from bigbugai_mcp.tools import get_trending_tokens, token_analysis_by_contract
class FakeResponse:
def __init__(self, status_code: int, data: Any) -> None:
self.status_code = status_code
self._data = data
self._text = ""
self.request = httpx.Request("GET", "https://api.bigbug.ai/")
self.response = self
def raise_for_status(self) -> None:
if self.status_code >= 400:
raise httpx.HTTPStatusError(
f"status {self.status_code}", request=self.request, response=self
)
def json(self) -> Any:
return self._data
@property
def text(self) -> str: # pragma: no cover - only for debugging
return self._text
class FakeAsyncClient:
def __init__(self, *args: Any, **kwargs: Any) -> None: # pragma: no cover
pass
async def __aenter__(self) -> FakeAsyncClient:
return self
async def __aexit__(self, exc_type, exc, tb) -> None: # type: ignore[no-untyped-def]
return None
async def get(
self,
url: str,
params: dict[str, Any] | None = None,
headers: dict[str, Any] | None = None,
) -> FakeResponse:
if url.endswith("/api/tokens/newly-ingested") or url.endswith("/v1/trending/tokens"):
# Normalize to items
return FakeResponse(200, {"items": [{"symbol": "AAA"}, {"symbol": "BBB"}]})
# Match current token analysis path: /api/token-intel/{chain}/{address}/report
if "/api/token-intel/" in url and url.endswith("/report"):
try:
parts = url.split("/api/token-intel/")[1].split("/")
chain = parts[0]
address = parts[1]
except Exception:
chain = "unknown"
address = "unknown"
return FakeResponse(200, {"chain": chain, "address": address, "score": 42})
return FakeResponse(404, {"error": "not found"})
@pytest.mark.asyncio
async def test_http_tools_with_mock(monkeypatch: pytest.MonkeyPatch) -> None:
# Configure auth + generous rate limit
monkeypatch.setenv("BIGBUGAI_MCP_API_KEY", "secret")
monkeypatch.setenv("MCP_RATE_LIMIT", "100/hour")
configure_rate_limiter_from_env()
# Patch httpx.AsyncClient with fake
monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)
trending = await get_trending_tokens(TrendingReq(limit=2))
assert isinstance(trending, list)
assert len(trending) == 2
analysis = await token_analysis_by_contract(
TokenAnalysisReq(chain="eth", address="0xabc")
)
assert analysis["chain"] == "ethereum"
assert analysis["address"] == "0xabc"
assert analysis["score"] == 42
