path: root/openbb_platform/extensions/tests/utils/integration_tests_testers.py

"""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 != "return":
                        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]],