jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-27 20:42:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Charting extension integration tests (#5547)

Browse Source 
* initial proposal for integration tests on the charting extension

* int tests for charting generation

* python generator

* ruff

* boilerplate templates

* python tests

* tests for the api

* using body instead so data don't get redefined

* unit tests for coverage

* ruff

* docstrings

* unnecessary access to keys()
This commit is contained in:
Henrique Joaquim
2023-10-12 15:59:57 +01:00
committed by GitHub
parent dc8fe89ab7
commit 5ea8bee755
8 changed files with 1049 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

375
openbb_platform/extensions/charting/integration/test_charting_api.py Normal file
View File

@@ -0,0 +1,375 @@
import base64
import json
import pytest
import requests
from openbb_core.env import Env
from openbb_provider.utils.helpers import get_querystring
@pytest.fixture(scope="session")
def headers():
"""Headers fixture."""
return get_headers()
data = {}
def get_headers():
"""Get headers for requests."""
if "headers" in data:
return data["headers"]
userpass = f"{Env().API_USERNAME}:{Env().API_PASSWORD}"
userpass_bytes = userpass.encode("ascii")
base64_bytes = base64.b64encode(userpass_bytes)
data["headers"] = {"Authorization": f"Basic {base64_bytes.decode('ascii')}"}
return data["headers"]
def get_stocks_data():
"""Get stocks data."""
if "stocks_data" in data:
return data["stocks_data"]
url = "http://0.0.0.0:8000/api/v1/stocks/load?symbol=AAPL&provider=fmp"
result = requests.get(url, headers=get_headers(), timeout=10)
data["stocks_data"] = result.json()["results"]
return data["stocks_data"]
@pytest.mark.parametrize(
"params",
[
(
{
"provider": "fmp",
"symbol": "AAPL",
"chart": True,
}
),
],
)
@pytest.mark.integration
def test_chart_stocks_load(params, headers):
"""Test chart stocks load."""
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/stocks/load?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[({"symbol": "AAPL", "limit": 100, "chart": True})],
)
@pytest.mark.integration
def test_chart_stocks_multiples(params, headers):
"""Test chart stocks multiples."""
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/stocks/multiples?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[({"provider": "yfinance", "symbols": "AAPL", "limit": 20, "chart": True})],
)
@pytest.mark.integration
def test_chart_stocks_news(params, headers):
"""Test chart stocks news."""
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/stocks/news?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "date",
"length": "60",
"scalar": "90.0",
"drift": "2",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_adx(params, headers):
"""Test chart ta adx."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/adx?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "date", "length": "30", "scalar": "110", "chart": True})],
)
@pytest.mark.integration
def test_chart_ta_aroon(params, headers):
"""Test chart ta aroon."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/aroon?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "",
"length": "60",
"offset": "10",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_ema(params, headers):
"""Test chart ta ema."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/ema?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "55",
"offset": "2",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_hma(params, headers):
"""Test chart ta hma."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/hma?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"fast": "10",
"slow": "30",
"signal": "10",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_macd(params, headers):
"""Test chart ta macd."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/macd?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "16",
"scalar": "90.0",
"drift": "2",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_rsi(params, headers):
"""Test chart ta rsi."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/rsi?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "55",
"offset": "2",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_sma(params, headers):
"""Test chart ta sma."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/sma?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "60",
"offset": "10",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_wma(params, headers):
"""Test chart ta wma."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/wma?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "55",
"offset": "5",
"chart": True,
}
)
],
)
@pytest.mark.integration
def test_chart_ta_zlma(params, headers):
"""Test chart ta zlma."""
params = {p: v for p, v in params.items() if v}
body = json.dumps(get_stocks_data())
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/zlma?{query_str}"
result = requests.post(url, headers=headers, timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]

372
openbb_platform/extensions/charting/integration/test_charting_python.py Normal file
View File

