"""Test crypto API endpoints.""" import base64 import pytest import requests from extensions.tests.conftest import parametrize from openbb_core.env import Env from openbb_core.provider.utils.helpers import get_querystring # pylint: disable=redefined-outer-name @pytest.fixture(scope="session") def headers(): """Get the headers for the API request.""" userpass = f"{Env().API_USERNAME}:{Env().API_PASSWORD}" userpass_bytes = userpass.encode("ascii") base64_bytes = base64.b64encode(userpass_bytes) return {"Authorization": f"Basic {base64_bytes.decode('ascii')}"} @parametrize( "params", [ ({"query": "asd"}), ({"query": "btc", "provider": "fmp"}), ], ) @pytest.mark.integration def test_crypto_search(params, headers): """Test the crypto search endpoint.""" params = {p: v for p, v in params.items() if v} query_str = get_querystring(params, []) url = f"http://0.0.0.0:8000/api/v1/crypto/search?{query_str}" result = requests.get(url, headers=headers, timeout=10) assert isinstance(result, requests.Response) assert result.status_code == 200 @parametrize( "params", [ ( { "interval": "1d", "provider": "fmp", "symbol": "BTCUSD", "start_date": "2023-01-01", "end_date": "2023-01-02", } ), ( { "interval": "1h", "provider": "fmp", "symbol": "BTCUSD,ETHUSD", "start_date": None, "end_date": None, } ), ( { "interval": "1m", "sort": "desc", "limit": 49999, "provider": "polygon", "symbol": "BTCUSD", "start_date": "2023-01-01", "end_date": "2023-01-02", } ), ( { "interval": "1d", "sort": "desc", "limit": 49999, "provider": "polygon", "symbol": "BTCUSD", "start_date": "2023-01-01", "end_date": "2023-06-06", } ), ( { "interval": "1d", "provider": "yfinance", "symbol": "BTCUSD", "start_date": "2023-01-01", "end_date": "2023-01-04", } ), ( { "provider": "tiingo", "interval": "1d", "exchanges": None, "symbol": "BTCUSD", "start_date": "2023-01-01", "end_date": "2023-06-06", } ), ( { "provider": "tiingo", "interval": "1h", "exchanges": ["POLONIEX", "GDAX"], "symbol": "BTCUSD", "start_date": "2023-01-01", "end_date": "2023-01-02", } ), ], ) @pytest.mark.integration def test_crypto_price_historical(params, headers): """Test the crypto historical price endpoint.""" params = {p: v for p, v in params.items() if v} query_str = get_querystring(params, []) url = f"http://0.0.0.0:8000/api/v1/crypto/price/historical?{query_str}" result = requests.get(url, headers=headers, timeout=10) assert isinstance(result, requests.Response) assert result.status_code == 200