diff --git a/openbb_platform/extensions/charting/integration/test_charting_api.py b/openbb_platform/extensions/charting/integration/test_charting_api.py new file mode 100644 index 000000000000..7e0815f67e0b --- /dev/null +++ b/openbb_platform/extensions/charting/integration/test_charting_api.py @@ -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"] diff --git a/openbb_platform/extensions/charting/integration/test_charting_python.py b/openbb_platform/extensions/charting/integration/test_charting_python.py new file mode 100644 index 000000000000..a6e7a982ad50 --- /dev/null +++ b/openbb_platform/extensions/charting/integration/test_charting_python.py @@ -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) diff --git a/openbb_platform/extensions/ta/integration/test_ta_api.py b/openbb_platform/extensions/ta/integration/test_ta_api.py index 68fe9f143af4..1da1d5f05ce9 100644 --- a/openbb_platform/extensions/ta/integration/test_ta_api.py +++ b/openbb_platform/extensions/ta/integration/test_ta_api.py @@ -12,6 +12,7 @@ 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 diff --git a/openbb_platform/extensions/tests/test_integration_tests_api.py b/openbb_platform/extensions/tests/test_integration_tests_api.py index 6aadd20ae77e..b85d0892bf28 100644 --- a/openbb_platform/extensions/tests/test_integration_tests_api.py +++ b/openbb_platform/extensions/tests/test_integration_tests_api.py @@ -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) diff --git a/openbb_platform/extensions/tests/test_integration_tests_python.py b/openbb_platform/extensions/tests/test_integration_tests_python.py index 0575ebdc64e4..c850ba9059a4 100644 --- a/openbb_platform/extensions/tests/test_integration_tests_python.py +++ b/openbb_platform/extensions/tests/test_integration_tests_python.py @@ -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) diff --git a/openbb_platform/extensions/tests/utils/integration_tests_api_generator.py b/openbb_platform/extensions/tests/utils/integration_tests_api_generator.py index d77df54d2b53..09b7638abf12 100644 --- a/openbb_platform/extensions/tests/utils/integration_tests_api_generator.py +++ b/openbb_platform/extensions/tests/utils/integration_tests_api_generator.py @@ -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 @@ 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 @@ def write_integration_tests( 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() diff --git a/openbb_platform/extensions/tests/utils/integration_tests_generator.py b/openbb_platform/extensions/tests/utils/integration_tests_generator.py index 2d4ecc6200e6..38ce33810347 100644 --- a/openbb_platform/extensions/tests/utils/integration_tests_generator.py +++ b/openbb_platform/extensions/tests/utils/integration_tests_generator.py @@ -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() diff --git a/openbb_platform/extensions/tests/utils/integration_tests_testers.py b/openbb_platform/extensions/tests/utils/integration_tests_testers.py index ced6ad7c7252..76bab8b14086 100644 --- a/openbb_platform/extensions/tests/utils/integration_tests_testers.py +++ b/openbb_platform/extensions/tests/utils/integration_tests_testers.py @@ -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 @@ ) -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):