Merge pull request #5244 from OpenBB-finance/feature/unit-tests

Setup unit tests folder structure and framework
This commit is contained in:
Igor Radovanovic
2023-08-07 22:04:57 +02:00
committed by GitHub
107 changed files with 4613 additions and 35 deletions

35
.github/workflows/sdkv4-core.yml vendored Normal file
View File

@@ -0,0 +1,35 @@
name: Test SDKV4
on:
pull_request:
branches:
- feature/openbb-sdk-v4
concurrency:
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
cancel-in-progress: true
jobs:
unit_tests:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python_version:
["3.8", "3.9", "3.10", "3.11"]
steps:
- name: Checkout code
uses: actions/checkout@v3
with:
ref: ${{ github.event.pull_request.head.ref }}
- name: Install Python ${{ matrix.python_version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python_version }}
allow-prereleases: true
cache: "pip"
- name: Run tests
run: |
pip install nox
nox -s tests --python ${{ matrix.python_version }}

8
noxfile.py Normal file
View File

@@ -0,0 +1,8 @@
import nox
@nox.session(python=["3.8", "3.9", "3.10", "3.11"])
def tests(session):
session.install("./openbb_sdk")
session.install("pytest")
session.run("pytest", "openbb_sdk/sdk/core/tests")

View File

View File

View File

View File

View File

View File

View File

View File

View File

View File

View File

@@ -2,9 +2,10 @@ from datetime import date
from typing import Any, Optional
from urllib.parse import urlencode
root_url = "https://api.stlouisfed.org/fred"
from openbb_provider import helpers
root_url = "https://api.stlouisfed.org/fred"
class Fred:
def __init__(self, api_key: Optional[str]):

View File

View File

View File

@@ -22,6 +22,10 @@ openbb-fmp = { path = "./providers/fmp" }
openbb-polygon = { path = "./providers/polygon" }
openbb-fred = { path = "./providers/fred" }
[tool.poetry.group.dev.dependencies]
pytest = "^6.2.2"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

View File

@@ -1,3 +1,4 @@
"""System dependency."""
from copy import deepcopy
from openbb_core.app.model.system_settings import SystemSettings
@@ -7,10 +8,12 @@ __system_service = SystemService()
async def get_system_settings() -> SystemSettings:
"""Get system settings."""
global __system_service # noqa: PLW0602 # pylint: disable=global-variable-not-assigned
return deepcopy(__system_service.system_settings)
async def get_system_service() -> SystemService:
"""Get system service."""
global __system_service # noqa: PLW0602 # pylint: disable=global-variable-not-assigned
return __system_service

View File

