exec-mcp/tests/test_bitstamp_private_ws.py

from __future__ import annotations

from exec_mcp import repo, storage
from exec_mcp.bitstamp_private_ws import _get_token
from exec_mcp.services_bitstamp import clear_bitstamp_trading_client


class _FakeBitstampClient:
    constructions = 0

    def __init__(self, username, api_key, api_secret):
        type(self).constructions += 1
        self.username = username
        self.api_key = api_key
        self.api_secret = api_secret

    def websocket_token(self):
        return {"token": "token-123", "user_id": "user-456"}


def test_get_token_reuses_cached_bitstamp_client(tmp_path, monkeypatch):
    db_path = tmp_path / "exec.sqlite3"
    monkeypatch.setattr(storage, "DB_PATH", db_path)
    storage.init_db()

    repo.create_account(display_name="strategy", venue="bitstamp", venue_account_ref="ref-1", api_key="k", api_secret="s")
    account_id = repo.list_accounts(enabled_only=False)[0]["id"]

    clear_bitstamp_trading_client(account_id)
    monkeypatch.setattr("exec_mcp.services_bitstamp.BitstampClient", _FakeBitstampClient)
    _FakeBitstampClient.constructions = 0

    first = _get_token(account_id)
    second = _get_token(account_id)

    assert first == {"token": "token-123", "user_id": "user-456"}
    assert second == first
    assert _FakeBitstampClient.constructions == 1