exec-mcp/tests/test_bitstamp_rate_limit.py

from __future__ import annotations

from exec_mcp import bitstamp_fx, bitstamp_metadata, bitstamp_rate_limit


class _FakeResponse:
    def __init__(self, payload):
        self._payload = payload

    def raise_for_status(self):
        return None

    def json(self):
        return self._payload


def test_global_rate_limiter_spaces_calls(monkeypatch):
    limiter = bitstamp_rate_limit.GlobalRateLimiter(default_delay_ms=250)
    monkeypatch.setenv("BITSTAMP_CALL_DELAY_MS", "250")

    now = [100.0]
    sleeps: list[float] = []

    monkeypatch.setattr(bitstamp_rate_limit.time, "monotonic", lambda: now[0])
    monkeypatch.setattr(bitstamp_rate_limit.time, "sleep", lambda seconds: sleeps.append(seconds))

    first = limiter.acquire()
    second = limiter.acquire()

    assert first == 0.0
    assert second == 0.25
    assert sleeps == [0.25]


def test_bitstamp_public_helpers_call_throttle(monkeypatch):
    calls: list[str] = []
    monkeypatch.setattr(bitstamp_metadata, "throttle_bitstamp_request", lambda: calls.append("meta"))
    monkeypatch.setattr(bitstamp_fx, "throttle_bitstamp_request", lambda: calls.append("fx"))
    monkeypatch.setattr("exec_mcp.repo.cache_get", lambda *args, **kwargs: None)
    monkeypatch.setattr("exec_mcp.repo.cache_put", lambda *args, **kwargs: None)

    def fake_get(url, *args, **kwargs):
        if url.endswith("/currencies/"):
            return _FakeResponse([{"code": "USD"}])
        if url.endswith("/eur_usd/"):
            return _FakeResponse({"buy": "1.0", "sell": "1.0"})
        raise AssertionError(f"unexpected url: {url}")

    monkeypatch.setattr(bitstamp_metadata.requests, "get", fake_get)

    assert bitstamp_metadata.fetch_currencies() == [{"code": "USD"}]
    assert bitstamp_fx.fetch_eur_usd() == {"buy": "1.0", "sell": "1.0"}
    assert calls == ["meta", "fx"]