@@ -0,0 +1,372 @@
"""Test charting extension."""
import pytest
from openbb_charting.core.openbb_figure import OpenBBFigure
from openbb_core.app.model.obbject import OBBject
@pytest.fixture(scope="session")
def obb(pytestconfig):
"""Fixture to setup obb."""
if pytestconfig.getoption("markexpr") != "not integration":
import openbb
return openbb.obb
data = {}
def get_stocks_data():
"""Get stocks data."""
import openbb # pylint:disable=import-outside-toplevel
if "stocks_data" in data:
return data["stocks_data"]
symbol = "AAPL"
provider = "fmp"
data["stocks_data"] = openbb.obb.stocks.load(
symbol=symbol, provider=provider
).results
return data["stocks_data"]
@pytest.mark.parametrize(
"params",
[
(
{
"provider": "fmp",
"symbol": "AAPL",
"chart": True,
}
),
],
)
@pytest.mark.integration
def test_chart_stocks_load(params, obb):
"""Test chart stocks load."""
result = obb.stocks.load(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "limit": 100, "chart": "True"}),
],
)
@pytest.mark.integration
def test_chart_stocks_multiples(params, obb):
"""Test chart stocks multiples."""
params = {p: v for p, v in params.items() if v}
result = obb.stocks.multiples(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"provider": "fmp",
"symbols": "AAPL",
"limit": 20,
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_stocks_news(params, obb):
"""Test chart stocks news."""
params = {p: v for p, v in params.items() if v}
result = obb.stocks.news(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "date",
"length": "60",
"scalar": "90.0",
"drift": "2",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_adx(params, obb):
"""Test chart ta adx."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.adx(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "date",
"length": "30",
"scalar": "110",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_aroon(params, obb):
"""Test chart ta aroon."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.aroon(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "",
"length": "60",
"offset": "10",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_ema(params, obb):
"""Test chart ta ema."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.ema(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "55",
"offset": "2",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_hma(params, obb):
"""Test chart ta hma."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.hma(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"fast": "10",
"slow": "30",
"signal": "10",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_macd(params, obb):
"""Test chart ta macd."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.macd(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "16",
"scalar": "90.0",
"drift": "2",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_rsi(params, obb):
"""Test chart ta rsi."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.rsi(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "55",
"offset": "2",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_sma(params, obb):
"""Test chart ta sma."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.sma(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "60",
"offset": "10",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_wma(params, obb):
"""Test chart ta wma."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.wma(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "high",
"index": "date",
"length": "55",
"offset": "5",
"chart": "True",
}
),
],
)
@pytest.mark.integration
def test_chart_ta_zlma(params, obb):
"""Test chart ta zlma."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_stocks_data()
result = obb.ta.zlma(**params)
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)

134
openbb_platform/extensions/ta/integration/test_ta_api.py
View File

@@ -12,6 +12,7 @@ data = {}
def get_headers():
"""Get headers."""
if "headers" in data:
return data["headers"]
@@ -24,6 +25,7 @@ def get_headers():
def get_data(menu: Literal["stocks", "crypto"]):
"""Get data either from stocks or crypto."""
funcs = {"stocks": get_stocks_data, "crypto": get_crypto_data}
return funcs[menu]()
@@ -36,6 +38,7 @@ def request_data(menu: str, symbol: str, provider: str):
def get_stocks_data():
"""Get stocks data."""
if "stocks_data" in data:
return data["stocks_data"]
@@ -47,6 +50,7 @@ def get_stocks_data():
def get_crypto_data():
"""Get crypto data."""
if "crypto_data" in data:
return data["crypto_data"]
@@ -91,12 +95,13 @@ def get_crypto_data():
)
@pytest.mark.integration
def test_ta_atr(params, data_type):
"""Test ta atr."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/atr?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -130,12 +135,13 @@ def test_ta_atr(params, data_type):
)
@pytest.mark.integration
def test_ta_fib(params, data_type):
"""Test ta fib."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/fib?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -149,12 +155,13 @@ def test_ta_fib(params, data_type):
)
@pytest.mark.integration
def test_ta_obv(params, data_type):
"""Test ta obv."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/obv?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -168,12 +175,13 @@ def test_ta_obv(params, data_type):
)
@pytest.mark.integration
def test_ta_fisher(params, data_type):
"""Test ta fisher."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/fisher?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -205,12 +213,13 @@ def test_ta_fisher(params, data_type):
)
@pytest.mark.integration
def test_ta_adosc(params, data_type):
"""Test ta adosc."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/adosc?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -246,12 +255,13 @@ def test_ta_adosc(params, data_type):
)
@pytest.mark.integration
def test_ta_bbands(params, data_type):
"""Test ta bbands."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/bbands?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -283,12 +293,13 @@ def test_ta_bbands(params, data_type):
)
@pytest.mark.integration
def test_ta_zlma(params, data_type):
"""Test ta zlma."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/zlma?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -310,12 +321,13 @@ def test_ta_zlma(params, data_type):
)
@pytest.mark.integration
def test_ta_aroon(params, data_type):
"""Test ta aroon."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/aroon?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -347,12 +359,13 @@ def test_ta_aroon(params, data_type):
)
@pytest.mark.integration
def test_ta_sma(params, data_type):
"""Test ta sma."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/sma?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -386,12 +399,13 @@ def test_ta_sma(params, data_type):
)
@pytest.mark.integration
def test_ta_demark(params, data_type):
"""Test ta demark."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/demark?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -405,12 +419,13 @@ def test_ta_demark(params, data_type):
)
@pytest.mark.integration
def test_ta_vwap(params, data_type):
"""Test ta vwap."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/vwap?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -444,12 +459,13 @@ def test_ta_vwap(params, data_type):
)
@pytest.mark.integration
def test_ta_macd(params, data_type):
"""Test ta macd."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/macd?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -481,12 +497,13 @@ def test_ta_macd(params, data_type):
)
@pytest.mark.integration
def test_ta_hma(params, data_type):
"""Test ta hma."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/hma?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -518,12 +535,13 @@ def test_ta_hma(params, data_type):
)
@pytest.mark.integration
def test_ta_donchian(params, data_type):
"""Test ta donchian."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/donchian?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -559,12 +577,13 @@ def test_ta_donchian(params, data_type):
)
@pytest.mark.integration
def test_ta_ichimoku(params, data_type):
"""Test ta ichimoku."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/ichimoku?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -586,12 +605,13 @@ def test_ta_ichimoku(params, data_type):
)
@pytest.mark.integration
def test_ta_clenow(params, data_type):
"""Test ta clenow."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/clenow?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -605,12 +625,13 @@ def test_ta_clenow(params, data_type):
)
@pytest.mark.integration
def test_ta_ad(params, data_type):
"""Test ta ad."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/ad?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -642,12 +663,13 @@ def test_ta_ad(params, data_type):
)
@pytest.mark.integration
def test_ta_adx(params, data_type):
"""Test ta adx."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/adx?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -679,12 +701,13 @@ def test_ta_adx(params, data_type):
)
@pytest.mark.integration
def test_ta_wma(params, data_type):
"""Test ta wma."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/wma?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -706,12 +729,13 @@ def test_ta_wma(params, data_type):
)
@pytest.mark.integration
def test_ta_cci(params, data_type):
"""Test ta cci."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/cci?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -745,12 +769,13 @@ def test_ta_cci(params, data_type):
)
@pytest.mark.integration
def test_ta_rsi(params, data_type):
"""Test ta rsi."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/rsi?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -782,12 +807,13 @@ def test_ta_rsi(params, data_type):
)
@pytest.mark.integration
def test_ta_stoch(params, data_type):
"""Test ta stoch."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/stoch?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -821,12 +847,13 @@ def test_ta_stoch(params, data_type):
)
@pytest.mark.integration
def test_ta_kc(params, data_type):
"""Test ta kc."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/kc?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -840,12 +867,13 @@ def test_ta_kc(params, data_type):
)
@pytest.mark.integration
def test_ta_cg(params, data_type):
"""Test ta cg."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/cg?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -879,12 +907,13 @@ def test_ta_cg(params, data_type):
)
@pytest.mark.integration
def test_ta_cones(params, data_type):
"""Test ta cones."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/cones?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -916,11 +945,12 @@ def test_ta_cones(params, data_type):
)
@pytest.mark.integration
def test_ta_ema(params, data_type):
"""Test ta ema."""
params = {p: v for p, v in params.items() if v}
data = json.dumps(get_data(data_type))
body = json.dumps(get_data(data_type))
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/ema?{query_str}"
result = requests.post(url, headers=get_headers(), timeout=10, data=data)
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200

