"""Test fixedincome API endpoints."""
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
# pylint: disable=redefined-outer-name
@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')}"}
@parametrize(
"params",
[({"start_date": "2023-01-01", "end_date": "2023-06-06", "provider": "fmp"})],
)
@pytest.mark.integration
def test_fixedincome_government_treasury_rates(params, headers):
"""Test the treasury rates 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/fixedincome/government/treasury_rates?{query_str}"
)
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[({"date": "2023-01-01", "inflation_adjusted": True, "provider": "fred"})],
)
@pytest.mark.integration
def test_fixedincome_government_us_yield_curve(params, headers):
"""Test the US yield curve 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/fixedincome/government/us_yield_curve?{query_str}"
)
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"provider": "federal_reserve",
}
),
(
{
"frequency": None,
"transform": None,
"aggregation_method": None,
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
@pytest.mark.integration
def test_fixedincome_sofr(params, headers):
"""Test the SOFR 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/fixedincome/sofr?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"provider": "federal_reserve",
}
),
(
{
"frequency": None,
"transform": None,
"aggregation_method": None,
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
@pytest.mark.integration
def test_fixedincome_rate_sofr(params, headers):
"""Test the SOFR 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/fixedincome/rate/sofr?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
({"start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"parameter": "volume_weighted_trimmed_mean_rate",
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
@pytest.mark.integration
def test_fix