Fix bug. Add router test. (#5888)

* Fix bug. Add router test.

* Simplify
This commit is contained in:
Igor Radovanovic
2023-12-14 18:09:48 +01:00
committed by GitHub
parent f9fbb2a831
commit 5faa50c0d4
2 changed files with 62 additions and 7 deletions

View File

@@ -1,9 +1,18 @@
"""Test the routers."""
from extensions.tests.utils.router_testers import check_router_function_models
from extensions.tests.utils.router_testers import (
check_router_function_models,
check_router_model_functions_signature,
)
def test_router_function_models() -> None:
"""Test if the models in the router functions exist in the provider interface map."""
missing_models = check_router_function_models()
assert not missing_models, "\n".join(missing_models)
def test_router_model_functions_signature() -> None:
"""Test if the router functions have the correct signature."""
missing_args = check_router_model_functions_signature()
assert not missing_args, "\n".join(missing_args)

View File

@@ -8,13 +8,22 @@ from typing import Dict, List, Optional
from openbb_core.app.provider_interface import ProviderInterface
def collect_routers(path) -> List:
"""Collect all routers in the path."""
def collect_routers(target_dir: str) -> List[str]:
"""Collect all routers in the target directory."""
current_dir = os.path.dirname(__file__)
base_path = os.path.abspath(os.path.join(current_dir, "../../../"))
full_target_path = os.path.abspath(os.path.join(base_path, target_dir))
routers = []
for root, _, files in os.walk(path):
for root, _, files in os.walk(full_target_path):
for name in files:
if name.endswith("_router.py"):
routers.append(os.path.join(root, name))
full_path = os.path.join(root, name)
# Convert the full path to a module path
relative_path = os.path.relpath(full_path, base_path)
module_path = relative_path.replace("/", ".").replace(".py", "")
routers.append(module_path)
return routers
@@ -23,7 +32,7 @@ def import_routers(routers: List) -> List:
"""Import all routers."""
loaded_routers: List = []
for router in routers:
module = importlib.import_module(router.replace("/", ".").replace(".py", ""))
module = importlib.import_module(router)
loaded_routers.append(module)
return loaded_routers
@@ -44,6 +53,10 @@ def collect_router_functions(loaded_routers: List) -> Dict:
def find_decorator(file_path: str, function_name: str) -> Optional[str]:
"""Find the decorator of the function in the file."""
this_dir = os.path.dirname(os.path.abspath(__file__))
file_path = os.path.join(
this_dir.split("openbb_platform/")[0], "openbb_platform", file_path
)
with open(file_path) as file:
lines = file.readlines()
for index, line in enumerate(lines):
@@ -85,9 +98,42 @@ def check_router_function_models() -> List[str]:
"""Check if the models in the router functions exist in the provider interface map."""
pi = ProviderInterface()
pi_map = pi.map
routers = collect_routers(os.path.join("openbb_platform", "extensions", "routers"))
routers = collect_routers("extensions")
loaded_routers = import_routers(routers)
router_functions = collect_router_functions(loaded_routers)
missing_models = find_missing_router_function_models(router_functions, pi_map)
return missing_models
def check_router_model_functions_signature() -> List[str]:
"""Check if the router model functions have the correct signature."""
expected_args = ["cc", "provider_choices", "standard_params", "extra_params"]
expected_return_type = "OBBject"
missing_args: List[str] = []
missing_return_type: List[str] = []
routers = collect_routers("extensions")
loaded_routers = import_routers(routers)
router_functions = collect_router_functions(loaded_routers)
for router_name, functions in router_functions.items():
for function in functions:
decorator_filer = ["POST", "GET", None]
decorator = find_decorator(
os.path.join(*router_name.split(".")) + ".py",
function.__name__,
)
if decorator:
args = list(function.__code__.co_varnames)
if args != expected_args and decorator not in decorator_filer:
missing_args.append(
f"{function.__name__} in {router_name} doesn't have the expected args: {expected_args}"
)
if expected_return_type not in str(function.__annotations__["return"]):
missing_return_type.append(
f"{function.__name__} in {router_name}"
f"doesn't have the expected return type: {expected_return_type}"
)
return missing_args + missing_return_type