mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 21:12:16 +00:00
869cfed125
* merge branch develop with main * bump desktop version * fix the merge * sync Cargo.lock to desktop 2.0.0 * fix cli conftest.py --------- Co-authored-by: deeleeramone <>
292 lines
9.8 KiB
Python
292 lines
9.8 KiB
Python
"""Tests for the FRED request builders and typed endpoint access."""
|
|
|
|
import pytest
|
|
from openbb_core.app.model.abstract.error import OpenBBError
|
|
|
|
from openbb_fred.utils.api import (
|
|
GEO_ROOT_URL,
|
|
ROOT_URL,
|
|
build_url,
|
|
get_observations,
|
|
get_observations_many,
|
|
headers,
|
|
observation_dates,
|
|
observations_url,
|
|
release_tables_url,
|
|
unwrap_series,
|
|
)
|
|
|
|
|
|
def _params(url: str) -> dict:
|
|
"""Return the query parameters of a URL."""
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
parsed = parse_qs(urlsplit(url).query, keep_blank_values=True)
|
|
|
|
return {k: v[0] for k, v in parsed.items()}
|
|
|
|
|
|
class TestBuildUrl:
|
|
"""The single constructor every FRED request URL is built by."""
|
|
|
|
def test_the_key_and_format_are_always_sent(self):
|
|
url = build_url("series", "abc")
|
|
|
|
assert url.startswith(f"{ROOT_URL}/series?")
|
|
assert _params(url) == {"api_key": "abc", "file_type": "json"}
|
|
|
|
def test_a_missing_key_is_sent_empty(self):
|
|
assert _params(build_url("series", None))["api_key"] == ""
|
|
|
|
def test_a_leading_slash_does_not_double(self):
|
|
assert build_url("/series", "abc").startswith(f"{ROOT_URL}/series?")
|
|
|
|
def test_an_absent_parameter_is_left_out(self):
|
|
assert "element_id" not in _params(build_url("series", "abc", element_id=None))
|
|
|
|
def test_a_value_is_escaped(self):
|
|
url = build_url("series/search", "abc", search_text="real gdp")
|
|
|
|
assert "real%20gdp" in url or "real+gdp" in url
|
|
assert _params(url)["search_text"] == "real gdp"
|
|
|
|
def test_another_root_is_addressable(self):
|
|
url = build_url("series/data", "abc", root=GEO_ROOT_URL)
|
|
|
|
assert url.startswith(f"{GEO_ROOT_URL}/series/data?")
|
|
|
|
|
|
class TestObservationsUrl:
|
|
"""The observations request for one series."""
|
|
|
|
def test_the_series_is_named(self):
|
|
assert _params(observations_url("IORB", "abc"))["series_id"] == "IORB"
|
|
|
|
def test_the_dates_become_the_observation_window(self):
|
|
from datetime import date
|
|
|
|
sent = _params(
|
|
observations_url("GDP", "abc", date(2024, 1, 1), date(2024, 6, 1))
|
|
)
|
|
|
|
assert sent["observation_start"] == "2024-01-01"
|
|
assert sent["observation_end"] == "2024-06-01"
|
|
|
|
def test_a_date_already_written_is_kept(self):
|
|
sent = _params(observations_url("GDP", "abc", "2024-01-01"))
|
|
|
|
assert sent["observation_start"] == "2024-01-01"
|
|
|
|
def test_no_window_sends_no_window(self):
|
|
sent = _params(observations_url("GDP", "abc"))
|
|
|
|
assert "observation_start" not in sent
|
|
assert "observation_end" not in sent
|
|
|
|
def test_the_cache_switch_never_reaches_the_api(self):
|
|
sent = _params(observations_url("GDP", "abc", use_cache=False))
|
|
|
|
assert "use_cache" not in sent
|
|
|
|
def test_the_preferences_never_reach_the_api(self):
|
|
sent = _params(observations_url("GDP", "abc", preferences={"a": 1}))
|
|
|
|
assert "preferences" not in sent
|
|
|
|
|
|
class TestReleaseTablesUrl:
|
|
"""The release tables request the four release models share."""
|
|
|
|
def test_the_release_and_element_are_named(self):
|
|
sent = _params(release_tables_url("50", "12345", "abc"))
|
|
|
|
assert sent["release_id"] == "50"
|
|
assert sent["element_id"] == "12345"
|
|
assert sent["include_observation_values"] == "true"
|
|
|
|
def test_a_release_without_an_element_is_read_whole(self):
|
|
assert "element_id" not in _params(release_tables_url("50", None, "abc"))
|
|
|
|
def test_a_period_is_carried(self):
|
|
sent = _params(release_tables_url("50", "1", "abc", "2024-01-01"))
|
|
|
|
assert sent["observation_date"] == "2024-01-01"
|
|
|
|
def test_no_period_asks_for_the_latest(self):
|
|
assert "observation_date" not in _params(release_tables_url("50", "1", "abc"))
|
|
|
|
|
|
class TestObservationDates:
|
|
"""The periods a release table is asked for."""
|
|
|
|
def test_nothing_asked_for_reads_the_latest(self):
|
|
assert observation_dates(None) == [None]
|
|
|
|
def test_a_date_becomes_its_period_start(self):
|
|
assert observation_dates("2024-03-17") == ["2024-03-01"]
|
|
|
|
def test_a_repeated_month_is_read_once(self):
|
|
assert observation_dates("2024-03-17,2024-03-02") == ["2024-03-01"]
|
|
|
|
def test_the_periods_run_oldest_first(self):
|
|
assert observation_dates("2024-05-02,2024-01-09") == [
|
|
"2024-01-01",
|
|
"2024-05-01",
|
|
]
|
|
|
|
def test_a_repeated_pair_does_not_corrupt_the_year(self):
|
|
"""'2011-11-11' must not become '2001-01-01'."""
|
|
assert observation_dates("2011-11-11") == ["2011-11-01"]
|
|
|
|
def test_a_date_object_is_read(self):
|
|
from datetime import date
|
|
|
|
assert observation_dates(date(2011, 11, 11)) == ["2011-11-01"]
|
|
|
|
|
|
class TestHeaders:
|
|
"""The headers a FRED request carries."""
|
|
|
|
def test_a_request_declares_its_own_agent(self):
|
|
sent = headers()
|
|
|
|
assert sent["Accept"] == "application/json"
|
|
assert sent["Accept-Encoding"] == "gzip, deflate"
|
|
assert sent["User-Agent"]
|
|
|
|
def test_no_key_sends_no_authorization(self):
|
|
assert "Authorization" not in headers()
|
|
|
|
def test_a_key_travels_as_a_bearer_token(self):
|
|
assert headers("abc")["Authorization"] == "Bearer abc"
|
|
|
|
def test_an_empty_key_still_declares_the_scheme(self):
|
|
assert headers("")["Authorization"] == "Bearer "
|
|
|
|
|
|
class TestGetObservations:
|
|
"""Reading one series through the transport."""
|
|
|
|
async def test_the_observations_are_returned(self, monkeypatch):
|
|
async def answer(url, use_cache=True, **kwargs):
|
|
return {"observations": [{"date": "2024-01-01", "value": "1.0"}]}
|
|
|
|
monkeypatch.setattr("openbb_fred.utils.rate_limiter.fred_get", answer)
|
|
|
|
assert await get_observations("GDP", "abc") == [
|
|
{"date": "2024-01-01", "value": "1.0"}
|
|
]
|
|
|
|
async def test_an_empty_response_is_reported(self, monkeypatch):
|
|
async def answer(url, use_cache=True, **kwargs):
|
|
return {}
|
|
|
|
monkeypatch.setattr("openbb_fred.utils.rate_limiter.fred_get", answer)
|
|
|
|
with pytest.raises(OpenBBError, match="No data exists for series id: GDP"):
|
|
await get_observations("GDP", "abc")
|
|
|
|
async def test_an_error_message_is_reported(self, monkeypatch):
|
|
async def answer(url, use_cache=True, **kwargs):
|
|
return {"error_message": "Bad Request."}
|
|
|
|
monkeypatch.setattr("openbb_fred.utils.rate_limiter.fred_get", answer)
|
|
|
|
with pytest.raises(OpenBBError, match="Bad Request."):
|
|
await get_observations("GDP", "abc")
|
|
|
|
async def test_the_cache_switch_reaches_the_transport(self, monkeypatch):
|
|
seen: list = []
|
|
|
|
async def answer(url, use_cache=True, **kwargs):
|
|
seen.append(use_cache)
|
|
return {"observations": []}
|
|
|
|
monkeypatch.setattr("openbb_fred.utils.rate_limiter.fred_get", answer)
|
|
await get_observations("GDP", "abc", use_cache=False)
|
|
|
|
assert seen == [False]
|
|
|
|
|
|
class TestGetObservationsMany:
|
|
"""Reading many series under one rate limit."""
|
|
|
|
async def test_the_results_line_up_with_the_series(self, monkeypatch):
|
|
async def answer(urls, use_cache=True, **kwargs):
|
|
return [
|
|
{"observations": [{"series": url.split("series_id=")[1][:4]}]}
|
|
for url in urls
|
|
]
|
|
|
|
monkeypatch.setattr("openbb_fred.utils.rate_limiter.fred_get_many", answer)
|
|
read = await get_observations_many(["AAAA", "BBBB", "CCCC"], "abc")
|
|
|
|
assert [rows[0]["series"] for rows in read] == ["AAAA", "BBBB", "CCCC"]
|
|
|
|
async def test_a_series_without_data_is_reported(self, monkeypatch):
|
|
async def answer(urls, use_cache=True, **kwargs):
|
|
return [{"observations": [{"date": "2024-01-01"}]}, {}]
|
|
|
|
monkeypatch.setattr("openbb_fred.utils.rate_limiter.fred_get_many", answer)
|
|
|
|
with pytest.raises(OpenBBError, match="No data exists for series id: BBBB"):
|
|
await get_observations_many(["AAAA", "BBBB"], "abc")
|
|
|
|
async def test_nothing_asked_for_reads_nothing(self, monkeypatch):
|
|
asked: list = []
|
|
|
|
async def answer(urls, use_cache=True, **kwargs):
|
|
asked.append(list(urls))
|
|
return []
|
|
|
|
monkeypatch.setattr("openbb_fred.utils.rate_limiter.fred_get_many", answer)
|
|
|
|
assert await get_observations_many([], "abc") == []
|
|
assert asked == [[]]
|
|
|
|
|
|
class TestPublishedValue:
|
|
"""Reading one observation as a number."""
|
|
|
|
def test_a_number_is_read(self):
|
|
from openbb_fred.utils.api import published_value
|
|
|
|
assert published_value("4.35") == 4.35
|
|
|
|
@pytest.mark.parametrize("missing", [".", "", None])
|
|
def test_a_date_with_nothing_published_reads_as_nothing(self, missing):
|
|
from openbb_fred.utils.api import published_value
|
|
|
|
assert published_value(missing) is None
|
|
|
|
def test_a_value_that_is_not_a_number_reads_as_nothing(self):
|
|
from openbb_fred.utils.api import published_value
|
|
|
|
assert published_value([]) is None
|
|
assert published_value("n/a") is None
|
|
|
|
|
|
class TestUnwrapSeries:
|
|
"""Splitting a series fetch into its rows and metadata."""
|
|
|
|
def test_an_annotated_result_gives_up_both_halves(self):
|
|
from openbb_core.provider.abstract.annotated_result import AnnotatedResult
|
|
|
|
rows, metadata = unwrap_series(
|
|
AnnotatedResult(result=[1, 2], metadata={"GDP": {"title": "GDP"}})
|
|
)
|
|
|
|
assert rows == [1, 2]
|
|
assert metadata == {"GDP": {"title": "GDP"}}
|
|
|
|
def test_rows_on_their_own_carry_no_metadata(self):
|
|
assert unwrap_series([1, 2]) == ([1, 2], {})
|
|
|
|
def test_nothing_reads_as_empty(self):
|
|
assert unwrap_series(None) == ([], {})
|
|
|
|
def test_an_annotated_result_without_metadata_reads_as_empty(self):
|
|
from openbb_core.provider.abstract.annotated_result import AnnotatedResult
|
|
|
|
assert unwrap_series(AnnotatedResult(result=[1], metadata=None)) == ([1], {})
|