mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-11-05 00:42:19 +00:00
Feature/integration tests python data processing cmds (#5511)
* chaging the generator to handle data processing commands * ruff n black
This commit is contained in:
Henrique Joaquim
@@ -0,0 +1,209 @@
|
||||
"""Test econometrics extension."""
|
||||
import pytest
|
||||
from openbb_core.app.model.obbject import OBBject
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def obb(pytestconfig):
|
||||
"""Fixture to setup obb."""
|
||||
|
||||
if pytestconfig.getoption("markexpr") != "not integration":
|
||||
import openbb
|
||||
|
||||
return openbb.obb
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_corr(params, obb):
|
||||
result = obb.econometrics.corr(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_ols(params, obb):
|
||||
result = obb.econometrics.ols(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_ols_summary(params, obb):
|
||||
result = obb.econometrics.ols_summary(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_dwat(params, obb):
|
||||
result = obb.econometrics.dwat(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "lags": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_bgot(params, obb):
|
||||
result = obb.econometrics.bgot(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_coint(params, obb):
|
||||
result = obb.econometrics.coint(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_column": "", "lag": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_granger(params, obb):
|
||||
result = obb.econometrics.granger(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "column": "", "regression": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_unitroot(params, obb):
|
||||
result = obb.econometrics.unitroot(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_panelre(params, obb):
|
||||
result = obb.econometrics.panelre(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_panelbols(params, obb):
|
||||
result = obb.econometrics.panelbols(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_panelpols(params, obb):
|
||||
result = obb.econometrics.panelpols(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_panelols(params, obb):
|
||||
result = obb.econometrics.panelols(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_panelfd(params, obb):
|
||||
result = obb.econometrics.panelfd(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "y_column": "", "x_columns": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_econometrics_panelfmac(params, obb):
|
||||
result = obb.econometrics.panelfmac(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
170
openbb_platform/extensions/qa/integration/test_qa_python.py
Normal file
170
openbb_platform/extensions/qa/integration/test_qa_python.py
Normal file
@@ -0,0 +1,170 @@
|
||||
"""Test qa extension."""
|
||||
import pytest
|
||||
from openbb_core.app.model.obbject import OBBject
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def obb(pytestconfig):
|
||||
"""Fixture to setup obb."""
|
||||
|
||||
if pytestconfig.getoption("markexpr") != "not integration":
|
||||
import openbb
|
||||
|
||||
return openbb.obb
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_normality(params, obb):
|
||||
result = obb.qa.normality(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_capm(params, obb):
|
||||
result = obb.qa.capm(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"threshold_start": "",
|
||||
"threshold_end": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_om(params, obb):
|
||||
result = obb.qa.om(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "window": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_kurtosis(params, obb):
|
||||
result = obb.qa.kurtosis(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "fuller_reg": "", "kpss_reg": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_unitroot(params, obb):
|
||||
result = obb.qa.unitroot(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "rfr": "", "window": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_sh(params, obb):
|
||||
result = obb.qa.sh(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"target_return": "",
|
||||
"window": "",
|
||||
"adjusted": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_so(params, obb):
|
||||
result = obb.qa.so(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "window": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_skew(params, obb):
|
||||
result = obb.qa.skew(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "window": "", "quantile_pct": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_quantile(params, obb):
|
||||
result = obb.qa.quantile(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "target": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_qa_summary(params, obb):
|
||||
result = obb.qa.summary(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
527
openbb_platform/extensions/ta/integration/test_ta_python.py
Normal file
527
openbb_platform/extensions/ta/integration/test_ta_python.py
Normal file
@@ -0,0 +1,527 @@
|
||||
"""Test ta extension."""
|
||||
import pytest
|
||||
from openbb_core.app.model.obbject import OBBject
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def obb(pytestconfig):
|
||||
"""Fixture to setup obb."""
|
||||
|
||||
if pytestconfig.getoption("markexpr") != "not integration":
|
||||
import openbb
|
||||
|
||||
return openbb.obb
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"mamode": "",
|
||||
"drift": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_atr(params, obb):
|
||||
result = obb.ta.atr(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"close_column": "",
|
||||
"period": "",
|
||||
"start_date": "",
|
||||
"end_date": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_fib(params, obb):
|
||||
result = obb.ta.fib(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "offset": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_obv(params, obb):
|
||||
result = obb.ta.obv(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "length": "", "signal": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_fisher(params, obb):
|
||||
result = obb.ta.fisher(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "fast": "", "slow": "", "offset": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_adosc(params, obb):
|
||||
result = obb.ta.adosc(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"std": "",
|
||||
"mamode": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_bbands(params, obb):
|
||||
result = obb.ta.bbands(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_zlma(params, obb):
|
||||
result = obb.ta.zlma(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "length": "", "scalar": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_aroon(params, obb):
|
||||
result = obb.ta.aroon(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_sma(params, obb):
|
||||
result = obb.ta.sma(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"target": "",
|
||||
"show_all": "",
|
||||
"asint": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_demark(params, obb):
|
||||
result = obb.ta.demark(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "anchor": "", "offset": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_vwap(params, obb):
|
||||
result = obb.ta.vwap(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"fast": "",
|
||||
"slow": "",
|
||||
"signal": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_macd(params, obb):
|
||||
result = obb.ta.macd(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_hma(params, obb):
|
||||
result = obb.ta.hma(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"lower_length": "",
|
||||
"upper_length": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_donchian(params, obb):
|
||||
result = obb.ta.donchian(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"conversion": "",
|
||||
"base": "",
|
||||
"lagging": "",
|
||||
"offset": "",
|
||||
"lookahead": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_ichimoku(params, obb):
|
||||
result = obb.ta.ichimoku(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "target": "", "period": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_clenow(params, obb):
|
||||
result = obb.ta.clenow(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"scalar": "",
|
||||
"drift": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_adx(params, obb):
|
||||
result = obb.ta.adx(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_wma(params, obb):
|
||||
result = obb.ta.wma(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "length": "", "scalar": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_cci(params, obb):
|
||||
result = obb.ta.cci(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"scalar": "",
|
||||
"drift": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_rsi(params, obb):
|
||||
result = obb.ta.rsi(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"fast_k_period": "",
|
||||
"slow_d_period": "",
|
||||
"slow_k_period": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_stoch(params, obb):
|
||||
result = obb.ta.stoch(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"scalar": "",
|
||||
"mamode": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_kc(params, obb):
|
||||
result = obb.ta.kc(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
({"data": "", "index": "", "length": "", "return": ""}),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_cg(params, obb):
|
||||
result = obb.ta.cg(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"index": "",
|
||||
"lower_q": "",
|
||||
"upper_q": "",
|
||||
"model": "",
|
||||
"is_crypto": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_cones(params, obb):
|
||||
result = obb.ta.cones(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
(
|
||||
{
|
||||
"data": "",
|
||||
"target": "",
|
||||
"index": "",
|
||||
"length": "",
|
||||
"offset": "",
|
||||
"return": "",
|
||||
}
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_ta_ema(params, obb):
|
||||
result = obb.ta.ema(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
@@ -1,18 +1,30 @@
|
||||
"""Integration test generator."""
|
||||
from pathlib import Path, PosixPath
|
||||
from typing import Any, Dict, List, Literal, get_origin
|
||||
from typing import Any, Dict, List, Literal, Tuple, Type, get_origin, get_type_hints
|
||||
|
||||
from openbb_core.app.provider_interface import ProviderInterface
|
||||
from openbb_core.app.router import CommandMap
|
||||
from pydantic.fields import FieldInfo
|
||||
from pydantic_core import PydanticUndefined
|
||||
|
||||
cm = CommandMap(coverage_sep=".")
|
||||
TEST_TEMPLATE = """\n\n@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
{params}
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_{test_name}(params, obb):
|
||||
result = obb.{command_name}(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
"""
|
||||
|
||||
|
||||
def find_extensions():
|
||||
"""Find extensions."""
|
||||
filter_ext = ["tests", "ta", "qa", "econometrics", "charting"]
|
||||
filter_ext = ["tests", "charting"]
|
||||
extensions = [x for x in Path("openbb_platform/extensions").iterdir() if x.is_dir()]
|
||||
extensions = [x for x in extensions if x.name not in filter_ext]
|
||||
return extensions
|
||||
@@ -103,83 +115,131 @@ def get_test_params(
|
||||
return test_params_list
|
||||
|
||||
|
||||
def get_test_params_data_processing(hints: Dict[str, Type]):
|
||||
return list(hints.keys())
|
||||
|
||||
|
||||
def get_full_command_name_and_test_name(route: str) -> Tuple[str, str]:
|
||||
"""Get the full command name and test name."""
|
||||
|
||||
cmd_parts = route.split("/")
|
||||
del cmd_parts[0]
|
||||
|
||||
menu = cmd_parts[0]
|
||||
command = cmd_parts[-1]
|
||||
sub_menus = cmd_parts[1:-1]
|
||||
|
||||
sub_menus_str_test_name = f"_{'_'.join(sub_menus)}" if sub_menus else ""
|
||||
sub_menu_str_cmd = f".{'.'.join(sub_menus)}" if sub_menus else ""
|
||||
|
||||
full_command = f"{menu}{sub_menu_str_cmd}.{command}"
|
||||
test_name = f"{menu}{sub_menus_str_test_name}_{command}"
|
||||
|
||||
return test_name, full_command
|
||||
|
||||
|
||||
def test_exists(command_name: str, path: str):
|
||||
with open(path) as f:
|
||||
return command_name in f.read()
|
||||
|
||||
|
||||
def write_to_file_w_template(test_file, params_list, full_command, test_name):
|
||||
params = ""
|
||||
for test_params in params_list:
|
||||
params += f"({test_params}),\n"
|
||||
|
||||
if not test_exists(command_name=full_command, path=test_file):
|
||||
with open(test_file, "a", encoding="utf-8", newline="\n") as f:
|
||||
f.write(
|
||||
TEST_TEMPLATE.format(
|
||||
test_name=test_name,
|
||||
command_name=full_command,
|
||||
params=params,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def write_test(
|
||||
test_file: PosixPath,
|
||||
commands_model: Dict[str, str],
|
||||
extension_name: str,
|
||||
provider_interface_map: Dict[str, Any],
|
||||
):
|
||||
"""Write test."""
|
||||
|
||||
for route, model in commands_model.items():
|
||||
if extension_name in route and route.startswith(f"/{extension_name}/"):
|
||||
test_name, full_command = get_full_command_name_and_test_name(route=route)
|
||||
|
||||
test_params_list = get_test_params(
|
||||
model_name=model,
|
||||
provider_interface_map=provider_interface_map,
|
||||
)
|
||||
|
||||
write_to_file_w_template(
|
||||
test_file=test_file,
|
||||
params_list=test_params_list,
|
||||
full_command=full_command,
|
||||
test_name=test_name,
|
||||
)
|
||||
|
||||
|
||||
def write_test_data_processing(
|
||||
test_file: PosixPath, commands_map: Dict[str, str], extension_name: str
|
||||
):
|
||||
"""Write test for data processing commands."""
|
||||
|
||||
for route, _ in commands_map.items():
|
||||
if extension_name in route and route.startswith(f"/{extension_name}/"):
|
||||
test_name, full_command = get_full_command_name_and_test_name(route=route)
|
||||
|
||||
hints = get_type_hints(commands_map[route])
|
||||
test_params_list = [{k: "" for k in get_test_params_data_processing(hints)}]
|
||||
|
||||
write_to_file_w_template(
|
||||
test_file=test_file,
|
||||
params_list=test_params_list,
|
||||
full_command=full_command,
|
||||
test_name=test_name,
|
||||
)
|
||||
|
||||
|
||||
def add_test_commands_to_file( # pylint: disable=W0102
|
||||
extensions: List[PosixPath],
|
||||
commands_model: Dict[str, str] = cm.commands_model,
|
||||
) -> None:
|
||||
"""Add test commands to file."""
|
||||
|
||||
provider_interface = ProviderInterface()
|
||||
provider_interface_map = provider_interface.map
|
||||
|
||||
template = """\n\n@pytest.mark.parametrize(
|
||||
"params",
|
||||
[
|
||||
{params}
|
||||
],
|
||||
)
|
||||
@pytest.mark.integration
|
||||
def test_{test_name}(**params, obb):
|
||||
result = obb.{command_name}(**params)
|
||||
assert result
|
||||
assert isinstance(result, OBBject)
|
||||
assert len(result.results) > 0
|
||||
"""
|
||||
cm = CommandMap()
|
||||
commands_model = cm.commands_model
|
||||
commands_map = cm.map
|
||||
|
||||
extensions_names = [path.name for path in extensions]
|
||||
extensions_w_models = list({route.split("/")[1] for route in commands_model})
|
||||
extensions_data_processing = [
|
||||
ext for ext in extensions_names if ext not in extensions_w_models
|
||||
]
|
||||
|
||||
for extension in extensions:
|
||||
extension_name = extension.name
|
||||
test_file_name = f"test_{extension_name}_python.py"
|
||||
test_file = extension / "integration" / test_file_name
|
||||
with open(test_file, "a", encoding="utf-8", newline="\n") as f:
|
||||
for command in commands_model:
|
||||
if extension_name in command and command.startswith(
|
||||
f".{extension_name}."
|
||||
):
|
||||
extension_name = command.split(".")[1]
|
||||
|
||||
if len(command.split(".")) > 3:
|
||||
test_name = (
|
||||
extension_name
|
||||
+ "_"
|
||||
+ command.split(".")[2]
|
||||
+ "_"
|
||||
+ command.split(".")[3]
|
||||
)
|
||||
elif len(command.split(".")) == 3:
|
||||
test_name = extension_name + "_" + command.split(".")[2]
|
||||
|
||||
command_name = command.split(".")[-1]
|
||||
if "_" in command_name:
|
||||
full_command_name = test_name.replace("_", ".", 2)
|
||||
# The code below double checks if the full command name is correct.
|
||||
# This eliminates edge cases.
|
||||
fix_full_command_name = full_command_name.split(".")
|
||||
for i, part in enumerate(fix_full_command_name):
|
||||
if part == command_name.split("_")[0]:
|
||||
fix_full_command_name[i] = command_name
|
||||
fix_full_command_name.pop()
|
||||
|
||||
full_command_name = ".".join(fix_full_command_name)
|
||||
|
||||
else:
|
||||
full_command_name = test_name.replace("_", ".")
|
||||
|
||||
test_params_list = get_test_params(
|
||||
model_name=commands_model[command],
|
||||
provider_interface_map=provider_interface_map,
|
||||
)
|
||||
|
||||
params = ""
|
||||
for test_params in test_params_list:
|
||||
params += f"({test_params}),\n"
|
||||
|
||||
read_file = open(test_file) # noqa
|
||||
if full_command_name not in read_file.read():
|
||||
f.write(
|
||||
template.format(
|
||||
test_name=test_name,
|
||||
command_name=full_command_name,
|
||||
params=params,
|
||||
)
|
||||
)
|
||||
if extension_name in extensions_data_processing:
|
||||
write_test_data_processing(
|
||||
test_file=test_file,
|
||||
commands_map=commands_map,
|
||||
extension_name=extension_name,
|
||||
)
|
||||
else:
|
||||
write_test(
|
||||
test_file=test_file,
|
||||
commands_model=commands_model,
|
||||
extension_name=extension_name,
|
||||
provider_interface_map=provider_interface_map,
|
||||
)
|
||||
|
||||
|
||||
def write_integration_test() -> None:
|
||||
|
||||
Reference in New Issue
Block a user