Files
OpenBB/openbb_platform/extensions/index/integration/test_index_api.py
Danglewood e2f7e597fa feature/cboe-async: Refactor Cboe for Async (#5900)
* refactor cboe for async

* codespell

* unused imports

* pylint

* more pylint

* platform metadata

* classmethod decorator

* test params

* intrinio quote

* test param names

* date in test needs to be a datetime.date not string

* make fmp unix timestamp tz-aware and normalize percent change

* different way of parsing datetime string

* recapture cboe equity historical test

* unused import

* index snapshot update

* black

* extension map

* pylint

* cboe pyproject.toml

* proposal to standardize equity_quote

* one more file

* removing changes on static assets

* index_historical

* index_historical again

* last_time -> last_timestamp

* obsolete router path

* fix tests

* test param

* restore market index deprecation that I forgot about

* fixing tests

---------

Co-authored-by: hjoaquim <h.joaquim@campus.fct.unl.pt>
2024-01-22 15:17:42 +00:00

349 lines
9.6 KiB
Python

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
@pytest.fixture(scope="session")
def headers():
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')}"}
# pylint: disable=redefined-outer-name
@parametrize(
"params",
[
({"index": "dowjones", "provider": "fmp"}),
({"index": "BUKBUS", "provider": "cboe"}),
],
)
@pytest.mark.integration
def test_index_constituents(params, headers):
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/index/constituents?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"provider": "fmp",
"sort": "desc",
}
),
(
{
"interval": "1m",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"use_cache": False,
}
),
(
{
"interval": "1d",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"use_cache": False,
}
),
(
{
"interval": "1min",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"timeseries": 1,
"sort": "desc",
}
),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"timeseries": 1,
"sort": "desc",
}
),
(
{
"timespan": "minute",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"multiplier": 1,
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"timespan": "day",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"multiplier": 1,
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"period": "max",
"prepost": True,
"rounding": True,
"provider": "yfinance",
"symbol": "DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"provider": "intrinio",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"symbol": "$DJI",
"tag": "level",
"sort": "desc",
"limit": 100,
"type": None,
}
),
],
)
@pytest.mark.integration
def test_index_price_historical(params, headers):
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/index/price/historical?{query_str}"
result = requests.get(url, headers=headers, timeout=20)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"provider": "fmp",
"sort": "desc",
}
),
(
{
"interval": "1m",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"use_cache": True,
}
),
(
{
"interval": "1d",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"use_cache": False,
}
),
(
{
"interval": "1min",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"timeseries": 1,
"sort": "desc",
}
),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"timeseries": 1,
"sort": "desc",
}
),
(
{
"timespan": "minute",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"multiplier": 1,
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"timespan": "day",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"multiplier": 1,
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"period": "max",
"prepost": True,
"rounding": True,
"provider": "yfinance",
"symbol": "DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"provider": "intrinio",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"symbol": "$DJI",
"tag": "level",
"sort": "desc",
"limit": 100,
"type": None,
}
),
],
)
@pytest.mark.integration
def test_index_market(params, headers):
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/index/market?{query_str}"
result = requests.get(url, headers=headers, timeout=20)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
({"provider": "cboe", "use_cache": False}),
({"provider": "fmp"}),
({"provider": "yfinance"}),
],
)
@pytest.mark.integration
def test_index_available(params, headers):
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/index/available?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
({"query": "D", "is_symbol": True, "provider": "cboe", "use_cache": False}),
],
)
@pytest.mark.integration
def test_index_search(params, headers):
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/index/search?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[({"provider": "cboe", "region": "us"})],
)
@pytest.mark.integration
def test_index_snapshots(params, headers):
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/index/snapshots?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"series_name": "PE Ratio by Month",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"collapse": "monthly",
"transform": "diff",
"provider": "nasdaq",
}
)
],
)
@pytest.mark.integration
def test_index_sp500_multiples(params, headers):
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/index/sp500_multiples?{query_str}"
result = requests.get(url, headers=headers, timeout=20)
assert isinstance(result, requests.Response)
assert result.status_code == 200