"""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)} ] 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)