File size: 3,239 Bytes
a2afe2f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 |
"""Test Commodity API endpoints."""
import base64
import pytest
import requests
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')}"}
@pytest.mark.parametrize(
"params",
[
(
{
"commodity": "all",
"start_date": None,
"end_date": None,
"frequency": None,
"transform": None,
"aggregation_method": None,
"provider": "fred",
}
),
],
)
@pytest.mark.integration
def test_commodity_price_spot(params, headers):
"""Test the commodity spot prices 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/commodity/price/spot?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"category": "balance_sheet",
"table": "stocks",
"start_date": None,
"end_date": None,
"provider": "eia",
"use_cache": True,
}
),
(
{
"category": "weekly_estimates",
"table": "crude_production",
"start_date": "2020-01-01",
"end_date": "2023-12-31",
"provider": "eia",
"use_cache": True,
}
),
],
)
@pytest.mark.integration
def test_commodity_petroleum_status_report(params, headers):
"""Test the Petroleum Status Report 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/commodity/petroleum_status_report?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"table": "01",
"symbol": None,
"start_date": "2024-09-01",
"end_date": "2024-10-01",
"provider": "eia",
"frequency": "month",
}
),
],
)
@pytest.mark.integration
def test_commodity_short_term_energy_outlook(params, headers):
"""Test the Short Term Energy Outlook 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/commodity/short_term_energy_outlook?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
|