22
openbb_platform/extensions/tests/test_integration_tests_api.py
View File

@@ -1,4 +1,6 @@
"""Test the integration tests."""
from openbb_core.app.charting_service import ChartingService
from extensions.tests.utils.integration_tests_testers import (
check_missing_integration_test_params,
check_missing_integration_test_providers,
@@ -24,3 +26,23 @@ def test_api_interface_integration_test_providers() -> None:
def test_api_interface_integration_test_params() -> None:
"""Test if there are any missing params for integration tests."""
run_test("api", check_missing_integration_test_params)
def test_charting_extension_function_coverage() -> None:
"""Test if all charting extension functions are covered by integration tests."""
functions = ChartingService.get_implemented_charting_functions()
test_names = [f"test_chart_{func}" for func in functions]
integration_tests_modules = get_integration_tests(
test_type="api", filter_charting_ext=False
)
charting_module = [
module for module in integration_tests_modules if "charting" in module.__name__
]
integration_tests_functions = get_module_functions(charting_module)
missing_items = [
test for test in test_names if test not in integration_tests_functions
]
assert missing_items == [], "\n".join(missing_items)

22
openbb_platform/extensions/tests/test_integration_tests_python.py
View File

@@ -1,4 +1,6 @@
"""Test the integration tests."""
from openbb_core.app.charting_service import ChartingService
from extensions.tests.utils.integration_tests_testers import (
check_missing_integration_test_params,
check_missing_integration_test_providers,
@@ -24,3 +26,23 @@ def test_python_interface_integration_test_providers() -> None:
def test_python_interface_integration_test_params() -> None:
"""Test if there are any missing params for integration tests."""
run_test("python", check_missing_integration_test_params)
def test_charting_extension_function_coverage() -> None:
"""Test if all charting extension functions are covered by integration tests."""
functions = ChartingService.get_implemented_charting_functions()
test_names = [f"test_chart_{func}" for func in functions]
integration_tests_modules = get_integration_tests(
test_type="python", filter_charting_ext=False
)
charting_module = [
module for module in integration_tests_modules if "charting" in module.__name__
]
integration_tests_functions = get_module_functions(charting_module)
missing_items = [
test for test in test_names if test not in integration_tests_functions
]
assert missing_items == [], "\n".join(missing_items)

102
openbb_platform/extensions/tests/utils/integration_tests_api_generator.py
View File

@@ -1,7 +1,9 @@
import argparse
import os
from typing import Dict, List, Type, get_type_hints
from typing import Dict, List, Literal, Type, get_type_hints
import requests
from openbb_core.app.charting_service import ChartingService
from openbb_core.app.provider_interface import ProviderInterface
from openbb_core.app.router import CommandMap
@@ -9,6 +11,7 @@ from extensions.tests.utils.integration_tests_generator import get_test_params
def get_http_method(api_paths: Dict[str, dict], route: str):
"""Given a set of paths and a route, return the http method for that route."""
route_info = api_paths.get(route, None)
if not route_info:
return route_info
@@ -16,10 +19,13 @@ def get_http_method(api_paths: Dict[str, dict], route: str):
def get_post_flat_params(hints: Dict[str, Type]):
"""Flattens the params for a post request."""
return list(hints.keys())
def write_init_test_template(http_method: str, path: str):
"""Write some common initialization for the tests with the defined template."""
http_template_imports = {"get": "", "post": "import json"}
template = http_template_imports[http_method]
template += """
@@ -44,8 +50,14 @@ def headers():
def write_test_w_template(
http_method: str, params_list: List[Dict[str, str]], route: str, path: str
http_method: Literal["post", "get"],
params_list: List[Dict[str, str]],
route: str,
path: str,
chart: bool = False,
):
"""Write the test with the defined template."""
params_str = ",\n".join([f"({params})" for params in params_list])
http_template_request = {
@@ -55,13 +67,20 @@ def write_test_w_template(
http_template_params = {"get": "", "post": "body = json.dumps(params.pop('data'))"}
test_name_extra = "chart_" if chart else ""
chart_extra_template = """
assert result.json()["chart"]
assert list(result.json()["chart"].keys()) == ["content", "format"]
"""
template = f"""
@pytest.mark.parametrize(
"params",
[{params_str}],
)
@pytest.mark.integration
def test_{route.replace("/", "_")[1:]}(params, headers):
def test_{test_name_extra}{route.replace("/", "_")[1:]}(params, headers):
params = {{p: v for p, v in params.items() if v}}
{http_template_params[http_method]}
@@ -71,21 +90,25 @@ def test_{route.replace("/", "_")[1:]}(params, headers):
assert isinstance(result, requests.Response)
assert result.status_code == 200
"""
if chart:
template += chart_extra_template
with open(path, "a") as f:
f.write(template)
def test_exists(route: str, path: str):
"""Check if a test exists."""
with open(path) as f:
return route.replace("/", "_")[1:] in f.read()
def write_integration_tests(
def write_commands_integration_tests(
command_map: CommandMap,
provider_interface: ProviderInterface,
api_paths: Dict[str, dict],
) -> List[str]:
"""Write the commands integration tests."""
commands_not_found = []
commandmap_map = command_map.map
@@ -129,7 +152,57 @@ def write_integration_tests(
return commands_not_found
def write_charting_extension_integration_tests():
"""Write the charting extension integration tests."""
functions = ChartingService.get_implemented_charting_functions()
# we assume test file exists
path = os.path.join(
"openbb_platform",
"extensions",
"charting",
"integration",
"test_charting_api.py",
)
for function in functions:
route = "/" + function.replace("_", "/")
if not test_exists(route=function, path=path):
write_test_w_template(
http_method="post",
params_list=[{"chart": True}],
route=route,
path=path,
chart=True,
)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="API Integration Tests Generator",
description="Generate API integration tests",
)
parser.add_argument(
"--charting",
dest="charting",
default=True,
action="store_false",
help="Generate charting extension integration tests",
)
parser.add_argument(
"--commands",
dest="commands",
default=True,
action="store_false",
help="Generate commands integration tests",
)
args = parser.parse_args()
charting = args.charting
commands = args.commands
r = requests.get("http://0.0.0.0:8000/openapi.json", timeout=10).json()
if not r:
@@ -138,13 +211,16 @@ if __name__ == "__main__":
command_map = CommandMap()
provider_interface = ProviderInterface()
commands_not_found_in_openapi = write_integration_tests(
command_map=command_map,
provider_interface=provider_interface,
api_paths=r["paths"],
)
if commands_not_found_in_openapi:
print( # noqa
f"Commands not found in openapi.json: {commands_not_found_in_openapi}"
if commands:
commands_not_found_in_openapi = write_commands_integration_tests(
command_map=command_map,
provider_interface=provider_interface,
api_paths=r["paths"],
)
if commands_not_found_in_openapi:
print( # noqa
f"Commands not found in openapi.json: {commands_not_found_in_openapi}"
)
if charting:
write_charting_extension_integration_tests()

78
openbb_platform/extensions/tests/utils/integration_tests_generator.py
View File

@@ -1,7 +1,19 @@
"""Integration test generator."""
import argparse
from pathlib import Path, PosixPath
from typing import Any, Dict, List, Literal, Tuple, Type, get_origin, get_type_hints
from typing import (
Any,
Dict,
List,
Literal,
Optional,
Tuple,
Type,
get_origin,
get_type_hints,
)
from openbb_core.app.charting_service import ChartingService
from openbb_core.app.provider_interface import ProviderInterface
from openbb_core.app.router import CommandMap
from pydantic.fields import FieldInfo
@@ -21,12 +33,15 @@ def test_{test_name}(params, obb):
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
{extra}
"""
def find_extensions():
def find_extensions(filter_chart: Optional[bool] = True):
"""Find extensions."""
filter_ext = ["tests", "charting", "__pycache__"]
filter_ext = ["tests", "__pycache__"]
if filter_chart:
filter_ext.append("charting")
extensions = [x for x in Path("openbb_platform/extensions").iterdir() if x.is_dir()]
extensions = [x for x in extensions if x.name not in filter_ext]
return extensions
@@ -146,7 +161,7 @@ def test_exists(command_name: str, path: str):
return command_name in f.read()
def write_to_file_w_template(test_file, params_list, full_command, test_name):
def write_to_file_w_template(test_file, params_list, full_command, test_name, extra=""):
"""Write to file with template."""
params = ""
for test_params in params_list:
@@ -159,6 +174,7 @@ def write_to_file_w_template(test_file, params_list, full_command, test_name):
test_name=test_name,
command_name=full_command,
params=params,
extra=extra,
)
)
@@ -245,12 +261,62 @@ def add_test_commands_to_file( # pylint: disable=W0102
)
def write_integration_test() -> None:
def write_commands_integration_tests() -> None:
"""Write integration test."""
extensions = find_extensions()
create_integration_test_files(extensions)
add_test_commands_to_file(extensions)
def write_charting_extension_integration_tests():
"""Write charting extension integration tests."""
import openbb_charting # pylint: disable=W0611
functions = ChartingService.get_implemented_charting_functions()
charting_ext_path = Path(openbb_charting.__file__).parent.parent
test_file = charting_ext_path / "integration" / "test_charting_python.py"
extra = """assert result.chart.content
assert isinstance(result.chart.fig, OpenBBFigure)"""
for func in functions:
write_to_file_w_template(
test_file=test_file,
params_list=[{"chart": "True"}],
full_command=func.replace("_", "."),
test_name=f"chart_{func}",
extra=extra,
)
if __name__ == "__main__":
write_integration_test()
parser = argparse.ArgumentParser(
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="API Integration Tests Generator",
description="Generate API integration tests",
)
parser.add_argument(
"--charting",
dest="charting",
default=True,
action="store_false",
help="Generate charting extension integration tests",
)
parser.add_argument(
"--commands",
dest="commands",
default=True,
action="store_false",
help="Generate commands integration tests",
)
args = parser.parse_args()
charting = args.charting
commands = args.commands
if commands:
write_commands_integration_tests()
if charting:
write_charting_extension_integration_tests()

18
openbb_platform/extensions/tests/utils/integration_tests_testers.py
View File

@@ -2,7 +2,17 @@
import importlib.util
import inspect
import os
from typing import Any, Callable, Dict, List, Literal, Tuple, Union, get_type_hints
from typing import (
Any,
Callable,
Dict,
List,
Literal,
Optional,
Tuple,
Union,
get_type_hints,
)
from openbb_core.app.provider_interface import ProviderInterface
from openbb_core.app.router import CommandMap
@@ -13,7 +23,9 @@ from extensions.tests.utils.integration_tests_generator import (
)
def get_integration_tests(test_type: Literal["api", "python"]) -> List[Any]:
def get_integration_tests(
test_type: Literal["api", "python"], filter_charting_ext: Optional[bool] = True
) -> List[Any]:
"""Get integration tests for the OpenBB Platform."""
integration_tests: List[Any] = []
@@ -22,7 +34,7 @@ def get_integration_tests(test_type: Literal["api", "python"]) -> List[Any]:
elif test_type == "api":
file_end = "_api.py"
for extension in find_extensions():
for extension in find_extensions(filter_charting_ext):
integration_folder = os.path.join(extension, "integration")
for file in os.listdir(integration_folder):
if file.endswith(file_end):