@@ -1,3 +1,4 @@
"""User dependency."""
from datetime import datetime, timedelta
from typing import Optional
@@ -23,10 +24,12 @@ pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password, hashed_password):
"""Verify password."""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
"""Get password hash."""
return pwd_context.hash(password)
@@ -35,6 +38,7 @@ def authenticate_user(
username: str,
user_service: UserService,
) -> Optional[UserSettings]:
"""Authenticate user."""
user = user_service.user_settings_repository.read_by_profile(
filter_list=[("username", username)],
)
@@ -53,6 +57,7 @@ def create_access_token(
sub: str,
expiration_delta: Optional[timedelta] = None,
) -> AccessToken:
"""Create access token."""
if expiration_delta is None:
expiration_delta = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
@@ -68,6 +73,7 @@ def create_jwt_token(
key=SECRET_KEY,
algorithm=ALGORITHM,
) -> str:
"""Create JWT token."""
claims = access_token.dict()
encoded_jwt = jwt.encode(claims=claims, key=key, algorithm=algorithm)
return encoded_jwt
@@ -76,6 +82,7 @@ def create_jwt_token(
async def get_user_service(
system_settings: Annotated[SystemSettings, Depends(get_system_settings)]
) -> UserService:
"""Get user service."""
global __user_service # pylint: disable=global-statement
if __user_service is None:
@@ -93,6 +100,7 @@ async def get_user(
jwt_token: Annotated[str, Depends(oauth2_scheme)],
user_service: Annotated[UserService, Depends(get_user_service)],
):
"""Get user."""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",

View File

@@ -1,3 +1,4 @@
"""Access Token Model."""
from datetime import datetime
from openbb_core.app.model.abstract.tagged import Tagged
@@ -5,10 +6,12 @@ from pydantic import Field
class AccessToken(Tagged):
"""This will be turn into a JWT Token"""
"""Turn into a JWT Token."""
sub: str = Field(description="Subject of the token, here : `UserSettings.id`")
exp: datetime = Field(description="Expiration datetime of the `access_token`.")
class Config:
"""Pydantic Config."""
validate_assignment = True

View File

@@ -1,6 +1,9 @@
"""Token response model."""
from pydantic import BaseModel
class TokenResponse(BaseModel):
"""Token response model."""
access_token: str
token_type: str

View File

@@ -8,9 +8,11 @@ router = APIRouter(prefix="/coverage", tags=["Coverage"])
@router.get("/providers")
async def get_provider_coverage() -> Dict[str, List[str]]:
"""Get provider coverage."""
return CommandMap().provider_coverage
@router.get("/commands")
async def get_command_coverage() -> Dict[str, List[str]]:
"""Get command coverage."""
return CommandMap().command_coverage

View File

@@ -1,9 +1,5 @@
from fastapi import APIRouter, Depends
from openbb_core.api.dependency.user import (
UserService,
get_user,
get_user_service,
)
from openbb_core.api.dependency.user import UserService, get_user, get_user_service
from openbb_core.app.model.credentials import Credentials
from openbb_core.app.model.user_settings import UserSettings
from typing_extensions import Annotated
@@ -15,6 +11,7 @@ router = APIRouter(prefix="/settings", tags=["Settings"])
async def read_users_settings(
user_settings: Annotated[UserSettings, Depends(get_user)],
) -> UserSettings:
"""Read users settings."""
return user_settings
@@ -24,6 +21,7 @@ async def patch_user_credentials(
user_settings: Annotated[UserSettings, Depends(get_user)],
user_service: Annotated[UserService, Depends(get_user_service)],
) -> UserSettings:
"""Patch user credentials."""
current = user_settings.credentials.dict()
incoming = credentials.dict(exclude_none=True)
current.update(incoming)

View File

@@ -1,3 +1,4 @@
"""System router."""
from fastapi import APIRouter, Depends
from openbb_core.api.dependency.system import get_system_settings
from openbb_core.app.model.system_settings import SystemSettings
@@ -10,4 +11,5 @@ router = APIRouter(prefix="/system", tags=["System"])
async def get_system_model(
system_settings: Annotated[SystemSettings, Depends(get_system_settings)],
) -> SystemSettings:
"""Get system model."""
return system_settings

View File

@@ -82,8 +82,8 @@ class FormatterWithExceptions(logging.Formatter):
return text_mocked
@classmethod
def filter_special_tags(cls, text: str) -> str:
@staticmethod
def filter_special_tags(text: str) -> str:
text_filtered = text.replace("\n", " MOCKING_BREAKLINE ")
text_filtered = text_filtered.replace("'Traceback", "Traceback")

View File

@@ -16,8 +16,6 @@ from openbb_core.app.logs.models.logging_settings import LoggingSettings
class HandlersManager:
def __init__(self, settings: LoggingSettings):
self.is_initialized = True
self._handlers = settings.handler_list
self._settings = settings

View File

@@ -8,7 +8,14 @@ from openbb_core.app.model.user_settings import UserSettings
# pylint: disable=too-many-instance-attributes
class LoggingSettings:
def __init__(self, user_settings: UserSettings, system_settings: SystemSettings):
def __init__(
self,
user_settings: Optional[UserSettings] = None,
system_settings: Optional[SystemSettings] = None,
):
user_settings = user_settings or UserSettings()
system_settings = system_settings or SystemSettings()
user_data_directory = (
str(Path.home() / "OpenBBUserData")
if not user_settings.preferences

View File

@@ -15,8 +15,8 @@ def get_timestamp_from_x_days(x: int) -> float:
def get_expired_file_list(directory: Path, before_timestamp: float) -> List[Path]:
expired_files = list()
if directory.exists and directory.is_dir(): # type: ignore
expired_files = []
if directory.is_dir(): # Check if the directory exists and is a directory
for file in directory.iterdir():
if file.is_file() and file.lstat().st_mtime < before_timestamp:
expired_files.append(file)

View File

@@ -24,8 +24,7 @@ def get_app_id(contextual_user_data_directory: str) -> str:
print(
"Note for macOS users: copy `OpenBB Terminal` folder outside the DMG."
)
else:
raise e
raise e
except Exception as e:
raise e
@@ -35,19 +34,37 @@ def get_app_id(contextual_user_data_directory: str) -> str:
def get_log_dir(contextual_user_data_directory: str) -> Path:
"""Retrieve application's log directory."""
log_dir = Path(contextual_user_data_directory).joinpath("logs").absolute()
log_dir = create_log_dir_if_not_exists(contextual_user_data_directory)
logging_uuid = create_log_uuid_if_not_exists(log_dir)
uuid_log_dir = create_uuid_dir_if_not_exists(log_dir, logging_uuid)
return uuid_log_dir
def create_log_dir_if_not_exists(contextual_user_data_directory: str) -> None:
log_dir = Path(contextual_user_data_directory).joinpath("logs").absolute()
if not log_dir.is_dir():
log_dir.mkdir(parents=True, exist_ok=True)
log_id = (log_dir / ".logid").absolute()
return log_dir
def create_log_uuid_if_not_exists(log_dir: Path) -> str:
log_id = get_log_id(log_dir)
if not log_id.is_file():
logging_id = f"{uuid.uuid4()}"
log_id.write_text(logging_id, encoding="utf-8")
else:
logging_id = log_id.read_text(encoding="utf-8").rstrip()
return logging_id
def get_log_id(log_dir):
return (log_dir / ".logid").absolute()
def create_uuid_dir_if_not_exists(log_dir, logging_id) -> str:
uuid_log_dir = (log_dir / logging_id).absolute()
if not uuid_log_dir.is_dir():

View File

@@ -26,22 +26,26 @@ class ChartingSettings:
# System
self.log_collect: bool = system_settings.log_collect
self.version = system_settings.version
self.python_version = system_settings.python_version
self.version: str = system_settings.version
self.python_version: str = system_settings.python_version
self.test_mode = system_settings.test_mode
self.app_id: str = get_app_id(user_data_directory)
self.debug_mode: bool = system_settings.debug_mode
self.headless: bool = system_settings.headless
# User
self.plot_enable_pywry = user_settings.preferences.plot_enable_pywry
self.plot_pywry_width = user_settings.preferences.plot_pywry_width
self.plot_pywry_height = user_settings.preferences.plot_pywry_height
self.plot_open_export = user_settings.preferences.plot_open_export
self.user_email = user_settings.profile.hub_session.email if has_hub else None
self.user_uuid = (
self.plot_enable_pywry: bool = user_settings.preferences.plot_enable_pywry
self.plot_pywry_width: int = user_settings.preferences.plot_pywry_width
self.plot_pywry_height: int = user_settings.preferences.plot_pywry_height
self.plot_open_export: bool = user_settings.preferences.plot_open_export
self.user_email: Optional[str] = (
user_settings.profile.hub_session.email if has_hub else None
)
self.user_uuid: Optional[str] = (
user_settings.profile.hub_session.user_uuid if has_hub else None
)
self.user_exports_directory = user_settings.preferences.export_directory
self.user_styles_directory = user_settings.preferences.user_styles_directory
self.user_exports_directory: str = user_settings.preferences.export_directory
self.user_styles_directory: str = (
user_settings.preferences.user_styles_directory
)
# Theme
self.chart_style = user_settings.preferences.chart_style
self.chart_style: str = user_settings.preferences.chart_style

View File

@@ -1,3 +1,4 @@
"""Hub manager class."""
from typing import Optional
import requests
@@ -12,11 +13,11 @@ from openbb_core.app.model.user_settings import UserSettings
class OpenBBError(Exception):
pass
"""OpenBB error."""
class HubService:
"""Hub manager class"""
"""Manage Hub service class."""
BASE_URL = "https://payments.openbb.dev"
TIMEOUT = 10

View File

@@ -1,3 +1,4 @@
"""Journal service."""
from typing import Optional
from openbb_core.app.repository.abstract.journal_entry_repository import (
@@ -22,8 +23,11 @@ from pymongo.mongo_client import MongoClient
class JournalService:
"""Journal service."""
@staticmethod
def valid_client(mongodb_client: Optional[MongoClient] = None) -> bool:
"""Check if the client is valid."""
if mongodb_client is None:
return False
@@ -41,6 +45,7 @@ class JournalService:
mongodb_client: Optional[MongoClient] = None,
journal_entry_repository: Optional[AbstractJournalEntryRepository] = None,
) -> AbstractJournalEntryRepository:
"""Build journal entry repository."""
if journal_entry_repository:
pass
elif mongodb_client and valid_client:
@@ -57,6 +62,7 @@ class JournalService:
mongodb_client: Optional[MongoClient] = None,
journal_repository: Optional[AbstractJournalRepository] = None,
) -> AbstractJournalRepository:
"""Build journal repository."""
if journal_repository:
pass
elif mongodb_client and valid_client:
@@ -87,12 +93,15 @@ class JournalService:
@property
def mongodb_client(self) -> Optional[MongoClient]:
"""Return mongodb client."""
return self._mongodb_client
@property
def journal_entry_repository(self) -> AbstractJournalEntryRepository:
"""Return journal entry repository."""
return self._journal_entry_repository
@property
def journal_repository(self) -> AbstractJournalRepository:
"""Return journal repository."""
return self._journal_repository

View File

@@ -7,6 +7,8 @@ from openbb_core.app.model.system_settings import SystemSettings
class SystemService:
"""System service."""
SYSTEM_SETTINGS_PATH = SYSTEM_SETTINGS_PATH
SYSTEM_SETTINGS_ALLOWED_FIELD_SET = {
"log_collect",
@@ -21,6 +23,7 @@ class SystemService:
def read_default_system_settings(
cls, path: Optional[Path] = None
) -> SystemSettings:
"""Read default system settings."""
path = path or cls.SYSTEM_SETTINGS_PATH
if path.exists():
@@ -45,6 +48,7 @@ class SystemService:
system_settings: SystemSettings,
path: Optional[Path] = None,
) -> None:
"""Write default system settings."""
path = path or cls.SYSTEM_SETTINGS_PATH
system_settings_json = system_settings.json(
@@ -64,13 +68,16 @@ class SystemService:
@property
def system_settings(self) -> SystemSettings:
"""Get system settings."""
return self._system_settings
@system_settings.setter
def system_settings(self, system_settings: SystemSettings) -> None:
"""Set system settings."""
self._system_settings = system_settings
def refresh_system_settings(self) -> SystemSettings:
"""Refresh system settings."""
self._system_settings = self.read_default_system_settings()
return self._system_settings

View File

@@ -1,3 +1,4 @@
"""User service."""
from functools import reduce
from pathlib import Path
from typing import Any, Dict, List, MutableMapping, Optional
@@ -26,6 +27,8 @@ from pymongo.mongo_client import MongoClient
class UserService:
"""User service."""
REPOSITORY_DIRECTORY = Path(__file__).parent.parent.parent.parent
USER_SETTINGS_PATH = USER_SETTINGS_PATH
USER_SETTINGS_ALLOWED_FIELD_SET = {"credentials", "preferences", "defaults"}
@@ -36,6 +39,7 @@ class UserService:
mongodb_client: Optional[MongoClient] = None,
access_token_repository: Optional[AbstractAccessTokenRepository] = None,
) -> AbstractAccessTokenRepository:
"""Build token repository."""
if access_token_repository:
pass
elif mongodb_client and valid_client:
@@ -52,6 +56,7 @@ class UserService:
mongodb_client: Optional[MongoClient] = None,
user_settings_repository: Optional[AbstractUserSettingsRepository] = None,
) -> AbstractUserSettingsRepository:
"""Build user settings repository."""
if user_settings_repository:
pass
elif mongodb_client and valid_client:
@@ -64,6 +69,7 @@ class UserService:
@staticmethod
def valid_client(client: Optional[MongoClient] = None) -> bool:
"""Check if the client is valid."""
if client is None:
return False
@@ -77,6 +83,7 @@ class UserService:
@classmethod
def read_default_user_settings(cls, path: Optional[Path] = None) -> UserSettings:
"""Read default user settings."""
path = path or cls.USER_SETTINGS_PATH
if path.exists():
@@ -95,6 +102,7 @@ class UserService:
user_settings: UserSettings,
path: Optional[Path] = None,
) -> None:
"""Write default user settings."""
path = path or cls.USER_SETTINGS_PATH
user_settings_json = user_settings.json(
@@ -107,6 +115,7 @@ class UserService:
@classmethod
def update_default(cls, user_settings: UserSettings) -> UserSettings:
"""Update default user settings."""
d1 = cls.read_default_user_settings().dict()
d2 = user_settings.dict() if user_settings else {}
updated = cls.merge_dicts([d1, d2])
@@ -158,25 +167,31 @@ class UserService:
@property
def mongodb_client(self) -> Optional[MongoClient]:
"""Return MongoDB client."""
return self._mongodb_client
@property
def default_user_settings(self) -> UserSettings:
"""Return default user settings."""
return self._default_user_settings
@default_user_settings.setter
def default_user_settings(self, default_user_settings: UserSettings) -> None:
"""Set default user settings."""
self._default_user_settings = default_user_settings
@property
def access_token_repository(self) -> AbstractAccessTokenRepository:
"""Access token repository."""
return self._token_repository
@property
def user_settings_repository(self) -> AbstractUserSettingsRepository:
"""User settings repository."""
return self._user_settings_repository
def refresh_default_default_user_settings(self) -> UserSettings:
"""Refresh default default user settings."""
self._default_user_settings = self.read_default_user_settings()
return self._default_user_settings

View File

@@ -1,3 +1,4 @@
"""App factory."""
# pylint: disable=W0231:super-init-not-called
from openbb_core.app.command_runner import CommandRunnerSession
@@ -8,6 +9,7 @@ from openbb_core.app.static.coverage import Coverage
def create_app():
"""Create the app."""
try:
# pylint: disable=import-outside-toplevel
from openbb_core.app.static.package.__extensions__ import Extensions

View File

@@ -1,6 +1,9 @@
"""Container class."""
from openbb_core.app.command_runner import CommandRunnerSession
class Container:
"""Container class for the command runner session."""
def __init__(self, command_runner_session: CommandRunnerSession) -> None:
self._command_runner_session = command_runner_session

View File

@@ -1,14 +1,19 @@
"""Coverage module."""
from openbb_core.app.router import CommandMap
class Coverage:
"""Coverage class."""
def __init__(self):
self._command_map = CommandMap(coverage_sep=".")
@property
def providers(self):
"""Return providers coverage."""
return self._command_map.provider_coverage
@property
def commands(self):
"""Return commands coverage."""
return self._command_map.command_coverage

View File

@@ -1,3 +1,5 @@
"""OpenBB filters."""
import builtins
from functools import wraps
@@ -10,10 +12,12 @@ from openbb_core.app.utils import df_to_basemodel
class OpenBBError(Exception):
pass
"""A custom exception for OpenBB errors."""
def filter_call(func):
"""Filter command call."""
@wraps(wrapped=func)
def inner(*args, **kwargs):
self = args[0]
@@ -36,7 +40,7 @@ def filter_call(func):
def filter_inputs(**kwargs) -> dict:
"""Filter command inputs"""
"""Filter command inputs."""
for key, value in kwargs.items():
if isinstance(value, pd.DataFrame):
kwargs[key] = df_to_basemodel(value, index=True)
@@ -45,7 +49,7 @@ def filter_inputs(**kwargs) -> dict:
def filter_output(command_output: CommandOutput) -> CommandOutput:
"""Filter command output"""
"""Filter command output."""
if command_output.warnings:
for w in command_output.warnings:
category = getattr(builtins, w.category, OpenBBWarning)

View File

@@ -108,8 +108,11 @@ class PackageBuilder:
class ModuleBuilder:
"""Build the module for the SDK."""
@staticmethod
def build(path: str) -> str:
"""Build the module."""
code = "### THIS FILE IS AUTO-GENERATED. DO NOT EDIT. ###\n"
code += ImportDefinition.build(path=path)
code += ClassDefinition.build(path=path)
@@ -118,8 +121,11 @@ class ModuleBuilder:
class ImportDefinition:
"""Build the import definition for the SDK."""
@staticmethod
def filter_hint_type_list(hint_type_list: List[Type]) -> List[Type]:
"""Filter the hint type list."""
new_hint_type_list = []
for hint_type in hint_type_list:
if hint_type != _empty and hint_type.__module__ != "builtins":
@@ -131,6 +137,7 @@ class ImportDefinition:
@classmethod
def get_function_hint_type_list(cls, func: Callable) -> List[Type]:
"""Get the hint type list from the function."""
sig = signature(func)
parameter_map = sig.parameters
return_type = sig.return_annotation
@@ -150,6 +157,7 @@ class ImportDefinition:
@classmethod
def get_path_hint_type_list(cls, path: str) -> List[Type]:
"""Get the hint type list from the path."""
route_map = PathHandler.build_route_map()
path_list = PathHandler.build_path_list(route_map=route_map)
child_path_list = PathHandler.get_child_path_list(
@@ -167,6 +175,7 @@ class ImportDefinition:
@classmethod
def build(cls, path: str) -> str:
"""Build the import definition."""
hint_type_list = cls.get_path_hint_type_list(path=path)
code = "\nfrom openbb_core.app.static.container import Container"
code += "\nfrom openbb_core.app.model.command_output import CommandOutput"
@@ -196,8 +205,11 @@ class ImportDefinition:
class ClassDefinition:
"""Build the class definition for the SDK."""
@staticmethod
def build(path: str) -> str:
"""Build the class definition."""
class_name = PathHandler.build_module_class(path=path)
code = f"\nclass {class_name}(Container):\n"
route_map = PathHandler.build_route_map()
@@ -375,8 +387,11 @@ class DocstringGenerator:
class MethodDefinition:
"""Build the method definition for the SDK."""
@staticmethod
def build_class_loader_method(path: str) -> str:
"""Build the class loader method."""
module_name = PathHandler.build_module_name(path=path)
class_name = PathHandler.build_module_class(path=path)
function_name = path.rsplit("/", maxsplit=1)[-1].strip("/")
@@ -390,6 +405,7 @@ class MethodDefinition:
@staticmethod
def get_type(field: ModelField) -> type:
"""Get the type of the field."""
field_type = getattr(field, "type", Parameter.empty)
if isclass(field_type):
name = field_type.__name__
@@ -401,6 +417,7 @@ class MethodDefinition:
@staticmethod
def get_default(field: ModelField):
"""Get the default value of the field."""
field_default = getattr(field, "default", None)
if field_default is None or field_default is MISSING:
return Parameter.empty
@@ -413,12 +430,14 @@ class MethodDefinition:
@staticmethod
def is_annotated_dc(annotation) -> bool:
"""Check if the annotation is an annotated dataclass."""
return type(annotation) is _AnnotatedAlias and hasattr(
annotation.__args__[0], "__dataclass_fields__"
)
@staticmethod
def reorder_params(params: Dict[str, Parameter]) -> "OrderedDict[str, Parameter]":
"""Reorder the params."""
formatted_keys = list(params.keys())
for k in ["provider", "extra_params"]:
if k in formatted_keys:
@@ -435,6 +454,7 @@ class MethodDefinition:
def format_params(
parameter_map: Dict[str, Parameter]
) -> "OrderedDict[str, Parameter]":
"""Format the params."""
# These are types we want to expand.
# For example, start_date is always a 'date', but we also accept 'str' as input.
# Be careful, if the type is not coercible by pydantic to the original type, you
@@ -498,6 +518,7 @@ class MethodDefinition:
@staticmethod
def build_func_params(parameter_map: Dict[str, Parameter]) -> str:
"""Build the function parameters."""
od = MethodDefinition.format_params(parameter_map=parameter_map)
func_params = ", ".join(str(param) for param in od.values())
func_params = func_params.replace("NoneType", "None")
@@ -509,6 +530,7 @@ class MethodDefinition:
@staticmethod
def build_func_returns(return_type: type) -> str:
"""Build the function returns."""
if return_type == _empty:
func_returns = "None"
elif return_type.__module__ == "builtins":
@@ -537,6 +559,7 @@ class MethodDefinition:
def build_command_method_signature(
func_name: str, parameter_map: Dict[str, Parameter], return_type: type
) -> str:
"""Build the command method signature."""
func_params = MethodDefinition.build_func_params(parameter_map)
func_returns = MethodDefinition.build_func_returns(return_type)
code = "\n @filter_call"
@@ -553,12 +576,14 @@ class MethodDefinition:
@staticmethod
def build_command_method_doc(func: Callable):
"""Build the command method docstring."""
code = f' """{func.__doc__}"""\n' if func.__doc__ else ""
return code
@staticmethod
def build_command_method_implementation(path: str, func: Callable):
"""Build the command method implementation."""
sig = signature(func)
parameter_map = dict(sig.parameters)
parameter_map.pop("cc", None)
@@ -596,6 +621,7 @@ class MethodDefinition:
def build_command_method(
cls, path: str, func: Callable, model_name: Optional[str]
) -> str:
"""Build the command method."""
func_name = func.__name__
sig = signature(func)
@@ -618,8 +644,11 @@ class MethodDefinition:
class PathHandler:
"""Handle the paths for the SDK."""
@staticmethod
def build_route_map() -> Dict[str, BaseRoute]:
"""Build the route map."""
router = RouterLoader.from_extensions()
route_map = {route.path: route for route in router.api_router.routes} # type: ignore
@@ -627,6 +656,7 @@ class PathHandler:
@staticmethod
def build_path_list(route_map: Dict[str, BaseRoute]) -> List[str]:
"""Build the path list."""
path_list = []
for route_path in route_map:
if route_path not in path_list:
@@ -643,10 +673,12 @@ class PathHandler:
@staticmethod
def get_route(path: str, route_map: Dict[str, BaseRoute]):
"""Get the route from the path."""
return route_map.get(path, None)
@staticmethod
def get_child_path_list(path: str, path_list: List[str]) -> List[str]:
"""Get the child path list."""
direct_children = []
for p in path_list:
if p.startswith(path):
@@ -658,28 +690,34 @@ class PathHandler:
@staticmethod
def clean_path(path: str) -> str:
"""Clean the path."""
if path.startswith("/"):
path = path[1:]
return path.replace("-", "_").replace("/", "_")
@classmethod
def build_module_name(cls, path: str) -> str:
"""Build the module name."""
if path == "":
return "__extensions__"
return cls.clean_path(path=path)
@classmethod
def build_module_class(cls, path: str) -> str:
"""Build the module class."""
if path == "":
return "Extensions"
return f"CLASS_{cls.clean_path(path=path)}"
class Linters:
"""Run the linters for the SDK."""
current_folder = str(Path(Path(__file__).parent, "package").absolute())
@staticmethod
def print_separator(symbol: str, length: int = 160):
"""Print a separator."""
print(symbol * length)
@staticmethod
@@ -701,12 +739,15 @@ class Linters:
@classmethod
def black(cls):
"""Run black."""
cls.run(linter="black")
@classmethod
def ruff(cls):
"""Run ruff."""
cls.run(linter="ruff", flags=["--fix"])
@classmethod
def mypy(cls):
"""Run mypy."""
cls.run(linter="mypy", flags=["--ignore-missing-imports"])

View File

View File

@@ -0,0 +1,34 @@
"""Test the system module."""
import asyncio
from unittest.mock import MagicMock, patch
from openbb_core.api.dependency.system import (
SystemSettings,
get_system_service,
get_system_settings,
)
@patch("openbb_core.api.dependency.system.SystemService")
def test_get_system_settings(mock_system_service):
"""Test get_system_settings."""
mock_system_settings = MagicMock(spec=SystemSettings)
mock_system_service.system_settings = mock_system_settings
with patch(
"openbb_core.api.dependency.system.__system_service", new=mock_system_service
):
system_settings = asyncio.run(get_system_settings())
assert system_settings == mock_system_settings
@patch("openbb_core.api.dependency.system.SystemService")
def test_get_system_service(mock_system_service):
"""Test get_system_service."""
with patch(
"openbb_core.api.dependency.system.__system_service", new=mock_system_service
):
system_service = asyncio.run(get_system_service())
assert system_service == mock_system_service

View File

@@ -0,0 +1,80 @@
"""Test the user module."""
import asyncio
from unittest.mock import MagicMock, patch
from openbb_core.api.dependency.user import (
AccessToken,
SystemSettings,
UserSettings,
authenticate_user,
create_access_token,
create_jwt_token,
get_password_hash,
get_user,
get_user_service,
verify_password,
)
def test_password_functions():
"""Test password functions."""
password = "my_password"
hashed_password = get_password_hash(password)
assert verify_password(password, hashed_password)
@patch("openbb_core.api.dependency.user.UserService")
def test_authenticate_user(mock_user_service):
"""Test authenticate_user."""
mock_user_settings = MagicMock(
spec=UserSettings, profile=MagicMock(password_hash="hashed_password")
)
mock_user_service.user_settings_repository.read_by_profile.return_value = (
mock_user_settings
)
with patch("openbb_core.api.dependency.user.verify_password", return_value=True):
result = authenticate_user("password", "username", mock_user_service)
assert result == mock_user_settings
def test_create_access_token():
"""Test create_access_token."""
sub = "subject"
access_token = create_access_token(sub)
assert isinstance(access_token, AccessToken)
assert access_token.sub == sub
def test_create_jwt_token():
"""Test create_jwt_token."""
mock_access_token = MagicMock(
spec=AccessToken, dict=lambda: {"sub": "subject", "exp": 12345}
)
jwt_token = create_jwt_token(mock_access_token)
assert isinstance(jwt_token, str)
@patch("openbb_core.api.dependency.user.UserService")
def test_get_user_service(mock_user_service):
"""Test get_user_service."""
mock_system_settings = MagicMock(spec=SystemSettings, dbms_uri=None)
with patch("openbb_core.api.dependency.user.__user_service", new=None):
asyncio.run(get_user_service(mock_system_settings))
mock_user_service.assert_called_once_with()
@patch("openbb_core.api.dependency.user.jwt.decode")
@patch("openbb_core.api.dependency.user.UserService")
def test_get_user(mock_user_service, mock_decode):
"""Test get_user."""
mock_decode.return_value = {"sub": "user_id"}
mock_user_settings = MagicMock(spec=UserSettings, profile=MagicMock(active=True))
mock_user_service.user_settings_repository.read.return_value = mock_user_settings
with patch("openbb_core.api.dependency.user.__user_service", new=mock_user_service):
result = asyncio.run(get_user("jwt_token", mock_user_service))
assert result == mock_user_settings

View File

@@ -0,0 +1,29 @@
"""Test coverage for the router module."""
from unittest.mock import patch
from openbb_core.api.router.coverage import get_command_coverage, get_provider_coverage
@patch("openbb_core.api.router.coverage.CommandMap")
def test_get_provider_coverage(mock_command_map):
"""Test get provider coverage."""
mock_command_map.return_value.provider_coverage = {
"provider1": ["coverage1", "coverage2"]
}
response = get_provider_coverage()
assert response
@patch("openbb_core.api.router.coverage.CommandMap")
def test_get_command_coverage(mock_command_map):
"""Test get command coverage."""
mock_command_map.return_value.command_coverage = {
"command1": ["coverage1", "coverage2"]
}
response = get_command_coverage()
assert response

View File

@@ -0,0 +1,18 @@
"""Test the router system module."""
import asyncio
from unittest.mock import Mock, patch
from openbb_core.api.router.system import get_system_model
from openbb_core.app.model.system_settings import SystemSettings
@patch("openbb_core.api.router.system.get_system_settings")
def test_get_system_model(mock_get_system_settings):
"""Test get system model."""
mock_system_settings = Mock(spec=SystemSettings)
mock_get_system_settings.return_value = mock_system_settings
response = asyncio.run(get_system_model(system_settings=mock_system_settings))
assert response == mock_system_settings

View File

@@ -0,0 +1,33 @@
"""Test the router settings.py module."""
import asyncio
from unittest.mock import Mock
from openbb_core.api.router.settings import patch_user_credentials, read_users_settings
from openbb_core.app.model.credentials import Credentials
from openbb_core.app.model.user_settings import UserSettings
def test_read_users_settings():
"""Test read users settings."""
mock_user_settings = Mock()
result = asyncio.run(read_users_settings(user_settings=mock_user_settings))
assert result == mock_user_settings
def test_patch_user_credentials():
"""Test patch user credentials."""
credentials = Credentials(username="testuser", password="testpass")
user_settings = UserSettings(credentials=credentials)
mock_user_service = Mock()
result = asyncio.run(
patch_user_credentials(credentials, user_settings, mock_user_service)
)
assert result == user_settings
mock_user_service.user_settings_repository.update.assert_called_once_with(
user_settings
)

View File

@@ -0,0 +1,308 @@
import logging
import os
from unittest.mock import Mock
import pytest
from openbb_core.app.logs.formatters.formatter_with_exceptions import (
FormatterWithExceptions,
)
logging_settings = Mock()
logging_settings.app_name = "test_app_name"
logging_settings.app_id = "test_app_id"
logging_settings.session_id = "test_session_id"
logging_settings.commit_hash = "test_commit_hash"
logging_settings.user_id = "test_user_id"
@pytest.fixture
def formatter():
return FormatterWithExceptions(logging_settings)
# Test when exc_text is not empty (should return "X")
def test_level_name_with_exc_text(formatter):
record = logging.LogRecord(
name="test_logger",
level=logging.INFO,
pathname="",
lineno=0,
msg="",
args=None,
exc_info=None,
func=None,
)
record.exc_text = "Exception occurred!"
assert formatter.calculate_level_name(record) == "X"
# Test when levelname is not empty (should return the first character of levelname)
def test_level_name_with_levelname(formatter):
record = logging.LogRecord(
name="test_logger",
level=logging.WARNING,
pathname="",
lineno=0,
msg="",
args=None,
exc_info=None,
func=None,
)
record.levelname = "WARNING"
assert formatter.calculate_level_name(record) == "W"
# Test when func_name_override and session_id are present in the record
def test_extract_log_extra_with_override_and_session_id(formatter):
record = logging.LogRecord(
name="test_logger",
level=logging.INFO,
pathname="",
lineno=0,
msg="",
args=None,
exc_info=None,
func=None,
)
record.func_name_override = "custom_function"
record.session_id = "1234567890"
log_extra = formatter.extract_log_extra(record)
assert log_extra == {
"sessionId": "1234567890",
}
# Test when only func_name_override is present in the record
def test_extract_log_extra_with_override(formatter):
record = logging.LogRecord(
name="test_logger",
level=logging.INFO,
pathname="",
lineno=0,
msg="",
args=None,
exc_info=None,
func=None,
)
record.func_name_override = "custom_function"
log_extra = formatter.extract_log_extra(record)
assert log_extra == {}
assert record.funcName == "custom_function"
assert record.lineno == 0
# Test when only session_id is present in the record
def test_extract_log_extra_with_session_id(formatter):
record = logging.LogRecord(
name="test_logger",
level=logging.INFO,
pathname="",
lineno=0,
msg="",
args=None,
exc_info=None,
func=None,
)
record.session_id = "1234567890"
log_extra = formatter.extract_log_extra(record)
assert log_extra == {
"sessionId": "1234567890",
}
# Test when neither func_name_override nor session_id are present in the record
def test_extract_log_extra_with_no_override_or_session_id(formatter):
record = logging.LogRecord(
name="test_logger",
level=logging.INFO,
pathname="",
lineno=0,
msg="",
args=None,
exc_info=None,
func=None,
)
log_extra = formatter.extract_log_extra(record)
assert log_extra == {}
@pytest.mark.parametrize(
"input_text, expected_output",
[
("Hello, this is 192.168.1.1", "Hello, this is FILTERED_IP "),
("IP address: 10.0.0.1", "IP address: FILTERED_IP "),
("No IPs here!", "No IPs here!"),
("Another IP: 172.16.32.1", "Another IP: FILTERED_IP "),
("1.2.3.4", " FILTERED_IP "),
],
)
def test_mock_ipv4(input_text, expected_output, formatter):
assert formatter.mock_ipv4(input_text) == expected_output
@pytest.mark.parametrize(
"input_text, expected_output",
[
("Send an email to john@example.com", "Send an email to FILTERED_EMAIL "),
("No emails here!", "No emails here!"),
("Another email: alice@example.co.uk", "Another email: FILTERED_EMAIL "),
("Contact us: support@example.net", "Contact us: FILTERED_EMAIL "),
("myemail@example.com 12345", " FILTERED_EMAIL 12345"),
],
)
def test_mock_email(input_text, expected_output, formatter):
assert formatter.mock_email(input_text) == expected_output
@pytest.mark.parametrize(
"input_text, expected_output",
[
('{"password": "secure_pass_123"}', '{"password": " FILTERED_PASSWORD "}'),
('{"password": "p@$$w0rd!"}', '{"password": " FILTERED_PASSWORD "}'),
('{"password": "12345"}', '{"password": " FILTERED_PASSWORD "}'),
('{"password": "abc"}', '{"password": " FILTERED_PASSWORD "}'),
(
'{"password": "my_password"} {"password": "test123"}',
'{"password": " FILTERED_PASSWORD "} {"password": " FILTERED_PASSWORD "}',
),
],
)
def test_mock_password(input_text, expected_output, formatter):
assert formatter.mock_password(input_text) == expected_output
@pytest.mark.parametrize(
"input_text, expected_output",
[
('{"FLAIR": "[python]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
('{"FLAIR": "[question]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
('{"FLAIR": "[bug]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
('{"FLAIR": "[feature]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
('{"FLAIR": "[announcement]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
('{"FLAIR": "[FILTERED_FLAIR]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
],
)
def test_mock_flair(input_text, expected_output, formatter):
assert formatter.mock_flair(input_text) == expected_output
@pytest.mark.parametrize(
"input_text, expected_output",
[
(
"This is C:\\Users\\username\\file.txt",
"This is C:\\Users\\username\\file.txt",
),
("No home directory here!", "No home directory here!"),
(
f"This is {os.path.expanduser('~')}\\file.txt",
"This is MOCKING_USER_PATH\\file.txt",
),
(
f"Some path: {os.path.expanduser('~/subfolder')}",
"Some path: MOCKING_USER_PATH/subfolder",
),
],
)
def test_mock_home_directory(input_text, expected_output, formatter):
assert formatter.mock_home_directory(input_text) == expected_output
@pytest.mark.parametrize(
"input_text, expected_output",
[
(
"This is a string with a\nnewline",
"This is a string with a MOCKING_BREAKLINE newline",
),
("'Traceback occurred", "Traceback occurred"),
("No newlines or 'Traceback'", "No newlines or Traceback'"),
("'Traceback in the\nmiddle'", "Traceback in the MOCKING_BREAKLINE middle'"),
(
"Multiple\n\nnewlines",
"Multiple MOCKING_BREAKLINE MOCKING_BREAKLINE newlines",
),
],
)
def test_filter_special_tags(input_text, expected_output, formatter):
assert formatter.filter_special_tags(input_text) == expected_output
@pytest.mark.parametrize(
"input_text, expected_output",
[
(
f"This is {os.path.expanduser('~')}\\file.txt",
"This is MOCKING_USER_PATH\\file.txt",
),
('{"FLAIR": "[announcement]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
("Another email: alice@example.co.uk", "Another email: FILTERED_EMAIL "),
("Another IP: 172.16.32.1", "Another IP: FILTERED_IP "),
('{"password": "secure_pass_123"}', '{"password": " FILTERED_PASSWORD "}'),
],
)
def test_filter_piis(input_text, expected_output, formatter):
assert formatter.filter_piis(input_text) == expected_output
@pytest.mark.parametrize(
"input_text, expected_output",
[
(
f"This is {os.path.expanduser('~')}\\file.txt",
"This is MOCKING_USER_PATH\\file.txt",
),
('{"FLAIR": "[announcement]"}', '{"FLAIR": "[ FILTERED_FLAIR ]"}'),
("Another email: alice@example.co.uk", "Another email: FILTERED_EMAIL "),
("Another IP: 172.16.32.1", "Another IP: FILTERED_IP "),
('{"password": "secure_pass_123"}', '{"password": " FILTERED_PASSWORD "}'),
("'Traceback in the\nmiddle'", "Traceback in the MOCKING_BREAKLINE middle'"),
],
)
def test_filter_log_line(input_text, expected_output, formatter):
assert formatter.filter_log_line(input_text) == expected_output
def test_formatException_invalid():
with pytest.raises(Exception):
formatter.formatException(Exception("Big bad error"))
def test_format(formatter):
# Prepare the log record
log_record = logging.LogRecord(
name="test_logger",
level=logging.INFO,
pathname="/path/to/module.py",
lineno=42,
msg="This is a test log message with | symbol",
args=(),
exc_info=None,
)
# Set up the mock objects
formatter.calculate_level_name = Mock(return_value="INFO")
formatter.filter_log_line = Mock(return_value="Filtered log message")
# Call the format method
formatted_log = formatter.format(log_record)
# Assertions
expected_log = "INFO|test_app_name|test_commit_hash|test_app_id|test_session_id|test_user_id|Filtered log message"
print(expected_log)
print(formatted_log)
assert formatted_log == expected_log
# Check if the mock methods were called
formatter.calculate_level_name.assert_called_once()
formatter.filter_log_line.assert_called_once()

View File

@@ -0,0 +1,64 @@
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from openbb_core.app.logs.handlers.path_tracking_file_handler import (
PathTrackingFileHandler,
)
class MockLoggingSettings:
def __init__(self, app_name, user_logs_directory, session_id, frequency):
self.app_name = app_name
self.user_logs_directory = Path(user_logs_directory)
self.session_id = session_id
self.frequency = frequency
logging_settings = MagicMock(spec=MockLoggingSettings)
logging_settings.app_name = "test_app_name_xpto"
logging_settings.user_logs_directory = MagicMock()
logging_settings.user_logs_directory.absolute.return_value = Path(
"/mocked/logs/directory"
)
logging_settings.session_id = "test_session_id"
logging_settings.frequency = "H"
@pytest.fixture
def handler():
# patch `pathlib.Path.joinpath` to return a string containing the joined path
with patch.object(Path, "joinpath", return_value=Path("mocked_path")):
return PathTrackingFileHandler(logging_settings)
def test_build_log_file_path(handler):
# Define a sample LoggingSettings object with mock attributes
settings = MagicMock(spec=MockLoggingSettings)
settings.app_name = "my_app"
settings.user_logs_directory = MagicMock()
settings.user_logs_directory.absolute.return_value = Path("/mocked/logs/directory")
settings.session_id = "abc123"
# patch `pathlib.Path.joinpath` to return a string containing the joined path
with patch.object(
Path, "joinpath", return_value=Path("mocked_path")
) as mock_joinpath:
result_path = handler.build_log_file_path(settings)
# Assert the result is correct
assert result_path == Path("mocked_path")
mock_joinpath.assert_called_once_with("my_app_abc123")
def test_clean_expired_files(handler):
with patch(
"openbb_core.app.logs.handlers.path_tracking_file_handler.get_expired_file_list"
) as mock_get_expired_file_list, patch(
"openbb_core.app.logs.handlers.path_tracking_file_handler.remove_file_list"
) as mock_remove_file_list:
handler.clean_expired_files(123)
assert mock_get_expired_file_list.call_count == 3
assert mock_remove_file_list.call_count == 3

View File

@@ -0,0 +1,225 @@
import logging
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from openbb_core.app.logs.handlers.posthog_handler import (
PosthogHandler,
)
class MockLoggingSettings:
def __init__(
self,
app_name,
user_logs_directory,
session_id,
frequency,
appid,
commit_hash,
platform,
python_version,
terminal_version,
branch,
userid,
):
self.app_name = app_name
self.user_logs_directory = Path(user_logs_directory)
self.session_id = session_id
self.frequency = frequency
self.app_id = appid
self.commit_hash = commit_hash
self.platform = platform
self.python_version = python_version
self.terminal_version = terminal_version
self.branch = branch
self.user_id = userid
logging_settings = MagicMock(spec=MockLoggingSettings)
logging_settings.app_name = "TestApp"
logging_settings.user_logs_directory = MagicMock()
logging_settings.user_logs_directory.absolute.return_value = Path(
"/mocked/logs/directory"
)
logging_settings.session_id = "session123"
logging_settings.frequency = "H"
logging_settings.app_id = "test123"
logging_settings.commit_hash = "commit123"
logging_settings.platform = "Windows"
logging_settings.python_version = "3.9"
logging_settings.terminal_version = "1.2.3"
logging_settings.branch = "main"
logging_settings.user_id = "user123"
@pytest.fixture
def handler():
return PosthogHandler(logging_settings)
def test_emit_calls_send(mocker, handler):
# Arrange
record = logging.LogRecord(
name="test_logger",
level=logging.INFO,
pathname="test.py",
lineno=42,
msg="Test message",
args=None,
exc_info=None,
)
# Mock the send method
handler.send = mocker.MagicMock()
# Act
handler.emit(record)
# Assert
handler.send.assert_called_once_with(record=record)
def test_emit_calls_handleError_when_send_raises_exception(mocker, handler):
# Arrange
record = logging.LogRecord(
name="test_logger",
level=logging.ERROR,
pathname="test.py",
lineno=42,
msg="Test error message",
args=None,
exc_info=None,
)
# Mock the send method to raise an exception
handler.send = mocker.MagicMock(side_effect=Exception)
# Mock the handleError method
handler.handleError = mocker.MagicMock()
# Act
handler.emit(record)
# Assert
handler.send.assert_called_once_with(record=record)
handler.handleError.assert_called_once_with(record)
def test_emit_calls_handleError_when_send_raises_exception_of_specific_type(
mocker, handler
):
# Arrange
record = logging.LogRecord(
name="test_logger",
level=logging.ERROR,
pathname="test.py",
lineno=42,
msg="Test error message",
args=None,
exc_info=None,
)
# Mock the send method to raise an exception of a specific type
handler.send = mocker.MagicMock(side_effect=ValueError)
# Mock the handleError method
handler.handleError = mocker.MagicMock()
# Act
handler.emit(record)
# Assert
handler.send.assert_called_once_with(record=record)
handler.handleError.assert_called_once_with(record)
def test_emit_calls_handleError_when_send_raises_exception_of_another_type(
mocker, handler
):
# Arrange
record = logging.LogRecord(
name="test_logger",
level=logging.ERROR,
pathname="test.py",
lineno=42,
msg="Test error message",
args=None,
exc_info=None,
)
# Mock the send method to raise an exception of another type
handler.send = mocker.MagicMock(side_effect=TypeError)
# Mock the handleError method
handler.handleError = mocker.MagicMock()
# Act
handler.emit(record)
# Assert
handler.send.assert_called_once_with(record=record)
handler.handleError.assert_called_once_with(record)
@pytest.mark.parametrize(
"log_info, expected_dict",
[
(
'STARTUP: {"status": "success"}',
{"STARTUP": {"status": "success"}},
),
(
'{"INPUT": {"something_something":"something_something"}}',
{
"SDK": {"INPUT": {"something_something": "something_something"}},
},
),
(
'CMD: {"path": "/stocks/", "known_cmd": "load", "other_args": "aapl", "input": "load aapl"}',
{
"CMD": {
"path": "/stocks/",
"known_cmd": "load",
"other_args": "aapl",
"input": "load aapl",
}
},
),
],
)
def test_log_to_dict(handler, log_info, expected_dict):
# Act
result = handler.log_to_dict(log_info)
# Assert
assert result == expected_dict
@pytest.mark.parametrize(
"record, expected_extra",
[
(
logging.LogRecord(
"name", logging.INFO, "pathname", 42, "message", (), None, None
),
{
"appName": "TestApp",
"appId": "test123",
"sessionId": "session123",
"commitHash": "commit123",
"platform": "Windows",
"pythonVersion": "3.9",
"terminalVersion": "1.2.3",
"branch": "main",
"userId": "user123",
},
),
],
)
def test_extract_log_extra(handler, record, expected_extra):
# Act
result = handler.extract_log_extra(record)
# Assert
assert result == expected_extra

View File

@@ -0,0 +1,74 @@
import logging
from unittest.mock import Mock, patch
from openbb_core.app.logs.handlers_manager import HandlersManager
class MockPosthogHandler(logging.NullHandler):
def __init__(self, settings):
self.settings = settings
class MockPathTrackingFileHandler(logging.NullHandler):
def __init__(self, settings):
self.settings = settings
class MockFormatterWithExceptions(logging.Formatter):
def __init__(self, settings):
self.settings = settings
def test_handlers_added_correctly():
with patch(
"openbb_core.app.logs.handlers_manager.PosthogHandler",
MockPosthogHandler,
), patch(
"openbb_core.app.logs.handlers_manager.PathTrackingFileHandler",
MockPathTrackingFileHandler,
), patch(
"openbb_core.app.logs.handlers_manager.FormatterWithExceptions",
MockFormatterWithExceptions,
):
settings = Mock()
settings.handler_list = ["stdout", "stderr", "noop", "file", "posthog"]
_ = HandlersManager(settings=settings)
handlers = logging.getLogger().handlers
assert len(handlers) >= 5
for handler in handlers:
assert isinstance(handler, (logging.NullHandler, logging.StreamHandler))
for mock in [MockPosthogHandler, MockPathTrackingFileHandler]:
assert any(isinstance(handler, mock) for handler in handlers)
def test_update_handlers():
with patch(
"openbb_core.app.logs.handlers_manager.PosthogHandler",
MockPosthogHandler,
), patch(
"openbb_core.app.logs.handlers_manager.PathTrackingFileHandler",
MockPathTrackingFileHandler,
), patch(
"openbb_core.app.logs.handlers_manager.FormatterWithExceptions",
MockFormatterWithExceptions,
):
settings = Mock()
settings.handler_list = ["file", "posthog"]
settings.any_other_attr = "mock_settings"
handlers_manager = HandlersManager(settings=settings)
changed_settings = Mock()
changed_settings.any_other_attr = "changed_settings"
handlers_manager.update_handlers(settings=changed_settings)
handlers = logging.getLogger().handlers
for hdlr in handlers:
if isinstance(hdlr, (MockPosthogHandler, MockPathTrackingFileHandler)):
assert hdlr.settings == changed_settings
assert hdlr.formatter.settings == changed_settings

View File

@@ -0,0 +1,188 @@
import json
from typing import Optional
from unittest.mock import MagicMock, Mock, patch
import pytest
from openbb_core.app.logs.logging_manager import LoggingManager
from pydantic import BaseModel
class MockLoggingSettings:
def __init__(self, system_settings, user_settings):
self.system_settings = system_settings
self.user_settings = user_settings
class MockCommandOutput(BaseModel):
output: Optional[str]
error: Optional[str]
@pytest.fixture(scope="function")
def logging_manager():
mock_system_settings = Mock()
mock_system_settings = "mock_system_settings"
mock_user_settings = Mock()
mock_user_settings = "mock_user_settings"
mock_setup_handlers = Mock()
mock_log_startup = Mock()
with patch(
"openbb_core.app.logs.logging_manager.LoggingSettings",
MockLoggingSettings,
), patch(
"openbb_core.app.logs.logging_manager.LoggingManager._setup_handlers",
mock_setup_handlers,
), patch(
"openbb_core.app.logs.logging_manager.LoggingManager._log_startup",
mock_log_startup,
):
logging_manager = LoggingManager(
system_settings=mock_system_settings,
user_settings=mock_user_settings,
)
assert mock_setup_handlers.assert_called_once
assert mock_log_startup.assert_called_once
return logging_manager
def test_correctly_initialized(logging_manager):
assert logging_manager
def test_logging_settings_property(logging_manager):
assert logging_manager.logging_settings.system_settings == "mock_system_settings"
assert logging_manager.logging_settings.user_settings == "mock_user_settings"
def test_logging_settings_setter(logging_manager):
custom_user_settings = "custom_user_settings"
custom_system_settings = "custom_system_settings"
with patch(
"openbb_core.app.logs.logging_manager.LoggingSettings",
MockLoggingSettings,
):
logging_manager.logging_settings = (
custom_system_settings,
custom_user_settings,
)
assert logging_manager.logging_settings.system_settings == "custom_system_settings"
assert logging_manager.logging_settings.user_settings == "custom_user_settings"
def test_log_startup(logging_manager):
with patch("logging.getLogger") as mock_get_logger:
mock_info = mock_get_logger.return_value.info
class MockCredentials(BaseModel):
username: str
password: str
logging_manager._user_settings = MagicMock(
preferences="your_preferences",
credentials=MockCredentials(username="username", password="password"),
)
logging_manager._system_settings = "your_system_settings"
logging_manager._log_startup()
expected_log_data = {
"PREFERENCES": "your_preferences",
"KEYS": {"username": "defined", "password": "defined"},
"SYSTEM": "your_system_settings",
}
mock_info.assert_called_once_with(
"STARTUP: %s ",
json.dumps(expected_log_data),
)
mock_get_logger.assert_called_once
@pytest.mark.parametrize(
"user_settings, system_settings, command_output, route, func, kwargs",
[
(
"mock_settings",
"mock_system",
MockCommandOutput(output="mock_output"),
"mock_route",
"mock_func",
{},
),
(
"mock_settings",
"mock_system",
MockCommandOutput(error="mock_error"),
"mock_route",
"mock_func",
{},
),
(
"mock_settings",
"mock_system",
MockCommandOutput(error="mock_error"),
"login",
"mock_func",
{},
),
],
)
def test_log(
logging_manager, user_settings, system_settings, command_output, route, func, kwargs
):
with patch(
"openbb_core.app.logs.logging_manager.LoggingSettings",
MockLoggingSettings,
), patch("logging.getLogger") as mock_get_logger:
if route == "login":
with patch(
"openbb_core.app.logs.logging_manager.LoggingManager._log_startup"
) as mock_log_startup:
logging_manager.log(
user_settings=user_settings,
system_settings=system_settings,
command_output=command_output,
route=route,
func=func,
kwargs=kwargs,
)
assert mock_log_startup.assert_called_once
else:
mock_info = mock_get_logger.return_value.info
mock_error = mock_get_logger.return_value.error
mock_callable = Mock()
mock_callable.__name__ = func
logging_manager.log(
user_settings=user_settings,
system_settings=system_settings,
command_output=command_output,
route=route,
func=mock_callable,
kwargs=kwargs,
)
expected_log_data = {
"route": route,
"input": kwargs,
"error": command_output.error,
}
if command_output.error:
mock_error.assert_called_once_with(
"ERROR: %s",
json.dumps(expected_log_data),
extra={"func_name_override": func},
)
else:
mock_info.assert_called_once_with(
"CMD: %s",
json.dumps(expected_log_data),
extra={"func_name_override": func},
)

View File

@@ -0,0 +1,148 @@
import os
import tempfile
from pathlib import Path
from time import time
from unittest.mock import MagicMock
import pytest
from openbb_core.app.logs.utils.expired_files import (
get_expired_file_list,
get_timestamp_from_x_days,
remove_file_list,
)
@pytest.fixture
def temp_test_files():
with tempfile.TemporaryDirectory() as temp_dir:
temp_files = [
Path(temp_dir) / "file1.txt",
Path(temp_dir) / "file2.txt",
Path(temp_dir) / "file3.txt",
]
# Create some test files
for file_path in temp_files:
file_path.touch()
yield temp_dir, temp_files
def test_get_timestamp_from_x_days():
result = get_timestamp_from_x_days(0)
assert isinstance(result, float)
# Test case when all files are expired
def test_all_files_expired(temp_test_files):
temp_dir, temp_files = temp_test_files
before_timestamp = time() + 3 * 86400 # timestamp 3 days from now
expired_files = get_expired_file_list(Path(temp_dir), before_timestamp)
assert set(expired_files) == set(temp_files)
# Test case when no files are expired
def test_no_files_expired(temp_test_files):
temp_dir, _ = temp_test_files
before_timestamp = 0
expired_files = get_expired_file_list(Path(temp_dir), before_timestamp)
assert expired_files == []
# Test case when some files are expired and some are not
def test_some_files_expired(temp_test_files):
temp_dir, _ = temp_test_files
# add temp file to temp_dir with timestamp in the future
temp_file = Path(temp_dir) / "file4.txt"
temp_file.touch()
time_in_future = time() + 4 * 86400 # timestamp 4 days from now
os.utime(temp_file, times=(time_in_future, time_in_future))
before_timestamp = time() + 3 * 86400 # timestamp 3 days from now
expired_files = get_expired_file_list(Path(temp_dir), before_timestamp)
assert len(expired_files) == 3
# assert number of files in temp_dir is 4
assert len(list(Path(temp_dir).iterdir())) == 4
# Test case when the directory does not exist
def test_directory_not_exists():
directory = Path("/path/that/does/not/exist")
before_timestamp = time()
expired_files = get_expired_file_list(directory, before_timestamp)
assert expired_files == []
# Test case when the directory is not a directory
def test_directory_not_dir(temp_test_files):
_, temp_files = temp_test_files
before_timestamp = time()
expired_files = get_expired_file_list(temp_files[0], before_timestamp)
assert expired_files == []
@pytest.fixture
def mock_path():
# Create a MagicMock for the Path class to represent a file path
return MagicMock()
def test_remove_file_list_no_files(mock_path, mocker):
# Arrange
# Let's assume the file list is empty, meaning there are no files to remove
file_list = []
# Act
remove_file_list(file_list)
# Assert
# No interaction with the filesystem should occur
assert not mock_path.unlink.called
def test_remove_file_list_remove_files_successfully(mock_path, mocker):
# Arrange
# Let's assume we have three files in the file list
file_list = [mock_path, mock_path, mock_path]
# Mock the unlink method to avoid actual filesystem interaction
mocker.patch.object(mock_path, "unlink")
# Act
remove_file_list(file_list)
# Assert
# unlink should have been called three times since there are three files in the list
assert mock_path.unlink.call_count == 3
def test_remove_file_list_ignore_permission_error(mock_path, mocker):
# Arrange
# Let's assume we have three files in the file list
file_list = [mock_path, mock_path, mock_path]
# Mock the unlink method to raise a PermissionError
mocker.patch.object(mock_path, "unlink", side_effect=PermissionError)
# Act
remove_file_list(file_list)
# Assert
# unlink should have been called three times since there are three files in the list
assert mock_path.unlink.call_count == 3
def test_remove_file_list_other_exception(mock_path, mocker):
# Arrange
# Let's assume we have three files in the file list
file_list = [mock_path, mock_path, mock_path]
# Mock the unlink method to raise an exception other than PermissionError
mocker.patch.object(mock_path, "unlink", side_effect=OSError)
# Act & Assert
# The exception should be raised since it is not suppressed by the contextlib
with pytest.raises(OSError):
remove_file_list(file_list)

View File

@@ -0,0 +1,9 @@
# TODO : add tests once the code is fixed
def test_get_commit_hash():
pass
def test_get_branch():
pass

View File

@@ -0,0 +1,93 @@
import uuid
from pathlib import Path
from unittest.mock import patch
import pytest
from openbb_core.app.logs.utils.utils import get_app_id, get_log_dir, get_session_id
## get_session_id
def test_get_session_id_return_type():
# Test if the returned value is a string
session_id = get_session_id()
assert isinstance(session_id, str)
def test_get_session_id_format():
# Test if the returned string has the format "UUID-current_time"
session_id = get_session_id()
parts = session_id.split("-")
assert len(parts) == 6
uuid_part = "-".join(parts[0:5])
time_part = str(parts[5])
# Check if the first part (UUID) is a valid UUID
assert uuid.UUID(uuid_part)
# Check if the second part (current_time) is numeric
assert int(time_part)
def test_get_session_id_uniqueness():
# Test if subsequent calls return different session IDs
session_id1 = get_session_id()
session_id2 = get_session_id()
assert session_id1 != session_id2
## get_app_id
def test_get_app_id_success():
# Mock the return value of get_log_dir to simulate a successful scenario
with patch("openbb_core.app.logs.utils.utils.get_log_dir") as mock_get_log_dir:
mock_get_log_dir
mock_get_log_dir.return_value = Path(
"/path/to/contextual_user_data_directory/app_id.log"
)
app_id = get_app_id("/path/to/contextual_user_data_directory")
assert app_id == "app_id"
def test_get_app_id_os_error():
# Test handling of OSError with errno 30 (Read-only file system)
with patch("openbb_core.app.logs.utils.utils.get_log_dir") as mock_get_log_dir:
mock_get_log_dir.side_effect = OSError(30, "Read-only file system")
with pytest.raises(OSError):
get_app_id("/path/to/contextual_user_data_directory")
def test_get_app_id_other_exception():
# Test handling of other exceptions
with patch("openbb_core.app.logs.utils.utils.get_log_dir") as mock_get_log_dir:
mock_get_log_dir.side_effect = Exception("Some other error")
with pytest.raises(Exception, match="Some other error"):
get_app_id("/path/to/contextual_user_data_directory")
## get_log_dir
def test_get_log_dir():
with patch(
"openbb_core.app.logs.utils.utils.create_log_dir_if_not_exists",
return_value="/test_dir",
) as mock_create_log_dir, patch(
"openbb_core.app.logs.utils.utils.create_log_uuid_if_not_exists",
return_value="12345",
) as mock_create_log_uuid, patch(
"openbb_core.app.logs.utils.utils.create_uuid_dir_if_not_exists",
return_value="/test_dir/12345",
) as mock_create_uuid_dir:
# Call the get_log_dir function
result = get_log_dir("contextual_user_data_directory")
# Assertions
assert result == "/test_dir/12345"
mock_create_log_dir.assert_called_once_with("contextual_user_data_directory")
mock_create_log_uuid.assert_called_once_with("/test_dir")
mock_create_uuid_dir.assert_called_once_with("/test_dir", "12345")

View File

@@ -0,0 +1,24 @@
import pytest
from openbb_core.app.model.abstract.error import Error
@pytest.mark.parametrize(
"error_kind, message",
[
("test", "test"),
("test2", "test2"),
],
)
def test_error_model(error_kind, message):
err = Error(error_kind=error_kind, message=message)
assert err.error_kind == error_kind
assert err.message == message
def test_fields():
fields = Error.__fields__
fields_keys = fields.keys()
assert "error_kind" in fields_keys
assert "message" in fields_keys

View File

@@ -0,0 +1,8 @@
from openbb_core.app.model.abstract.results import Results
from pydantic import BaseModel
def test_results_model():
res = Results()
assert isinstance(res, BaseModel)

View File

@@ -0,0 +1,32 @@
from openbb_core.app.model.abstract.singleton import SingletonMeta
class MyClass(metaclass=SingletonMeta):
def __init__(self, value):
self.value = value
def test_singleton_instance_creation():
# Arrange
instance1 = MyClass(42)
instance2 = MyClass(100)
# Act & Assert
assert instance1 is instance2
assert instance1.value == instance2.value
assert instance1.value == 42
def test_singleton_multiple_classes():
# Arrange
class AnotherClass(metaclass=SingletonMeta):
def __init__(self, data):
self.data = data
instance1 = MyClass(42)
instance2 = AnotherClass("test")
# Act & Assert
assert instance1 is not instance2
assert instance1.value == 42
assert instance2.data == "test"

View File

@@ -0,0 +1,14 @@
from openbb_core.app.model.abstract.tagged import Tagged
def test_tagged_model():
tagged = Tagged()
assert hasattr(tagged, "id")
def test_fields():
fields = Tagged.__fields__
fields_keys = fields.keys()
assert "id" in fields_keys

View File

@@ -0,0 +1,36 @@
from unittest.mock import Mock
import pytest
from openbb_core.app.model.abstract.warning import Warning_, cast_warning
@pytest.mark.parametrize(
"category, message",
[
("test", "test"),
("test2", "test2"),
],
)
def test_warn_model(category, message):
war = Warning_(category=category, message=message)
assert war.category == category
assert war.message == message
def test_fields():
fields = Warning_.__fields__
fields_keys = fields.keys()
assert "category" in fields_keys
assert "message" in fields_keys
def test_cast_warning():
mock_warning_message = Mock()
mock_warning_message.category.__name__ = "test"
mock_warning_message.message = "test"
warning = cast_warning(mock_warning_message)
assert warning.category == "test"
assert warning.message == "test"

View File

@@ -0,0 +1,55 @@
import pytest
from openbb_core.app.model.charts.chart import Chart, ChartFormat, Error
def test_chart_default_values():
# Arrange & Act
chart = Chart()
# Assert
assert chart.content is None
assert chart.format == ChartFormat.plotly
assert chart.error is None
def test_chart_custom_values():
# Arrange
content = {"data": [1, 2, 3]}
chart_format = ChartFormat.plotly
error = Error(message="Test Error")
# Act
chart = Chart(content=content, format=chart_format, error=error)
# Assert
assert chart.content == content
assert chart.format == chart_format
assert chart.error == error
def test_chart_assignment_validation():
# Arrange
chart = Chart()
# Act & Assert
with pytest.raises(ValueError):
chart.invalid_field = "Invalid Value"
def test_chart_config_validation():
# Arrange
content = {"data": [1, 2, 3]}
chart_format = ChartFormat.plotly
chart = Chart(content=content, format=chart_format)
with pytest.raises(ValueError):
chart.content = "Invalid Content"
assert chart.content == content
assert chart.format == chart_format
def test_show():
# TODO : add test after the function is properly refactored
pass

View File

@@ -0,0 +1,49 @@
from unittest.mock import MagicMock
from openbb_core.app.model.charts.charting_settings import ChartingSettings
mock_user_settings = MagicMock()
mock_user_settings.preferences.data_directory = "mock_data_directory"
mock_user_settings.preferences.plot_enable_pywry = True
mock_user_settings.preferences.plot_pywry_width = 100
mock_user_settings.preferences.plot_pywry_height = 100
mock_user_settings.preferences.plot_open_export = True
mock_user_settings.profile.hub_session.email = "mock_email"
mock_user_settings.profile.hub_session.user_uuid = "mock_uuid"
mock_user_settings.preferences.export_directory = "mock_export_directory"
mock_user_settings.preferences.user_styles_directory = "mock_styles_directory"
mock_user_settings.preferences.chart_style = "mock_chart_style"
mock_system_settings = MagicMock()
mock_system_settings.log_collect = True
mock_system_settings.version = "mock_version"
mock_system_settings.python_version = "mock_python_version"
mock_system_settings.test_mode = True
mock_system_settings.debug_mode = True
mock_system_settings.headless = True
def test_charting_settings(mocker):
mocker.patch(
target="openbb_core.app.model.charts.charting_settings.get_app_id",
new=mocker.Mock(return_value="mock_app_id"),
)
charting_settings = ChartingSettings(
user_settings=mock_user_settings, system_settings=mock_system_settings
)
assert charting_settings.log_collect is True
assert charting_settings.version == "mock_version"
assert charting_settings.python_version == "mock_python_version"
assert charting_settings.test_mode is True
assert charting_settings.app_id == "mock_app_id"
assert charting_settings.debug_mode is True
assert charting_settings.headless is True
assert charting_settings.plot_enable_pywry is True
assert charting_settings.plot_pywry_width == 100
assert charting_settings.plot_pywry_height == 100
assert charting_settings.plot_open_export is True
assert charting_settings.user_email == "mock_email"
assert charting_settings.user_uuid == "mock_uuid"
assert charting_settings.user_exports_directory == "mock_export_directory"
assert charting_settings.user_styles_directory == "mock_styles_directory"
assert charting_settings.chart_style == "mock_chart_style"

View File

@@ -0,0 +1,8 @@
from openbb_core.app.model.hub.features_keys import FeaturesKeys
def test_features_keys():
fk = FeaturesKeys(API_BITQUERY_KEY="test", API_BIZTOC_TOKEN="test2")
assert fk.API_BITQUERY_KEY == "test"
assert fk.API_BIZTOC_TOKEN == "test2"

View File

@@ -0,0 +1,30 @@
from openbb_core.app.model.hub.hub_session import HubSession
def test_hub_session():
session = HubSession(
access_token="mock_access_token",
token_type="mock_token_type",
email="mock_email",
user_uuid="mock_user_uuid",
username="mock_username",
primary_usage="mock_primary_usage",
)
assert session.access_token == "mock_access_token"
assert session.token_type == "mock_token_type"
assert session.email == "mock_email"
assert session.user_uuid == "mock_user_uuid"
assert session.username == "mock_username"
assert session.primary_usage == "mock_primary_usage"
def test_fields():
fields = HubSession.__fields__
fields_keys = fields.keys()
assert "access_token" in fields_keys
assert "token_type" in fields_keys
assert "email" in fields_keys
assert "user_uuid" in fields_keys
assert "username" in fields_keys
assert "primary_usage" in fields_keys

View File

@@ -0,0 +1,7 @@
from openbb_core.app.model.hub.hub_user_settings import HubUserSettings
def test_hub_user_settings():
hub_settings = HubUserSettings()
assert isinstance(hub_settings, HubUserSettings)

View File

@@ -0,0 +1,20 @@
from openbb_core.app.model.command_context import (
CommandContext,
SystemSettings,
UserSettings,
)
def test_command_context():
cc = CommandContext()
assert isinstance(cc, CommandContext)
assert isinstance(cc.user_settings, UserSettings)
assert isinstance(cc.system_settings, SystemSettings)
def test_fields():
fields = CommandContext.__fields__
fields_keys = fields.keys()
assert "user_settings" in fields_keys
assert "system_settings" in fields_keys

View File

@@ -0,0 +1,182 @@
from unittest.mock import MagicMock, patch
import pandas as pd
import pytest
from openbb_core.app.model.command_output import Chart, CommandOutput, OpenBBError
def test_command_output():
co = CommandOutput()
assert isinstance(co, CommandOutput)
def test_fields():
fields = CommandOutput.__fields__.keys()
assert "results" in fields
assert "provider" in fields
assert "warnings" in fields
assert "error" in fields
assert "chart" in fields
def test_to_dataframe_no_results():
co = CommandOutput()
with pytest.raises(Exception):
co.to_dataframe()
@pytest.mark.parametrize(
"results, expected_df",
[
# Test case 1: Normal results with "date" column
(
[{"date": "2023-07-30", "value": 10}, {"date": "2023-07-31", "value": 20}],
pd.DataFrame(
{"value": [10, 20]},
index=pd.to_datetime(["2023-07-30", "2023-07-31"]),
),
),
# Test case 2: Normal results without "date" column
(
[{"value": 10}, {"value": 20}],
pd.DataFrame({"value": [10, 20]}, index=pd.RangeIndex(start=0, stop=2)),
),
# Test case 3: Empty results
([], pd.DataFrame()),
# Test case 4: Results as None, should raise OpenBBError
(None, OpenBBError("Results not found.")),
],
)
def test_to_dataframe(results, expected_df):
# Arrange
co = CommandOutput(results=results)
# Act and Assert
if isinstance(expected_df, pd.DataFrame):
result = co.to_dataframe()
assert result.equals(expected_df)
else:
with pytest.raises(expected_df.__class__) as exc_info:
co.to_dataframe()
assert str(exc_info.value) == str(expected_df)
@pytest.mark.parametrize(
"results, expected",
[
# Test case 1: Normal results with "date" column
(
[{"date": "2023-07-30", "value": 10}, {"date": "2023-07-31", "value": 20}],
{
"date": [pd.to_datetime("2023-07-30"), pd.to_datetime("2023-07-31")],
"value": [10, 20],
},
),
# Test case 2: Normal results without "date" column
(
[{"value": 10}, {"value": 20}],
{
"value": [10, 20],
},
),
# Test case 3: Empty results
([], {}),
],
)
def test_to_dict(results, expected):
# Arrange
co = CommandOutput(results=results)
# Act
result = co.to_dict()
result.pop("index", None)
# Assert
assert result == expected
@patch("openbb_core.app.model.command_output.ChartingManager")
@patch("openbb_core.app.model.command_output.Chart")
def test_to_plotly_json_with_existing_chart(mock_chart, mock_charting_manager):
# Arrange
mock_instance = CommandOutput()
mock_instance.chart = Chart(content={"existing_chart_data": "some_data"})
# Act
result = mock_instance.to_plotly_json()
# Assert
assert result == {"existing_chart_data": "some_data"}
# Ensure ChartingManager and Chart constructors were not called
mock_charting_manager.assert_not_called()
mock_chart.assert_not_called()
@patch("openbb_core.app.model.command_output.CommandOutput.to_dataframe")
@patch("openbb_core.app.model.command_output.ChartingManager")
@patch("openbb_core.app.model.command_output.Chart")
def test_to_plotly_json_with_new_chart(
mock_chart,
mock_charting_manager,
mock_to_dataframe,
):
def get_mock_dataframe():
data = {
"col1": [1, 2, 3],
"col2": ["a", "b", "c"],
"col3": [True, False, True],
}
return pd.DataFrame(data)
# Arrange
mock_instance = CommandOutput()
mock_charting_manager_instance = mock_charting_manager.return_value
mock_charting_manager_instance.to_plotly_json.return_value = {
"new_chart_data": "some_data"
}
mock_to_dataframe.return_value = get_mock_dataframe()
# Act
result = mock_instance.to_plotly_json(
create_chart=True, chart_option="option_value"
)
# Assert
assert result == {"new_chart_data": "some_data"}
# Ensure self.to_dataframe() was called
mock_to_dataframe.assert_called_once()
# Ensure ChartingManager was called with the correct parameters
mock_charting_manager_instance.to_plotly_json.assert_called_once()
# Ensure Chart constructor was called with the correct parameters
mock_chart.assert_called_once_with(
content={"new_chart_data": "some_data"}, format="plotly"
)
def test_show_chart_exists():
mock_instance = CommandOutput()
# Arrange
mock_instance.chart = MagicMock(spec=Chart)
# Act
mock_instance.show()
# Assert
mock_instance.chart.show.assert_called_once()
def test_show_chart_not_found():
mock_instance = CommandOutput()
# Arrange
mock_instance.chart = None
# Act and Assert
with pytest.raises(OpenBBError, match="Chart not found."):
mock_instance.show()

View File

@@ -0,0 +1,22 @@
import typing
from unittest.mock import patch
def test_credentials():
with patch(
target="openbb_core.app.model.credentials.get_provider_interface"
) as mock_provider_interface:
mock_provider_interface.required_credentials = {
"benzinga_api_key": (typing.Optional[str], None),
"polygon_api_key": (typing.Optional[str], None),
}
from openbb_core.app.model.credentials import Credentials
creds = Credentials(
benzinga_api_key="mock_benzinga_api_key",
polygon_api_key="mock_polygon_api_key",
)
assert isinstance(creds, Credentials)
assert creds.benzinga_api_key == "mock_benzinga_api_key"
assert creds.polygon_api_key == "mock_polygon_api_key"

View File

@@ -0,0 +1,14 @@
from openbb_core.app.model.defaults import Defaults
def test_defaultst():
cc = Defaults(routes={"test": {"test": "test"}})
assert isinstance(cc, Defaults)
assert cc.routes == {"test": {"test": "test"}}
def test_fields():
fields = Defaults.__fields__
fields_keys = fields.keys()
assert "routes" in fields_keys

View File

@@ -0,0 +1,19 @@
from openbb_core.app.model.journal import Journal
def test_journal_query():
journal_query = Journal(
name="mock_name",
journal_entry_id_list=["id_1", "id_2"],
)
assert isinstance(journal_query, Journal)
assert journal_query.name == "mock_name"
assert journal_query.journal_entry_id_list == ["id_1", "id_2"]
def test_fields():
fields = Journal.__fields__
fields_keys = fields.keys()
assert "name" in fields_keys
assert "journal_entry_id_list" in fields_keys

View File

@@ -0,0 +1,30 @@
import datetime
from openbb_core.app.model.journal_entry import JournalEntry
def test_journal_entry():
journal = JournalEntry(
journal_id="mock_journal_id",
arguments={"mock_key": "mock_value"},
duration=19,
output={},
route="mock_route",
timestamp=datetime.datetime(2021, 1, 1),
alias_list=["mock_alias"],
)
assert isinstance(journal, JournalEntry)
def test_fields():
fields = JournalEntry.__fields__
fields_keys = fields.keys()
assert "id" in fields_keys
assert "journal_id" in fields_keys
assert "arguments" in fields_keys
assert "duration" in fields_keys
assert "output" in fields_keys
assert "route" in fields_keys
assert "timestamp" in fields_keys
assert "alias_list" in fields_keys

View File

@@ -0,0 +1,19 @@
from openbb_core.app.model.journal_query import JournalQuery
def test_journal_query():
journal_query = JournalQuery(
journal_entry_id="mock_journal_entry_id",
ignore_error_output=True,
)
assert isinstance(journal_query, JournalQuery)
assert journal_query.journal_entry_id == "mock_journal_entry_id"
assert journal_query.ignore_error_output is True
def test_fields():
fields = JournalQuery.__fields__
fields_keys = fields.keys()
assert "journal_entry_id" in fields_keys
assert "ignore_error_output" in fields_keys

View File

@@ -0,0 +1,6 @@
from openbb_core.app.model.preferences import Preferences
def test_preferences():
preferences = Preferences()
assert isinstance(preferences, Preferences)

View File

@@ -0,0 +1,6 @@
from openbb_core.app.model.profile import Profile
def test_preferences():
preferences = Profile()
assert isinstance(preferences, Profile)

View File

@@ -0,0 +1,203 @@
import os
from unittest.mock import patch
import pytest
from openbb_core.app.model.system_settings import SystemSettings
def test_system_settings():
sys = SystemSettings()
assert isinstance(sys, SystemSettings)
def test_create_openbb_directory_directory_and_files_not_exist(tmpdir):
# Arrange
values = {
"openbb_directory": str(tmpdir.join("openbb")),
"user_settings_path": str(tmpdir.join("user_settings.json")),
"system_settings_path": str(tmpdir.join("system_settings.json")),
}
# Act
SystemSettings.create_openbb_directory(values)
# Assert
assert os.path.exists(values["openbb_directory"])
assert os.path.exists(values["user_settings_path"])
assert os.path.exists(values["system_settings_path"])
def test_create_openbb_directory_directory_exists_user_settings_missing(tmpdir):
# Arrange
values = {
"openbb_directory": str(tmpdir.join("openbb")),
"user_settings_path": str(tmpdir.join("user_settings.json")),
"system_settings_path": str(tmpdir.join("system_settings.json")),
}
# Create the openbb directory
os.makedirs(values["openbb_directory"])
# Act
SystemSettings.create_openbb_directory(values)
# Assert
assert os.path.exists(values["openbb_directory"])
assert os.path.exists(values["user_settings_path"])
assert os.path.exists(values["system_settings_path"])
def test_create_openbb_directory_directory_exists_system_settings_missing(tmpdir):
# Arrange
values = {
"openbb_directory": str(tmpdir.join("openbb")),
"user_settings_path": str(tmpdir.join("user_settings.json")),
"system_settings_path": str(tmpdir.join("system_settings.json")),
}
# Create the openbb directory
os.makedirs(values["openbb_directory"])
# Create the user_settings.json file
with open(values["user_settings_path"], "w") as f:
f.write("{}")
# Act
SystemSettings.create_openbb_directory(values)
# Assert
assert os.path.exists(values["openbb_directory"])
assert os.path.exists(values["user_settings_path"])
assert os.path.exists(values["system_settings_path"])
@pytest.mark.parametrize(
"values, expected_handlers",
[
# Test case: test_mode is True, logging_suppress is True
(
{
"test_mode": True,
"logging_suppress": True,
"log_collect": True,
"logging_handlers": [],
},
[],
),
# Test case: test_mode is False, logging_suppress is True
(
{
"test_mode": False,
"logging_suppress": True,
"log_collect": True,
"logging_handlers": [],
},
[],
),
# Test case: test_mode is False, logging_suppress is False, log_collect is True,
# and "posthog" handler is not present in logging_handlers
(
{
"test_mode": False,
"logging_suppress": False,
"log_collect": True,
"logging_handlers": ["file", "console"],
},
["file", "console", "posthog"],
),
# Test case: test_mode is False, logging_suppress is False, log_collect is True,
# and "posthog" handler is already present in logging_handlers
(
{
"test_mode": False,
"logging_suppress": False,
"log_collect": True,
"logging_handlers": ["file", "console", "posthog"],
},
["file", "console", "posthog"],
),
],
)
def test_validate_posthog_handler(values, expected_handlers):
# Act
result = SystemSettings.validate_posthog_handler(values)
# Assert
assert result["logging_handlers"] == expected_handlers
@pytest.mark.parametrize(
"handlers, valid",
[
# Test case: Valid handlers
(["stdout", "file", "noop"], True),
# Test case: Invalid handler
(["stdout", "invalid_handler", "file"], False),
# Test case: Empty list of handlers
([], True),
# Test case: Repeated valid handlers
(["stdout", "stderr", "stdout", "noop", "stderr"], True),
],
)
def test_validate_logging_handlers(handlers, valid):
# Act and Assert
if valid:
assert SystemSettings.validate_logging_handlers(handlers) == handlers
else:
with pytest.raises(ValueError, match="Invalid logging handler"):
SystemSettings.validate_logging_handlers(handlers)
@pytest.mark.parametrize(
"commit_hash, expected_result",
[
# Test case: Valid commit hash provided
("1234567890", "1234567890"),
# Test case: Empty commit hash, mocked get_commit_hash returns "abcdef1234"
("", "abcdef1234"),
],
)
def test_validate_commit_hash(commit_hash, expected_result):
with patch(
"openbb_core.app.model.system_settings.get_commit_hash",
return_value="abcdef1234",
):
# Act
result = SystemSettings.validate_commit_hash(commit_hash)
# Assert
assert result == expected_result
@patch("openbb_core.app.model.system_settings.get_branch", return_value="main")
@pytest.mark.parametrize(
"branch, commit_hash, expected_branch",
[
# Test case: Empty branch, non-empty commit hash
("", "abcdef1234", "main"),
# Test case: Non-empty branch, non-empty commit hash
("feature-branch", "abcdef1234", "feature-branch"),
# Test case: Empty branch, empty commit hash
("", "", ""),
# Test case: Non-empty branch, empty commit hash
("feature-branch", "", "feature-branch"),
],
)
def test_validate_branch(mock_get_branch, branch, commit_hash, expected_branch):
# Arrange
values = {
"logging_branch": branch,
"logging_commit_hash": commit_hash,
}
# Act
result = SystemSettings.validate_branch(values)
# Assert
assert result["logging_branch"] == expected_branch
# Ensure that get_branch was called with the correct commit_hash
if not branch and commit_hash:
mock_get_branch.assert_called_once_with(commit_hash)
else:
mock_get_branch.assert_not_called()

View File

@@ -0,0 +1,13 @@
from unittest.mock import MagicMock
from openbb_core.app.model.user_settings import UserSettings
def test_user_settings():
settings = UserSettings(
credentials=MagicMock(),
profile=MagicMock(),
preferences=MagicMock(),
defaults=MagicMock(),
)
assert isinstance(settings, UserSettings)

View File

@@ -0,0 +1,56 @@
from typing import Any, List, Optional, Tuple, TypeVar
from openbb_core.app.repository.abstract.access_token_repository import (
AccessTokenRepository,
)
from pydantic import BaseModel
T = TypeVar("T", bound=BaseModel)
def test_journal_entry_repository():
class TestAccessTokenRepository(AccessTokenRepository):
def __init__(self, test_arg_1, test_arg_2):
self.test_arg_1 = test_arg_1
self.test_arg_2 = test_arg_2
super().__init__()
@property
def collection_name(self):
return "test_collection"
def create(self, model):
pass
@property
def database_name(self):
return "test_database"
def delete(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def delete_all(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def read(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def read_all(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def update(self, model: T):
pass
def mock_method(self):
return self.test_arg_1, self.test_arg_2
jr = TestAccessTokenRepository("xpto_1", "xpto_2")
assert isinstance(jr, AccessTokenRepository)

View File

@@ -0,0 +1,56 @@
from typing import Any, List, Optional, Tuple, TypeVar
from openbb_core.app.repository.abstract.journal_entry_repository import (
JournalEntryRepository,
)
from pydantic import BaseModel
T = TypeVar("T", bound=BaseModel)
def test_journal_entry_repository():
class TestJournalEntryRepository(JournalEntryRepository):
def __init__(self, test_arg_1, test_arg_2):
self.test_arg_1 = test_arg_1
self.test_arg_2 = test_arg_2
super().__init__()
@property
def collection_name(self):
return "test_collection"
def create(self, model):
pass
@property
def database_name(self):
return "test_database"
def delete(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def delete_all(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def read(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def read_all(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def update(self, model: T):
pass
def mock_method(self):
return self.test_arg_1, self.test_arg_2
jr = TestJournalEntryRepository("xpto_1", "xpto_2")
assert isinstance(jr, JournalEntryRepository)

View File

@@ -0,0 +1,56 @@
from typing import Any, List, Optional, Tuple, TypeVar
from openbb_core.app.repository.abstract.journal_repository import (
JournalRepository,
)
from pydantic import BaseModel
T = TypeVar("T", bound=BaseModel)
def test_journal_repository():
class TestJournalRepository(JournalRepository):
def __init__(self, test_arg_1, test_arg_2):
self.test_arg_1 = test_arg_1
self.test_arg_2 = test_arg_2
super().__init__()
@property
def collection_name(self):
return "test_collection"
def create(self, model):
pass
@property
def database_name(self):
return "test_database"
def delete(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def delete_all(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def read(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def read_all(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def update(self, model: T):
pass
def mock_method(self):
return self.test_arg_1, self.test_arg_2
jr = TestJournalRepository("xpto_1", "xpto_2")
assert isinstance(jr, JournalRepository)

View File

@@ -0,0 +1,56 @@
from typing import Any, List, Optional, Tuple, TypeVar
from openbb_core.app.repository.abstract.repository import (
Repository,
)
from pydantic import BaseModel
T = TypeVar("T", bound=BaseModel)
def test_repository():
class TestRepository(Repository):
def __init__(self, test_arg_1, test_arg_2):
self.test_arg_1 = test_arg_1
self.test_arg_2 = test_arg_2
super().__init__()
@property
def collection_name(self):
return "test_collection"
def create(self, model):
pass
@property
def database_name(self):
return "test_database"
def delete(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def delete_all(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def read(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def read_all(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def update(self, model: T):
pass
def mock_method(self):
return self.test_arg_1, self.test_arg_2
jr = TestRepository("xpto_1", "xpto_2")
assert isinstance(jr, Repository)

View File

@@ -0,0 +1,63 @@
from typing import Any, List, Optional, Tuple, TypeVar
from openbb_core.app.repository.abstract.user_settings_repository import (
UserSettingsRepository,
)
from pydantic import BaseModel
T = TypeVar("T", bound=BaseModel)
def test_journal_entry_repository():
class TestUserSettingsRepository(UserSettingsRepository):
def __init__(self, test_arg_1, test_arg_2):
self.test_arg_1 = test_arg_1
self.test_arg_2 = test_arg_2
super().__init__()
@property
def collection_name(self):
return "test_collection"
def create(self, model):
pass
@property
def database_name(self):
return "test_database"
def delete(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def delete_all(self, filter_list: Optional[List[Tuple[str, Any]]] = None):
pass
def read(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def read_all(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
def update(self, model: T):
pass
def mock_method(self):
return self.test_arg_1, self.test_arg_2
def read_by_profile(
self,
field_list: Optional[List[str]] = None,
filter_list: Optional[List[Tuple[str, Any]]] = None,
):
pass
jr = TestUserSettingsRepository("xpto_1", "xpto_2")
assert isinstance(jr, UserSettingsRepository)

View File

@@ -0,0 +1,19 @@
from unittest.mock import patch
from openbb_core.app.repository.mongodb.access_token_repository import (
AccessTokenRepository,
)
@patch("openbb_core.app.repository.mongodb.access_token_repository.MongoClient")
def test_access_token_repository(mock_mongo_client):
atr = AccessTokenRepository(
client=mock_mongo_client,
collection_name="collection_name",
database_name="database_name",
)
assert isinstance(atr, AccessTokenRepository)
assert atr.client == mock_mongo_client
assert atr.collection_name == "collection_name"
assert atr.database_name == "database_name"

View File

@@ -0,0 +1,9 @@
from openbb_core.app.model.results.empty import Empty
from pydantic import BaseModel
def test_empty_model():
empty = Empty()
assert isinstance(empty, Empty)
assert isinstance(empty, BaseModel)

View File

@@ -0,0 +1,174 @@
"""Test the hub_service.py module."""
from unittest.mock import MagicMock, patch
import pytest
from openbb_core.app.service.hub_service import (
Credentials,
HubService,
HubSession,
HubUserSettings,
OpenBBError,
)
@pytest.fixture
def mocker():
"""Fixture for mocker."""
with patch("openbb_core.app.service.hub_service.HubService") as mock:
yield mock
def test_connect_with_email_password():
"""Test connect with email and password."""
mock_hub_session = MagicMock(spec=HubSession)
with patch(
"requests.post", return_value=MagicMock(status_code=200, json=lambda: {})
):
with patch.object(
HubService, "get_session_from_email_password", return_value=mock_hub_session
):
hub_service = HubService()
result = hub_service.connect(email="test@example.com", password="password")
assert result == mock_hub_session
assert hub_service.session == mock_hub_session
def test_connect_with_sdk_token():
"""Test connect with SDK personal access token."""
mock_hub_session = MagicMock(spec=HubSession)
with patch(
"requests.post", return_value=MagicMock(status_code=200, json=lambda: {})
):
with patch.object(
HubService, "get_session_from_sdk_token", return_value=mock_hub_session
):
hub_service = HubService()
result = hub_service.connect(sdk_token="sdk_token")
assert result == mock_hub_session
assert hub_service.session == mock_hub_session
def test_connect_without_credentials():
"""Test connect without credentials."""
hub_service = HubService()
with pytest.raises(
OpenBBError, match="Please provide 'email' and 'password' or 'sdk_token'"
):
hub_service.connect()
@patch("requests.post")
def test_get_session_from_email_password(mock_post):
"""Test get session from email and password."""
mock_post.return_value = MagicMock(
status_code=200,
json=lambda: {
"access_token": "token",
"token_type": "Bearer",
"uuid": "uuid",
"email": "email",
"username": "username",
"primary_usage": "primary_usage",
},
)
result = HubService.get_session_from_email_password("email", "password")
assert isinstance(result, HubSession)
@patch("requests.post")
def test_get_session_from_sdk_token(mock_post):
"""Test get session from SDK personal access token."""
mock_post.return_value = MagicMock(
status_code=200,
json=lambda: {
"access_token": "token",
"token_type": "Bearer",
"uuid": "uuid",
"username": "username",
"email": "email",
"primary_usage": "primary_usage",
},
)
mock_token = (
"eyJ0eXAiOiJKV1QiLCJhbGciOiJFUzI1NiIsImtpZCI6ImRiMjEyZDdhZj"
"c2MWI0ZTNlOGNjZGM3OWQ5Zjk4YWM5In0.eyJhY2Nlc3NfdG9rZW4iOiJ0"
"b2tlbiIsInRva2VuX3R5cGUiOiJCZWFyZXIiLCJ1dWlkIjoidXVpZCIsInV"
"zZXJuYW1lIjoidXNlcm5hbWUiLCJlbWFpbCI6ImVtYWlsIiwicHJpbWFyeV9"
"1c2FnZSI6InByaW1hcnlfdXNhZ2UifQ.FAtE8-a1a-313Zoa6dREIxGZOHaW9"
"-JLZnFzyJ6dlHBZnkjQT2tfaaefxnTdAlSmToQwxGykvuatmI7L0wztPQ"
)
result = HubService.get_session_from_sdk_token(mock_token)
assert isinstance(result, HubSession)
@patch("requests.get")
def test_disconnect(mock_get):
"""Test disconnect."""
mock_get.return_value = MagicMock(status_code=200, json=lambda: {"success": True})
mock_hub_session = MagicMock(
spec=HubSession, access_token="token", token_type="Bearer"
)
hub_service = HubService(session=mock_hub_session)
assert hub_service.disconnect() is True
assert hub_service.session is None
@patch("requests.get")
def test_get_user_settings(mock_get):
"""Test get user settings."""
mock_get.return_value = MagicMock(status_code=200, json=lambda: {})
mock_hub_session = MagicMock(
spec=HubSession, access_token="token", token_type="Bearer"
)
user_settings = HubService.get_user_settings(mock_hub_session)
assert isinstance(user_settings, HubUserSettings)
@patch("requests.put")
def test_put_user_settings(mock_put):
"""Test put user settings."""
mock_put.return_value = MagicMock(status_code=200)
mock_hub_session = MagicMock(
spec=HubSession, access_token="token", token_type="Bearer"
)
mock_user_settings = MagicMock(spec=HubUserSettings)
assert HubService.put_user_settings(mock_hub_session, mock_user_settings) is True
def test_hub2sdk():
"""Test hub2sdk."""
mock_user_settings = MagicMock(spec=HubUserSettings)
mock_user_settings.features_keys = MagicMock(
API_KEY_FINANCIALMODELINGPREP="fmp",
API_POLYGON_KEY="polygon",
API_FRED_KEY="fred",
)
credentials = HubService.hub2sdk(mock_user_settings)
assert isinstance(credentials, Credentials)
assert credentials.fmp_api_key == "fmp"
assert credentials.polygon_api_key == "polygon"
assert credentials.fred_api_key == "fred"
def test_sdk2hub():
"""Test sdk2hub."""
mock_credentials = MagicMock(spec=Credentials)
mock_credentials.fmp_api_key = "fmp"
mock_credentials.polygon_api_key = "polygon"
mock_credentials.fred_api_key = "fred"
user_settings = HubService.sdk2hub(mock_credentials)
assert isinstance(user_settings, HubUserSettings)
assert user_settings.features_keys.API_KEY_FINANCIALMODELINGPREP == "fmp"
assert user_settings.features_keys.API_POLYGON_KEY == "polygon"
assert user_settings.features_keys.API_FRED_KEY == "fred"

View File

@@ -0,0 +1,103 @@
"""Test the journal_service.py module."""
# pylint: disable=redefined-outer-name
from unittest.mock import MagicMock
import pytest
from openbb_core.app.service.journal_service import (
AbstractJournalEntryRepository,
InMemoryJournalEntryRepository,
JournalService,
MongoClient,
MongoDBJournalEntryRepository,
)
@pytest.fixture
def journal_service():
"""Fixture for journal service."""
return JournalService()
def test_journal_service_init(journal_service):
"""Test journal service init."""
assert journal_service
def test_invalid_client(journal_service):
"""Test valid client."""
assert not journal_service.valid_client()
def test_build_journal_entry_repository_with_given_repository():
"""Test build with repository."""
mock_repository = MagicMock(spec=AbstractJournalEntryRepository)
result = JournalService.build_journal_entry_repository(
valid_client=True, journal_entry_repository=mock_repository
)
assert result == mock_repository
def test_build_journal_entry_repository_with_mongodb_client():
"""Test build with mongodb client."""
mock_client = MagicMock(spec=MongoClient)
result = JournalService.build_journal_entry_repository(
valid_client=True, mongodb_client=mock_client
)
assert isinstance(result, MongoDBJournalEntryRepository)
def test_build_journal_entry_repository_memory():
"""Test build with in-memory access."""
result = JournalService.build_journal_entry_repository(valid_client=True)
assert isinstance(result, InMemoryJournalEntryRepository)
def test_build_journal_repository_with_given_repository():
"""Test build with repository."""
mock_repository = MagicMock(spec=AbstractJournalEntryRepository)
result = JournalService.build_journal_entry_repository(
valid_client=True, journal_entry_repository=mock_repository
)
assert result == mock_repository
def test_build_journal_repository_with_mongodb_client():
"""Test build with mongodb client."""
mock_client = MagicMock(spec=MongoClient)
result = JournalService.build_journal_entry_repository(
valid_client=True, mongodb_client=mock_client
)
assert isinstance(result, MongoDBJournalEntryRepository)
def test_build_journal_repository_memory():
"""Test build with in-memory access."""
result = JournalService.build_journal_entry_repository(valid_client=True)
assert isinstance(result, InMemoryJournalEntryRepository)
def test_journal_service_mongodb_client(journal_service):
"""Test mongodb client."""
assert journal_service.mongodb_client is None
def test_journal_service_journal_entry_repository(journal_service):
"""Test journal entry repository."""
assert isinstance(
journal_service.journal_entry_repository, InMemoryJournalEntryRepository
)
def test_journal_service_journal_repository(journal_service):
"""Test journal repository."""
assert journal_service.journal_repository
assert isinstance(
type(journal_service.journal_repository), type(InMemoryJournalEntryRepository)
)

View File

@@ -0,0 +1,54 @@
"""Test the system_service.py module."""
# pylint: disable=redefined-outer-name
import pytest
from openbb_core.app.service.system_service import SystemService
@pytest.fixture
def system_service():
"""Fixture for system service."""
return SystemService()
def test_system_service_init(system_service):
"""Test system service init."""
assert system_service
def test_read_default_system_settings(system_service):
"""Test read default system settings."""
system_settings = system_service.read_default_system_settings()
assert system_settings
def test_write_default_system_settings(system_service):
"""Test write default system settings."""
system_settings = system_service.read_default_system_settings()
system_service.write_default_system_settings(system_settings=system_settings)
assert system_service
def test_system_settings(system_service):
"""Test system settings."""
system_settings = system_service.system_settings
assert system_settings
def test_system_settings_setter(system_service):
"""Test system settings setter."""
system_settings = system_service.system_settings
system_service.system_settings = system_settings
assert system_service.system_settings == system_settings
def test_refresh_system_settings(system_service):
"""Test refresh system settings."""
system_settings = system_service.refresh_system_settings()
assert system_settings

View File

@@ -0,0 +1,119 @@
"""Test the user_service.py module."""
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from openbb_core.app.service.user_service import (
AbstractAccessTokenRepository,
InMemoryAccessTokenRepository,
MongoClient,
MongoDBAccessTokenRepository,
UserService,
UserSettings,
)
def test_build_token_repository_with_given_repository():
"""Test build with repository."""
mock_repository = MagicMock(spec=AbstractAccessTokenRepository)
result = UserService.build_token_repository(
valid_client=True, access_token_repository=mock_repository
)
assert result == mock_repository
def test_build_token_repository_with_mongodb_client():
"""Test build with mongodb client."""
mock_client = MagicMock(spec=MongoClient)
result = UserService.build_token_repository(
valid_client=True, mongodb_client=mock_client
)
assert isinstance(result, MongoDBAccessTokenRepository)
def test_build_token_repository_memory():
"""Test build with in-memory access."""
result = UserService.build_token_repository(valid_client=True)
assert isinstance(result, InMemoryAccessTokenRepository)
def test_build_user_settings_repository_with_given_repository():
"""Test build with repository."""
mock_repository = MagicMock(spec=AbstractAccessTokenRepository)
result = UserService.build_user_settings_repository(
valid_client=True, user_settings_repository=mock_repository
)
assert result == mock_repository
def test_build_user_settings_repository_with_mongodb_client():
"""Test build with mongodb client."""
mock_client = MagicMock(spec=MongoClient)
result = UserService.build_user_settings_repository(
valid_client=True, mongodb_client=mock_client
)
assert result
assert isinstance(type(result), type(MongoDBAccessTokenRepository))
def test_build_user_settings_repository_memory():
"""Test build with in-memory access."""
result = UserService.build_user_settings_repository(valid_client=True)
assert result
assert isinstance(type(result), type(InMemoryAccessTokenRepository))
def test_valid_client_false():
"""Test valid client."""
result = UserService.valid_client()
assert result is False
def test_valid_client_true():
"""Test valid client."""
mock_client = MagicMock(spec=MongoClient)
result = UserService.valid_client(client=mock_client)
assert result is True
# @pytest.mark.skip(reason="TODO")
def test_read_default_user_settings_file_exists():
"""Test read default user settings."""
result = UserService.read_default_user_settings(path=Path("some_path"))
assert result
assert isinstance(result, UserSettings)
@pytest.mark.skip(reason="TODO")
def test_write_default_user_settings():
"""Test write default user settings."""
@pytest.mark.skip(reason="TODO")
def test_update_default():
"""Test update default user settings."""
def test_merge_dicts():
"""Test merge dicts."""
result = UserService.merge_dicts(
list_of_dicts=[
{"a": 1, "b": 2},
{"a": 3, "b": 4},
]
)
assert result
assert isinstance(result, dict)
assert result["a"] == 3
assert result["b"] == 4

View File

@@ -0,0 +1,48 @@
"""Test static app factory."""
# pylint: disable=redefined-outer-name
import pytest
from openbb_core.app.model.system_settings import SystemSettings
from openbb_core.app.model.user_settings import UserSettings
from openbb_core.app.static.account import Account
from openbb_core.app.static.app_factory import create_app
from openbb_core.app.static.coverage import Coverage
@pytest.fixture(scope="module")
def app_factory():
"""Return app factory."""
return create_app()
def test_app_factory_init(app_factory):
"""Test app factory init."""
assert app_factory
def test_app_system_settings(app_factory):
"""Test app system settings."""
system_settings = app_factory.system
assert system_settings
assert isinstance(system_settings, SystemSettings)
def test_app_user_settings(app_factory):
"""Test app user settings."""
user_settings = app_factory.settings
assert user_settings
assert isinstance(user_settings, UserSettings)
def test_app_account(app_factory):
"""Test app account."""
account = app_factory.account
assert account
assert isinstance(account, Account)
def test_app_coverage(app_factory):
"""Test app coverage."""
coverage = app_factory.coverage
assert coverage
assert isinstance(coverage, Coverage)

View File

@@ -0,0 +1,9 @@
"""Test the container.py file."""
from openbb_core.app.static.container import Container
def test_container_init():
"""Test container init."""
container = Container(None)
assert container

View File

@@ -0,0 +1,30 @@
"""Test the coverage.py file."""
# pylint: disable=redefined-outer-name
import pytest
from openbb_core.app.static.coverage import Coverage
@pytest.fixture(scope="module")
def coverage():
"""Return coverage."""
return Coverage()
def test_coverage_init(coverage):
"""Test coverage init."""
assert coverage
def test_coverage_providers(coverage):
"""Test coverage providers."""
provider_coverage = coverage.providers
assert provider_coverage
assert isinstance(provider_coverage, dict)
def test_coverage_commands(coverage):
"""Test coverage commands."""
command_coverage = coverage.commands
assert command_coverage
assert isinstance(command_coverage, dict)

View File

@@ -0,0 +1,44 @@
"""Test the filters.py file."""
import pytest
from openbb_core.app.model.command_output import CommandOutput
from openbb_core.app.static.filters import filter_call, filter_inputs, filter_output
from pandas import DataFrame
def test_filter_call_error():
"""Test filter_call."""
class Test:
@filter_call
def test(self):
return 1 / 0
test = Test()
with pytest.raises(Exception):
test.test()
def test_filter_inputs_not_df():
"""Test filter_inputs."""
kwargs = {"num": 1}
kwargs = filter_inputs(**kwargs)
assert kwargs["num"] == 1
def test_filter_inputs_df():
"""Test filter_inputs."""
df = DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
kwargs = {"df": df}
kwargs = filter_inputs(**kwargs)
assert isinstance(kwargs["df"], list)
def test_filter_output():
"""Test filter_output."""
command_output = filter_output(CommandOutput())
assert command_output is not None

View File

@@ -0,0 +1,449 @@
"""Test the package_builder.py file."""
# pylint: disable=redefined-outer-name
from dataclasses import dataclass
from inspect import _empty
import pytest
from openbb_core.app.static.package_builder import (
ClassDefinition,
ImportDefinition,
Linters,
MethodDefinition,
ModuleBuilder,
PackageBuilder,
Parameter,
PathHandler,
)
from pydantic import Field
from typing_extensions import Annotated
@pytest.fixture(scope="module")
def package_builder():
"""Return package builder."""
return PackageBuilder()
def test_package_builder_init(package_builder):
"""Test package builder init."""
assert package_builder
@pytest.mark.skip("Rebuilds the packages which takes some time.")
def test_package_builder_build(package_builder):
"""Test package builder build."""
package_builder.build()
@pytest.mark.skip()
def test_save_module_map(package_builder):
"""Test save module map."""
package_builder.save_module_map()
@pytest.mark.skip()
def test_save_modules(package_builder):
"""Test save module."""
package_builder.save_modules()
@pytest.mark.skip()
def test_save_package(package_builder):
"""Test save package."""
package_builder.save_package()
def test_run_linters(package_builder):
"""Test run linters."""
package_builder.run_linters()
@pytest.mark.skip("We avoid writing to the package.")
def test_write_to_package(package_builder):
"""Test save to package."""
package_builder.write_to_package()
@pytest.fixture(scope="module")
def module_builder():
"""Return module builder."""
return ModuleBuilder()
def test_module_builder_init(module_builder):
"""Test module builder init."""
assert module_builder
@pytest.fixture(scope="module")
def class_definition():
"""Return class definition."""
return ClassDefinition()
def test_class_definition_init(class_definition):
"""Test class definition init."""
assert class_definition
def test_build(class_definition):
"""Test build."""
code = class_definition.build("openbb_core.app.static.container.Container")
assert code
@pytest.fixture(scope="module")
def method_definition():
"""Return method definition."""
return MethodDefinition()
def test_method_definition_init(method_definition):
"""Test method definition init."""
assert method_definition
def test_build_class_loader_method(method_definition):
"""Test build class loader method."""
code = method_definition.build_class_loader_method(
"openbb_core.app.static.container.Container"
)
assert code
def test_get_type(method_definition):
"""Test get type."""
type_ = method_definition.get_type(field=Parameter.empty)
assert type_
assert isinstance(type_, type)
def test_get_type_hint(method_definition):
"""Test get type hint."""
class TestField:
type = int
field = TestField()
result = method_definition.get_type(field)
assert result == int
def test_field_with_type_attribute_missing_type(method_definition):
"""Test field with type attribute missing type."""
class TestField:
type = Parameter.empty
field = TestField()
result = method_definition.get_type(field)
assert result is Parameter.empty
def test_get_default(method_definition):
"""Test get default."""
class TestField:
default = Field(default=42)
field = TestField()
result = method_definition.get_default(field)
assert result == 42
def test_get_default_none(method_definition):
"""Test get default."""
class TestField:
default = 42
field = TestField()
result = method_definition.get_default(field)
assert result is None
def test_get_default_default_value(method_definition):
"""Test get default default value."""
class TestField:
default = type(Ellipsis)()
field = TestField()
result = method_definition.get_default(field)
assert result is None
def test_get_default_no_default(method_definition):
"""Test get default no default."""
class TestField:
pass
field = TestField()
result = method_definition.get_default(field)
assert result == _empty
def test_is_annotated_dc(method_definition):
"""Test is annotated dc."""
result = method_definition.is_annotated_dc(annotation=Parameter.empty)
assert not result
def test_is_annotated_dc_annotated(method_definition):
"""Test is annotated dc annotated."""
@dataclass
class TestAnnotatedDataClass:
"""Test annotated data class."""
value: int
annotated_dataclass = Annotated[TestAnnotatedDataClass, "test_annotation"]
result = method_definition.is_annotated_dc(annotation=annotated_dataclass)
assert result
def test_reorder_params(method_definition):
"""Test reorder params."""
params = {
"provider": Parameter.empty,
"extra_params": Parameter.empty,
"param1": Parameter.empty,
"param2": Parameter.empty,
}
result = method_definition.reorder_params(params=params)
assert result
assert list(result.keys()) == ["param1", "param2", "provider", "extra_params"]
def test_build_func_params(method_definition):
"""Test build func params."""
param_map = {
"param1": Parameter("NoneType", kind=Parameter.POSITIONAL_OR_KEYWORD),
"param2": Parameter("int", kind=Parameter.POSITIONAL_OR_KEYWORD),
}
expected_output = "param1, param2, chart: bool = False"
output = method_definition.build_func_params(param_map)
assert output == expected_output
@pytest.mark.parametrize(
"return_type, expected_output",
[
(_empty, "None"),
(int, "int"),
],
)
def test_build_func_returns(method_definition, return_type, expected_output):
"""Test build func returns."""
output = method_definition.build_func_returns(return_type=return_type)
assert output == expected_output
def test_build_command_method_signature(method_definition):
"""Test build command method signature."""
parameter_map = {
"param1": Parameter("NoneType", kind=Parameter.POSITIONAL_OR_KEYWORD),
"param2": Parameter("int", kind=Parameter.POSITIONAL_OR_KEYWORD),
}
return_type = int
output = method_definition.build_command_method_signature(
func_name="test_func", parameter_map=parameter_map, return_type=return_type
)
assert output
def test_build_command_method_doc(method_definition):
"""Test build command method doc."""
def some_func():
"""Do some func doc."""
output = method_definition.build_command_method_doc(func=some_func)
assert output
assert isinstance(output, str)
def test_build_command_method_implementation(method_definition):
"""Test build command method implementation."""
def some_func():
"""Do some func doc."""
return 42
output = method_definition.build_command_method_implementation(
path="openbb_core.app.static.container.Container", func=some_func
)
assert output
assert isinstance(output, str)
def test_build_command_method(method_definition):
"""Test build command method."""
def some_func():
"""Do some func doc."""
return 42
output = method_definition.build_command_method(
path="openbb_core.app.static.container.Container",
func=some_func,
model_name=None,
)
assert output
assert isinstance(output, str)
@pytest.fixture(scope="module")
def import_definition():
"""Return import definition."""
return ImportDefinition()
def test_import_definition_init(import_definition):
"""Test import definition init."""
assert import_definition
def test_filter_hint_type_list(import_definition):
"""Test filter type hint list."""
output = import_definition.filter_hint_type_list(
hint_type_list=[int, str, float, bool, _empty, _empty, _empty, _empty]
)
assert output == []
def test_import_definition_get_path_hint_type_list(import_definition):
"""Test import definition get path hint type list."""
hint_type_list = import_definition.get_path_hint_type_list(
path="openbb_core.app.static.container.Container"
)
assert hint_type_list == []
def test_import_definition_build(import_definition):
"""Test import definition build."""
code = import_definition.build(path="openbb_core.app.static.container.Container")
assert code
@pytest.fixture(scope="module")
def path_handler():
"""Return path handler."""
return PathHandler()
def test_path_handler_init(path_handler):
"""Test path handler init."""
assert path_handler
@pytest.fixture(scope="module")
def route_map(path_handler):
"""Return route map."""
return path_handler.build_route_map()
def test_build_route_map(route_map):
"""Test build route map."""
assert route_map
assert isinstance(route_map, dict)
@pytest.fixture(scope="module")
def path_list(path_handler, route_map):
"""Return path list."""
return path_handler.build_path_list(route_map=route_map)
def test_build_path_list(path_list):
"""Test build path list."""
assert path_list
assert isinstance(path_list, list)
def test_get_route(path_handler, route_map):
"""Test get route."""
route = path_handler.get_route(route_map=route_map, path="/stocks/load")
assert route
def test_get_child_path_list(path_handler, path_list):
"""Test get child path list."""
child_path_list = path_handler.get_child_path_list(
path="/stocks", path_list=path_list
)
assert child_path_list
assert isinstance(child_path_list, list)
def test_clean_path(path_handler):
"""Test clean path."""
path = "/stocks/load"
result = path_handler.clean_path(path=path)
assert result == "stocks_load"
def test_build_module_name(path_handler):
"""Test build module name."""
module_name = path_handler.build_module_name(path="")
assert module_name == "__extensions__"
module_name = path_handler.build_module_name(path="/stocks/load")
assert module_name == "stocks_load"
def test_build_module_class(path_handler):
"""Test build module class."""
module_class = path_handler.build_module_class(path="")
assert module_class == "Extensions"
module_class = path_handler.build_module_class(path="/stocks/load")
assert module_class == "CLASS_stocks_load"
@pytest.fixture(scope="module")
def linters():
"""Return linters."""
return Linters()
def test_linters_init(linters):
"""Test linters init."""
assert linters
def test_print_separator(linters):
"""Test print separator."""
linters.print_separator(symbol="AAPL")
def test_run(linters):
"""Test run."""
linters.run(linter="ruff")
def test_ruff(linters):
"""Test ruff."""
linters.ruff()
def test_black(linters):
"""Test black."""
linters.black()
def test_mypy(linters):
"""Test mypy."""
linters.mypy()
# TODO: Implement unit test for the DocstringGenerator class once refactored.

View File

@@ -0,0 +1,85 @@
"""Test provider interface."""
# pylint: disable=redefined-outer-name
import pytest
from openbb_core.app.provider_interface import (
ProviderChoices,
ProviderInterface,
get_provider_interface,
)
@pytest.fixture(scope="module")
def provider_interface():
"""Set up provider_interface."""
return ProviderInterface()
def test_init(provider_interface):
"""Test init."""
assert provider_interface
def test_get_provider_interface():
"""Test get provider interface."""
assert isinstance(get_provider_interface(), ProviderInterface)
def test_map(provider_interface):
"""Test map."""
provider_interface_map = provider_interface.map
assert isinstance(provider_interface_map, dict)
assert len(provider_interface_map) > 0
assert "StockEOD" in provider_interface_map
def test_required_credentials(provider_interface):
"""Test required credentials."""
required_credentials = provider_interface.required_credentials
assert isinstance(required_credentials, list)
assert len(required_credentials) > 0
def test_model_providers(provider_interface):
"""Test model providers."""
model_providers = provider_interface.model_providers
assert isinstance(model_providers, dict)
assert len(model_providers) > 0
def test_params(provider_interface):
"""Test params."""
params = provider_interface.params
assert isinstance(params, dict)
assert len(params) > 0
assert "StockEOD" in params
def test_data(provider_interface):
"""Test data."""
data = provider_interface.data
assert isinstance(data, dict)
assert len(data) > 0
assert "StockEOD" in data
def test_providers_literal(provider_interface):
"""Test providers literal."""
providers_literal = provider_interface.providers_literal
assert isinstance(type(providers_literal), type)
assert len(providers_literal.__args__) > 0
assert "openbb" not in providers_literal.__args__
def test_provider_choices(provider_interface):
"""Test provider choices."""
provider_choices = provider_interface.provider_choices
assert isinstance(provider_choices, type(ProviderChoices))
def test_models(provider_interface):
"""Test models."""
models = provider_interface.models
assert isinstance(models, list)
assert len(models) > 0
assert "StockEOD" in models

Some files were not shown because too many files have changed in this diff Show More