mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-27 04:31:13 +00:00
9eca7b9f23
* improve cftc endpoints and make a standalone workspace app * mypy * forgot couple of files to commit * test params * integration test param name * fix sec-sic-search * fill some contracts missing category and subcategory, add a workspace workchoices endpoint * linter * integration test for cot_choices
66 lines
1.7 KiB
Python
66 lines
1.7 KiB
Python
"""CFTC Fetcher Tests."""
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
from openbb_cftc.models.cot import CftcCotFetcher
|
|
from openbb_cftc.models.cot_search import CftcCotSearchFetcher
|
|
from openbb_core.app.service.user_service import UserService
|
|
|
|
test_credentials = UserService().default_user_settings.credentials.model_dump(
|
|
mode="json"
|
|
)
|
|
|
|
|
|
def scrub_string(key):
|
|
"""Scrub a string from the response."""
|
|
|
|
def before_record_response(response):
|
|
response["headers"][key] = response["headers"].update({key: "MOCK_VALUE"})
|
|
return response
|
|
|
|
return before_record_response
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def vcr_config():
|
|
"""VCR configuration."""
|
|
return {
|
|
"filter_headers": [("User-Agent", None)],
|
|
"filter_query_parameters": [
|
|
("$$app_token", "MOCK_APP_TOKEN"),
|
|
("$limit", "MOCK_LIMIT"),
|
|
("$order", "MOCK_ORDER"),
|
|
("$where", "MOCK_WHERE"),
|
|
],
|
|
"before_record_response": [
|
|
scrub_string("Etag"),
|
|
scrub_string("X-Socrata-RequestId"),
|
|
scrub_string("X-Socrata-Region"),
|
|
],
|
|
}
|
|
|
|
|
|
@pytest.mark.record_http
|
|
def test_cftc_cot_fetcher(credentials=test_credentials):
|
|
"""Test the CFTC COT fetcher."""
|
|
params = {
|
|
"code": "239747",
|
|
"start_date": date(2024, 8, 19),
|
|
"end_date": date(2024, 8, 21),
|
|
}
|
|
|
|
fetcher = CftcCotFetcher()
|
|
result = fetcher.test(params, credentials)
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.record_http
|
|
def test_cftc_cot_sarch_fetcher(credentials=test_credentials):
|
|
"""Test the CFTC COT Search fetcher."""
|
|
params = {"query": "S&P 500"}
|
|
|
|
fetcher = CftcCotSearchFetcher()
|
|
result = fetcher.test(params, credentials)
|
|
assert result is None
|