-
Notifications
You must be signed in to change notification settings - Fork 3.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Charting extension integration tests (#5547)
* 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()
- Loading branch information
Showing
8 changed files
with
1,049 additions
and
74 deletions.
There are no files selected for viewing
375 changes: 375 additions & 0 deletions
375
openbb_platform/extensions/charting/integration/test_charting_api.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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"] |
Oops, something went wrong.