jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-11-02 07:23:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

add integration tests

Browse Source
This commit is contained in:
Danglewood
2024-11-14 22:50:04 -08:00
parent 31b05c0c15
commit b10d64034a
4 changed files with 772 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
openbb_platform/extensions/websockets/README.md vendored
View File

@@ -607,8 +607,14 @@ class FmpWebSocketData(WebSocketData):
#### WebSocketConnection
This model is what we return from the `FmpWebSocketFetcher`.
This model is what we return from the `FmpWebSocketFetcher`. The provider will inherit two fields, only the `client` needs to be included on output. No additional fields should be defined.
- client
- Instance of WebSocketClient, not returned to from the Router.
- status
- Leave empty, the Router fills this and returns it.
The model will not accept additional fields.
```python
class FmpWebSocketConnection(WebSocketConnection):

376
openbb_platform/extensions/websockets/integration/test_websockets_api.py Normal file
View File

@@ -0,0 +1,376 @@
"""Test WebSockets API Integration."""
import base64
import pytest
import requests
from extensions.tests.conftest import parametrize
from openbb_core.env import Env
from openbb_core.provider.utils.helpers import get_querystring
@pytest.fixture(scope="session")
def headers():
"""Get the headers for the API request."""
userpass = f"{Env().API_USERNAME}:{Env().API_PASSWORD}"
userpass_bytes = userpass.encode("ascii")
base64_bytes = base64.b64encode(userpass_bytes)
return {"Authorization": f"Basic {base64_bytes.decode('ascii')}"}
# pylint: disable=redefined-outer-name
@parametrize(
"params",
[
(
{
"name": "test_fmp",
"provider": "fmp",
"symbol": "btcusd,dogeusd",
"asset_type": "crypto",
"auth_token": None,
"results_file": None,
"save_results": False,
"table_name": "records",
"limit": 10,
"sleep_time": 0.25,
"broadcast_host": "localhost",
"broadcast_port": 6666,
"start_broadcast": False,
"connect_kwargs": None,
}
),
(
{
"name": "test_tiingo",
"provider": "tiingo",
"symbol": "btcusd,dogeusd",
"asset_type": "crypto",
"feed": "trade_and_quote",
"auth_token": None,
"results_file": None,
"save_results": False,
"table_name": "records",
"limit": 10,
"sleep_time": 0.25,
"broadcast_host": "localhost",
"broadcast_port": 6666,
"start_broadcast": False,
"connect_kwargs": None,
}
),
(
{
"name": "test_polygon",
"provider": "polygon",
"symbol": "btcusd,dogeusd",
"asset_type": "crypto",
"feed": "quote",
"auth_token": None,
"results_file": None,
"save_results": False,
"table_name": "records",
"limit": 10,
"sleep_time": 0.25,
"broadcast_host": "localhost",
"broadcast_port": 6666,
"start_broadcast": False,
"connect_kwargs": None,
}
),
],
)
@pytest.mark.integration
def test_websockets_create_connection(params, headers):
"""Test the websockets_create_connection endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/create_connection?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
res = result.json()["results"]
assert isinstance(res, dict)
assert res.get("status", {}).get("is_running")
assert not res.get("status", {}).get("is_broadcasting")
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_get_results(params, headers):
"""Test the websockets_get_results endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/get_results?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_clear_results(params, headers):
"""Test the websockets_clear_results endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/clear_results?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
{
"name": "test_fmp",
"symbol": "ethusd",
"auth_token": None,
},
{
"name": "test_tiingo",
"symbol": "ethusd",
"auth_token": None,
},
{
"name": "test_polygon",
"symbol": "ethusd",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_subscribe(params, headers):
"""Test the websockets_subscribe endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/subscribe?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
"host": None,
"port": None,
"uvicorn_kwargs": None,
},
{
"name": "test_tiingo",
"auth_token": None,
"host": None,
"port": None,
"uvicorn_kwargs": None,
},
{
"name": "test_polygon",
"auth_token": None,
"host": None,
"port": None,
"uvicorn_kwargs": None,
},
],
)
@pytest.mark.integration
def test_websockets_start_broadcasting(params, headers):
"""Test the websockets_start_broadcasting endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/start_broadcasting?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
"symbol": "ethusd",
},
{
"name": "test_tiingo",
"auth_token": None,
"symbol": "ethusd",
},
{
"name": "test_polygon",
"auth_token": None,
"symbol": "ethusd",
},
],
)
@pytest.mark.integration
def test_websockets_unsubscribe(params, headers):
"""Test the websockets_unsubscribe endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/unsubscribe?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_stop_connection(params, headers):
"""Test the websockets_stop_connection endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/stop_connection?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_restart_connection(params, headers):
"""Test the websockets_restart_connection endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/restart_connection?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_stop_broadcasting(params, headers):
"""Test the websockets_stop_broadcasting endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/stop_broadcasting?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_kill(params, headers):
"""Test the websockets_kill endpoint."""
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/websockets/kill?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200

388
openbb_platform/extensions/websockets/integration/test_websockets_python.py Normal file
View File

@@ -0,0 +1,388 @@
"""Test WebSockets Python Integration."""
# pylint: disable=redefined-outer-name, inconsistent-return-statements, import-outside-toplevel
import pytest
from extensions.tests.conftest import parametrize
from openbb_core.app.model.obbject import OBBject
from openbb_websockets.models import WebSocketConnectionStatus
@pytest.fixture(scope="session")
def obb(pytestconfig):
"""Fixture to setup obb."""
if pytestconfig.getoption("markexpr") != "not integration":
import openbb
return openbb.obb
@parametrize(
"params",
[
(
{
"name": "test_fmp",
"provider": "fmp",
"symbol": "btcusd,dogeusd",
"asset_type": "crypto",
"auth_token": None,
"results_file": None,
"save_results": False,
"table_name": "records",
"limit": 10,
"sleep_time": 0.25,
"broadcast_host": "localhost",
"broadcast_port": 6666,
"start_broadcast": False,
"connect_kwargs": None,
}
),
(
{
"name": "test_tiingo",
"provider": "tiingo",
"symbol": "btcusd,dogeusd",
"asset_type": "crypto",
"feed": "trade_and_quote",
"auth_token": None,
"results_file": None,
"save_results": False,
"table_name": "records",
"limit": 10,
"sleep_time": 0.25,
"broadcast_host": "localhost",
"broadcast_port": 6666,
"start_broadcast": False,
"connect_kwargs": None,
}
),
(
{
"name": "test_polygon",
"provider": "polygon",
"symbol": "btcusd,dogeusd",
"asset_type": "crypto",
"feed": "quote",
"auth_token": None,
"results_file": None,
"save_results": False,
"table_name": "records",
"limit": 10,
"sleep_time": 0.25,
"broadcast_host": "localhost",
"broadcast_port": 6666,
"start_broadcast": False,
"connect_kwargs": None,
}
),
],
)
@pytest.mark.integration
def test_websockets_create_connection(params, obb):
"""Test the websockets_create_connection endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.create_connection(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
assert isinstance(result.results.status, WebSocketConnectionStatus)
assert result.results.status.is_running is True
assert result.results.status.is_broadcasting is False
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_get_results(params, obb):
"""Test the websockets_get_results endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.get_results(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_clear_results(params, obb):
"""Test the websockets_clear_results endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.clear_results(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"symbol": "ethusd",
"auth_token": None,
},
{
"name": "test_tiingo",
"symbol": "ethusd",
"auth_token": None,
},
{
"name": "test_polygon",
"symbol": "ethusd",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_subscribe(params, obb):
"""Test the websockets_subscribe endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.subscribe(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
"host": None,
"port": None,
"uvicorn_kwargs": None,
},
{
"name": "test_tiingo",
"auth_token": None,
"host": None,
"port": None,
"uvicorn_kwargs": None,
},
{
"name": "test_polygon",
"auth_token": None,
"host": None,
"port": None,
"uvicorn_kwargs": None,
},
],
)
@pytest.mark.integration
def test_websockets_start_broadcasting(params, obb):
"""Test the websockets_start_broadcasting endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.start_broadcasting(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"symbol": "ethusd",
"auth_token": None,
},
{
"name": "test_tiingo",
"symbol": "ethusd",
"auth_token": None,
},
{
"name": "test_polygon",
"symbol": "ethusd",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_unsubscribe(params, obb):
"""Test the websockets_unsubscribe endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.unsubscribe(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_get_client(params, obb):
"""Test the websockets_get_client endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.get_client(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_stop_connection(params, obb):
"""Test the websockets_stop_connection endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.stop_connection(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_restart_connection(params, obb):
"""Test the websockets_restart_connection endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.restart_connection(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_stop_broadcasting(params, obb):
"""Test the websockets_stop_broadcasting endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.stop_broadcasting(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None
@parametrize(
"params",
[
{
"name": "test_fmp",
"auth_token": None,
},
{
"name": "test_tiingo",
"auth_token": None,
},
{
"name": "test_polygon",
"auth_token": None,
},
],
)
@pytest.mark.integration
def test_websockets_kill(params, obb):
"""Test the websockets_kill endpoint."""
params = {p: v for p, v in params.items() if v}
result = obb.websockets.kill(**params)
assert result
assert isinstance(result, OBBject)
assert result.results is not None

1
openbb_platform/extensions/websockets/tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""WebSockets Extension Tests."""