mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 21:12:16 +00:00
94c82cbd7c
* update FOMC Documents app * tests * fix error when integration test function has no params
364 lines
14 KiB
Python
364 lines
14 KiB
Python
"""Test if there are any missing providers for python interface integration tests."""
|
|
|
|
import importlib.util
|
|
import inspect
|
|
import os
|
|
from typing import (
|
|
Any,
|
|
Callable,
|
|
Dict,
|
|
List,
|
|
Literal,
|
|
Optional,
|
|
Tuple,
|
|
Union,
|
|
get_type_hints,
|
|
)
|
|
|
|
from openbb_core.app.provider_interface import ProviderInterface
|
|
from openbb_core.app.router import CommandMap
|
|
|
|
from extensions.tests.utils.integration_tests_generator import (
|
|
find_extensions,
|
|
get_test_params_data_processing,
|
|
)
|
|
|
|
|
|
def get_integration_tests(
|
|
test_type: Literal["api", "python"], filter_charting_ext: Optional[bool] = True
|
|
) -> List[Any]:
|
|
"""Get integration tests for the OpenBB Platform."""
|
|
integration_tests: List[Any] = []
|
|
|
|
if test_type == "python":
|
|
file_end = "_python.py"
|
|
elif test_type == "api":
|
|
file_end = "_api.py"
|
|
else:
|
|
raise ValueError(f"test_type '{test_type}' not valid")
|
|
|
|
for extension in find_extensions(filter_charting_ext):
|
|
integration_folder = os.path.join(extension, "integration")
|
|
for file in os.listdir(integration_folder):
|
|
if file.endswith(file_end):
|
|
file_path = os.path.join(integration_folder, file)
|
|
module_name = file[:-3] # Remove .py from file name
|
|
|
|
spec = importlib.util.spec_from_file_location(module_name, file_path)
|
|
if spec:
|
|
module = importlib.util.module_from_spec(spec)
|
|
if spec.loader:
|
|
spec.loader.exec_module(module)
|
|
integration_tests.append(module)
|
|
|
|
return integration_tests
|
|
|
|
|
|
def get_module_functions(module_list: List[Any]) -> Dict[str, Any]:
|
|
"""Get all functions from a list of modules."""
|
|
functions = {}
|
|
for module in module_list:
|
|
for name, obj in inspect.getmembers(module):
|
|
if inspect.isfunction(obj):
|
|
functions[name] = obj
|
|
return functions
|
|
|
|
|
|
def check_missing_providers(
|
|
command_params: Union[Dict[str, Dict[str, dict]], List[Tuple[Dict[str, str], str]]],
|
|
function_params: List[dict],
|
|
function,
|
|
processing: bool = False,
|
|
) -> List[str]:
|
|
"""Check if there are any missing providers for a command."""
|
|
if processing or not isinstance(command_params, dict):
|
|
return []
|
|
|
|
missing_providers: List[str] = []
|
|
providers = list(command_params.keys())
|
|
providers.remove("openbb")
|
|
|
|
for test_params in function_params:
|
|
provider = test_params.get("provider", None)
|
|
if provider:
|
|
try: # noqa
|
|
providers.remove(provider)
|
|
except ValueError:
|
|
pass
|
|
|
|
if providers:
|
|
# if there is only one provider left and the length of the
|
|
# test_params is 1, we can ignore because it is picked up by default
|
|
if len(providers) == 1 and len(function_params) == 1:
|
|
pass
|
|
else:
|
|
missing_providers.append(f"Missing providers for {function}: {providers}")
|
|
|
|
return missing_providers
|
|
|
|
|
|
def check_wrong_params(
|
|
command_params: Union[Dict[str, Dict[str, dict]], List[Tuple[Dict[str, str], str]]],
|
|
function_params: List[dict],
|
|
function,
|
|
processing: bool = False,
|
|
) -> List[str]:
|
|
"""Check if there are any wrong params passed to a command."""
|
|
wrong_params = []
|
|
if not processing:
|
|
for i, test_params in enumerate(function_params):
|
|
if "provider" in test_params and i != 0:
|
|
provider = test_params["provider"]
|
|
if provider in command_params:
|
|
for param in test_params:
|
|
if (
|
|
param
|
|
not in command_params[provider]["QueryParams"]["fields"]
|
|
and param
|
|
not in command_params["openbb"]["QueryParams"]["fields"] # type: ignore
|
|
and param != "provider"
|
|
):
|
|
wrong_params.append(
|
|
f"Wrong param {param} for provider {provider} in function {function}"
|
|
)
|
|
elif isinstance(command_params, dict):
|
|
providers = list(command_params.keys())
|
|
providers.remove("openbb")
|
|
for param in test_params:
|
|
is_wrong_param = True
|
|
for provider in providers:
|
|
if (
|
|
param in command_params[provider]["QueryParams"]["fields"]
|
|
or param
|
|
in command_params["openbb"]["QueryParams"]["fields"]
|
|
or param == "provider"
|
|
):
|
|
is_wrong_param = False
|
|
break
|
|
|
|
if is_wrong_param:
|
|
wrong_params.append(
|
|
f"Wrong param {param} in function {function}"
|
|
)
|
|
|
|
else:
|
|
for test_params in function_params:
|
|
if isinstance(command_params, list):
|
|
try:
|
|
iter_commands_params = command_params[0][0]
|
|
except KeyError:
|
|
iter_commands_params = command_params[0] # type: ignore
|
|
|
|
if isinstance(test_params, dict):
|
|
param_keys = test_params.keys()
|
|
elif isinstance(test_params, tuple) and all(
|
|
isinstance(item, dict) for item in test_params
|
|
):
|
|
param_keys = [key for item in test_params for key in item]
|
|
else:
|
|
continue # Skip this iteration if test_params is neither a dict nor a tuple of dicts
|
|
|
|
for key in param_keys:
|
|
if key not in iter_commands_params and key != "return":
|
|
wrong_params.append(f"Wrong param {key} in function {function}")
|
|
return wrong_params
|
|
|
|
|
|
def check_missing_params(
|
|
command_params: Union[Dict[str, Dict[str, dict]], List[Tuple[Dict[str, str], str]]],
|
|
function_params: List[dict],
|
|
function,
|
|
processing: bool = False,
|
|
) -> List[str]:
|
|
"""Check if there are any missing params for a command."""
|
|
missing_params = []
|
|
if not processing:
|
|
for i, test_params in enumerate(function_params):
|
|
if "provider" in test_params and i != 0:
|
|
provider = test_params["provider"]
|
|
if provider in command_params:
|
|
for expected_param in command_params[provider]["QueryParams"][
|
|
"fields"
|
|
]:
|
|
if expected_param not in test_params:
|
|
missing_params.append(
|
|
f"Missing param {expected_param} for provider {provider} in function {function}"
|
|
)
|
|
elif isinstance(command_params, dict):
|
|
for expected_param in command_params["openbb"]["QueryParams"]["fields"]:
|
|
if expected_param not in test_params:
|
|
missing_params.append(
|
|
f"Missing standard param {expected_param} in function {function}"
|
|
)
|
|
else:
|
|
for test_params in function_params:
|
|
if isinstance(command_params, list):
|
|
try:
|
|
iter_commands_params = command_params[0][0]
|
|
except KeyError:
|
|
iter_commands_params = command_params[0] # type: ignore
|
|
|
|
for expected_param in iter_commands_params:
|
|
try:
|
|
used_params = test_params[0].keys()
|
|
except KeyError:
|
|
used_params = test_params.keys()
|
|
if expected_param not in used_params and expected_param not in (
|
|
"return",
|
|
"chart",
|
|
):
|
|
missing_params.append(
|
|
f"Missing param {expected_param} in function {function}"
|
|
)
|
|
return missing_params
|
|
|
|
|
|
def check_integration_tests(
|
|
functions: Dict[str, Any],
|
|
check_function: Callable[
|
|
[
|
|
Union[Dict[str, Dict[str, dict]], List[Tuple[Dict[str, str], str]]],
|
|
List[dict],
|
|
str,
|
|
bool,
|
|
],
|
|
List[str],
|
|
],
|
|
) -> List[str]:
|
|
"""Check if there are any missing items for integration tests."""
|
|
pi = ProviderInterface()
|
|
provider_interface_map = pi.map
|
|
cm = CommandMap(coverage_sep=".")
|
|
|
|
function_params: List[dict] = []
|
|
all_missing_items: List[str] = []
|
|
used_functions: List[str] = []
|
|
|
|
for command, model in cm.commands_model.items():
|
|
for function in functions:
|
|
if command[1:].replace(".", "_") == function.replace("test_", ""):
|
|
command_params: Dict[str, Dict[str, dict]] = provider_interface_map[
|
|
model
|
|
]
|
|
try:
|
|
function_params = functions[function].pytestmark[1].args[1]
|
|
except IndexError:
|
|
# Another decorator is below the parametrize decorator
|
|
function_params = functions[function].pytestmark[2].args[1]
|
|
missing_items = check_function(
|
|
command_params, function_params, function, False
|
|
)
|
|
all_missing_items.extend(missing_items)
|
|
used_functions.append(function)
|
|
|
|
# the processing commands are the ones that are left
|
|
processing_functions = [
|
|
function for function in functions if function not in used_functions
|
|
]
|
|
|
|
for route, _ in cm.map.items():
|
|
for function in processing_functions:
|
|
if route.replace("/", "_")[1:] == function.replace("test_", ""):
|
|
hints = get_type_hints(cm.map[route])
|
|
processing_command_params = [
|
|
{k: "" for k in get_test_params_data_processing(hints)}
|
|
]
|
|
if (
|
|
not processing_command_params
|
|
or len(functions[function].pytestmark) < 2
|
|
):
|
|
# If there are no params, we can skip this function
|
|
continue
|
|
try:
|
|
function_params = functions[function].pytestmark[1].args[1]
|
|
except IndexError:
|
|
# Another decorator is below the parametrize decorator
|
|
function_params = functions[function].pytestmark[2].args[1]
|
|
|
|
missing_items = check_function(
|
|
processing_command_params, function_params, function, True # type: ignore
|
|
)
|
|
|
|
# if "chart" is in missing_items, remove it
|
|
if "chart" in missing_items:
|
|
missing_items.remove("chart")
|
|
|
|
all_missing_items.extend(missing_items)
|
|
|
|
return all_missing_items
|
|
|
|
|
|
def check_missing_integration_tests(test_type: Literal["api", "python"]) -> List[str]:
|
|
"""Check if all endpoints have integration tests."""
|
|
cm = CommandMap(coverage_sep=".")
|
|
routes = [route[1:].replace("/", "_") for route in cm.map]
|
|
missing_integration_tests: List[str] = []
|
|
|
|
if test_type == "api":
|
|
functions = get_module_functions(get_integration_tests(test_type="api"))
|
|
else:
|
|
functions = get_module_functions(get_integration_tests(test_type="python"))
|
|
|
|
tested_functions = [
|
|
function.replace("test_", "", 1)
|
|
for function in functions
|
|
if function.startswith("test_")
|
|
]
|
|
|
|
for route in routes:
|
|
if route not in tested_functions:
|
|
# TODO: See how to handle edge cases that are excluded from the schema
|
|
# on purpose. This is currently on the econometrics router.
|
|
if test_type == "api" and "econometrics" in route:
|
|
continue
|
|
missing_integration_tests.append(
|
|
f"Missing {test_type} integration test for route {route}"
|
|
)
|
|
|
|
return missing_integration_tests
|
|
|
|
|
|
def check_outdated_integration_tests(test_type: Literal["api", "python"]) -> List[str]:
|
|
"""Check if there are any outdated integration tests."""
|
|
cm = CommandMap(coverage_sep=".")
|
|
routes = [route[1:].replace("/", "_") for route in cm.map]
|
|
outdated_integration_tests: List[str] = []
|
|
|
|
if test_type == "api":
|
|
functions = get_module_functions(get_integration_tests(test_type="api"))
|
|
else:
|
|
functions = get_module_functions(get_integration_tests(test_type="python"))
|
|
|
|
for function, f_callable in functions.items():
|
|
if function.startswith("test_"):
|
|
route = function.replace("test_", "", 1)
|
|
try:
|
|
if f_callable.pytestmark[1].name != "skip":
|
|
args = f_callable.pytestmark[1].args[1]
|
|
else:
|
|
continue
|
|
except IndexError:
|
|
continue
|
|
if route not in routes and len(args) > 0:
|
|
# If it doesn't have any args it is because it is not installed.
|
|
outdated_integration_tests.append(
|
|
f"Outdated {test_type} integration test for route {route}"
|
|
)
|
|
|
|
return outdated_integration_tests
|
|
|
|
|
|
def check_missing_integration_test_providers(functions: Dict[str, Any]) -> List[str]:
|
|
"""Check if there are any missing providers for integration tests."""
|
|
return check_integration_tests(functions, check_missing_providers)
|
|
|
|
|
|
def check_missing_integration_test_params(functions: Dict[str, Any]) -> List[str]:
|
|
"""Check if there are any missing params for integration tests."""
|
|
return check_integration_tests(functions, check_missing_params)
|
|
|
|
|
|
def check_wrong_integration_test_params(functions: Dict[str, Any]) -> List[str]:
|
|
"""Check if there are any wrong params for integration tests."""
|
|
return check_integration_tests(functions, check_wrong_params)
|