jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-27 20:42:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Routine remote storage (#4143)

Browse Source 
* add upload routine command

* remove prints

* point to local again

* add commands

* add delete

* add download command and upload 409

* list comprehension

* add download & save plus upload fixe

* add user uuid folder

* add list command

* ruff

* en.yml detail

* provide choices to delete & download

* add choices to upload

* change menu text

* starting adding unit tests

* bad english

* Update test_print_help.txt

* genearl msgs

* remove print

* add docstrings

* add unit tests to account controller

* change parser

* fix routine upload test

* add to be implemented tests hub

* add to be implemented tests local

* add total pages to title

* split list into model and view

* add docstring

* print msg if character

* add local model tests

* hub tests routines

* fix test

* point to dev and glob

* fix small bug on upload

* missing return

* log decorators missing

* black

* ruff
This commit is contained in:
montezdesousa
2023-02-09 09:55:05 +00:00
committed by GitHub
parent 759a2c7557
commit ead2e2e76a
14 changed files with 951 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

244
openbb_terminal/account/account_controller.py
View File

@@ -1,16 +1,19 @@
import argparse
import json
import logging
import os
from pathlib import Path
from typing import Dict, List
from prompt_toolkit.completion import NestedCompleter
from openbb_terminal import feature_flags as obbff
from openbb_terminal.account.account_model import get_diff
from openbb_terminal.account.account_model import get_diff, get_routines_info
from openbb_terminal.account.account_view import display_routines_list
from openbb_terminal.core.config.paths import USER_ROUTINES_DIRECTORY
from openbb_terminal.decorators import log_start_end
from openbb_terminal.featflags_controller import FeatureFlagsController
from openbb_terminal.helper_funcs import check_positive
from openbb_terminal.menu import session
from openbb_terminal.parent_classes import BaseController
from openbb_terminal.rich_config import MenuText, console
@@ -30,6 +33,10 @@ class AccountController(BaseController):
"sync",
"pull",
"clear",
"list",
"upload",
"download",
"delete",
]
PATH = "/account/"
@@ -37,21 +44,36 @@ class AccountController(BaseController):
def __init__(self, queue: List[str] = None):
super().__init__(queue)
self.ROUTINE_FILES: Dict[str, Path] = {}
self.ROUTINE_CHOICES: Dict[str, None] = {}
self.REMOTE_CHOICES: List[str] = []
self.update_runtime_choices()
def update_runtime_choices(self):
"""Update runtime choices"""
self.ROUTINE_FILES = {
filepath.name: filepath
for filepath in USER_ROUTINES_DIRECTORY.rglob("*.openbb")
}
self.ROUTINE_CHOICES = {filename: None for filename in self.ROUTINE_FILES}
self.ROUTINE_FILES = self.get_routines()
if session and obbff.USE_PROMPT_TOOLKIT:
choices: dict = {c: {} for c in self.controller_choices}
choices["sync"] = {"--on": {}, "--off": {}}
choices["upload"]["--file"] = {c: {} for c in self.ROUTINE_FILES}
choices["upload"]["-f"] = choices["upload"]["--file"]
choices["download"]["--name"] = {c: {} for c in self.REMOTE_CHOICES}
choices["download"]["-n"] = choices["download"]["--name"]
choices["delete"]["--name"] = {c: {} for c in self.REMOTE_CHOICES}
choices["delete"]["-n"] = choices["delete"]["--name"]
self.completer = NestedCompleter.from_nested_dict(choices)
def get_routines(self):
"""Get routines"""
routines = {
filepath.name: filepath
for filepath in USER_ROUTINES_DIRECTORY.glob("*.openbb")
}
user_folder = USER_ROUTINES_DIRECTORY / User.get_uuid()
if os.path.exists(user_folder):
routines.update(
{filepath.name: filepath for filepath in user_folder.rglob("*.openbb")}
)
return routines
def print_help(self):
"""Print help"""
@@ -60,6 +82,12 @@ class AccountController(BaseController):
mt.add_cmd("sync")
mt.add_cmd("pull")
mt.add_cmd("clear")
mt.add_raw("\n")
mt.add_info("_routines_")
mt.add_cmd("list")
mt.add_cmd("upload")
mt.add_cmd("download")
mt.add_cmd("delete")
console.print(text=mt.menu_text, menu="Account")
@log_start_end(log=logger)
@@ -85,7 +113,7 @@ class AccountController(BaseController):
action="store_true",
default=False,
)
ns_parser = self.parse_simple_args(parser, other_args)
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
if ns_parser.on:
if not obbff.SYNC_ENABLED:
@@ -108,6 +136,7 @@ class AccountController(BaseController):
else:
console.print(f"sync is {sync}, use --on or --off to change.")
@log_start_end(log=logger)
def call_pull(self, other_args: List[str]):
"""Pull data"""
parser = argparse.ArgumentParser(
@@ -116,7 +145,7 @@ class AccountController(BaseController):
prog="pull",
description="Pull and apply stored configurations from the cloud.",
)
ns_parser = self.parse_simple_args(parser, other_args)
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
response = Hub.fetch_user_configs(User.get_session())
if response:
@@ -134,6 +163,7 @@ class AccountController(BaseController):
else:
console.print("[info]No changes to apply.[/info]")
@log_start_end(log=logger)
def call_clear(self, other_args: List[str]):
"""Clear data"""
parser = argparse.ArgumentParser(
@@ -142,7 +172,7 @@ class AccountController(BaseController):
prog="clear",
description="Clear stored configurations from the cloud.",
)
ns_parser = self.parse_simple_args(parser, other_args)
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
i = console.input(
"[red]This action is irreversible![/red]\n"
@@ -153,3 +183,197 @@ class AccountController(BaseController):
Hub.clear_user_configs(auth_header=User.get_auth_header())
else:
console.print("\n[info]Aborted.[/info]")
@log_start_end(log=logger)
def call_list(self, other_args: List[str]):
"""List routines"""
parser = argparse.ArgumentParser(
add_help=False,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="list",
description="List routines available in the cloud.",
)
parser.add_argument(
"-p",
"--page",
type=check_positive,
dest="page",
default=1,
help="The page number.",
)
parser.add_argument(
"-s",
"--size",
type=check_positive,
dest="size",
default=10,
help="The number of results per page.",
)
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
response = Hub.list_routines(
auth_header=User.get_auth_header(),
page=ns_parser.page,
size=ns_parser.size,
)
df, page, pages = get_routines_info(response)
if not df.empty:
self.REMOTE_CHOICES = list(df["name"])
self.update_runtime_choices()
display_routines_list(df, page, pages)
else:
console.print("[red]No routines found.[/red]")
@log_start_end(log=logger)
def call_upload(self, other_args: List[str]):
"""Upload"""
parser = argparse.ArgumentParser(
add_help=False,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="upload",
description="Upload a routine to the cloud.",
)
parser.add_argument(
"-f",
"--file",
type=str,
dest="file",
required="-h" not in other_args,
help="The file to be loaded",
metavar="FILE",
nargs="+",
)
parser.add_argument(
"-d",
"--description",
type=str,
dest="description",
help="The description of the routine",
default="",
nargs="+",
)
parser.add_argument(
"-n",
"--name",
type=str,
dest="name",
help="The name of the routine.",
nargs="+",
)
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
routine = Local.get_routine(file_name=" ".join(ns_parser.file))
if routine:
description = " ".join(ns_parser.description)
name = (
" ".join(ns_parser.name)
if ns_parser.name
else ns_parser.file.split(".openbb")[0]
)
response = Hub.upload_routine(
auth_header=User.get_auth_header(),
name=name,
description=description,
routine=routine,
)
if response is not None and response.status_code == 409:
i = console.input(
"A routine with the same name already exists, "
"do you want to overwrite it? (y/n): "
)
console.print("")
if i.lower() in ["y", "yes"]:
response = Hub.upload_routine(
auth_header=User.get_auth_header(),
name=name,
description=description,
routine=routine,
override=True,
)
else:
console.print("[info]Aborted.[/info]")
if response and response.status_code == 200:
self.REMOTE_CHOICES.append(name)
self.update_runtime_choices()
@log_start_end(log=logger)
def call_download(self, other_args: List[str]):
"""Download"""
parser = argparse.ArgumentParser(
add_help=False,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="download",
description="Download a routine from the cloud.",
)
parser.add_argument(
"-n",
"--name",
type=str,
dest="name",
help="The name of the routine.",
required="-h" not in other_args,
nargs="+",
)
if other_args and "-" not in other_args[0][0]:
other_args.insert(0, "-n")
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
response = Hub.download_routine(
auth_header=User.get_auth_header(),
name=" ".join(ns_parser.name),
)
if response and response.status_code == 200:
data = response.json()
if data:
name = data.get("name", "")
if name:
console.print(f"[info]Name:[/info] {name}")
description = data.get("description", "")
if description:
console.print(f"[info]Description:[/info] {description}")
script = data.get("script", "")
if script:
file_name = f"{name}.openbb"
file_path = Local.save_routine(
file_name=file_name,
routine=script,
)
if file_path:
console.print(f"[info]Location:[/info] {file_path}")
@log_start_end(log=logger)
def call_delete(self, other_args: List[str]):
"""Delete"""
parser = argparse.ArgumentParser(
add_help=False,
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
prog="delete",
description="Delete a routine on the cloud.",
)
parser.add_argument(
"-n",
"--name",
type=str,
dest="name",
help="The name of the routine",
required="-h" not in other_args,
nargs="+",
)
if other_args and "-" not in other_args[0][0]:
other_args.insert(0, "-n")
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
name = " ".join(ns_parser.name)
response = Hub.delete_routine(
auth_header=User.get_auth_header(),
name=name,
)
if response and response.status_code == 200:
if name in self.REMOTE_CHOICES:
self.REMOTE_CHOICES.remove(name)
self.update_runtime_choices()

29
openbb_terminal/account/account_model.py
View File

@@ -1,6 +1,9 @@
from pathlib import Path
from typing import Any, Dict, Tuple
import numpy as np
import pandas as pd
from openbb_terminal import (
config_plot as cfg_plot,
config_terminal as cfg,
@@ -145,3 +148,29 @@ def get_var_diff(obj, name, value) -> Tuple[Any, Any]:
return current_value, cast_value
return None, None
def get_routines_info(response) -> Tuple[pd.DataFrame, int, int]:
"""Get the routines list.
Parameters
----------
response : requests.Response
The response.
Returns
-------
Tuple[pd.DataFrame, int, int]
The routines list, the current page and the total number of pages.
"""
df = pd.DataFrame()
if response and response.status_code == 200:
data = response.json()
page = data.get("page", 1)
pages = data.get("pages", 1)
items = data.get("items", [])
if items:
df = pd.DataFrame(items)
df.index = np.arange(1, len(df) + 1)
return df, page, pages

28
openbb_terminal/account/account_view.py Normal file
View File

@@ -0,0 +1,28 @@
import pandas as pd
from openbb_terminal.helper_funcs import print_rich_table
def display_routines_list(df: pd.DataFrame, page: int, pages: int):
"""Display the routines list.
Parameters
----------
df : pd.DataFrame
The routines list.
page : int
The current page.
pages : int
The total number of pages.
"""
title = f"Available routines - page {page}"
if pages:
title += f" of {pages}"
print_rich_table(
df=df,
title=title,
headers=["Name", "Description"],
show_index=True,
index_name="#",
)

4
openbb_terminal/miscellaneous/i18n/en.yml vendored
View File

@@ -22,9 +22,11 @@ en:
account/sync: Turns the cloud synchronization on/off
account/pull: Pull data from cloud
account/clear: Clear data from cloud
account/_routines_: Routines
account/_routines_: Cloud storage of routines
account/list: List available routines
account/upload: Upload routine
account/download: Download routine
account/delete: Delete routine
keys/_keys_: This menu allows you to set your own API keys. Type 'about' for more information
keys/mykeys: get your defined keys
keys/_status_: Status

173
openbb_terminal/session/hub_model.py
View File

@@ -158,7 +158,6 @@ def fetch_user_configs(
Optional[requests.Response]
The response from the get request.
"""
token_type = session.get("token_type", "")
token = session.get("access_token", "")
@@ -168,7 +167,7 @@ def fetch_user_configs(
headers={"Authorization": f"{token_type.title()} {token}"},
timeout=timeout,
)
if response.status_code not in [200, 401]:
if response.status_code not in [200, 401]: # 401: invalid token
console.print("[red]\nFailed to fetch configurations.[/red]")
return response
except requests.exceptions.ConnectionError:
@@ -212,7 +211,6 @@ def patch_user_configs(
Optional[requests.Response]
The response from the patch request.
"""
if type_ not in ["keys", "settings"]:
console.print("[red]\nInvalid patch type.[/red]")
return None
@@ -289,7 +287,9 @@ def clear_user_configs(
def upload_routine(
auth_header: str,
name: str = "",
description: str = "",
routine: str = "",
override: bool = False,
base_url=BASE_URL,
timeout: int = TIMEOUT,
) -> Optional[requests.Response]:
@@ -298,18 +298,29 @@ def upload_routine(
Parameters
----------
auth_header : str
The authorization header, e.g. "Bearer <token>".
name : str
The name of the routine.
routine : str
The routine.
override : bool
Whether to override the routine if it already exists.
base_url : str
The base url, by default BASE_URL
timeout : int
The timeout, by default TIMEOUT
"""
data = {"name": name, "script": routine}
Returns
-------
Optional[requests.Response]
The response from the post request.
"""
data = {
"name": name,
"description": description,
"script": routine,
"override": override,
}
try:
response = requests.post(
@@ -320,8 +331,10 @@ def upload_routine(
)
if response.status_code == 200:
console.print("[green]Successfully uploaded your routine.[/green]")
else:
console.print("[red]Error uploading your routine.[/red]")
elif response.status_code != 409: # 409: routine already exists
console.print(
"[red]" + response.json().get("detail", "Unknown error.") + "[/red]"
)
return response
except requests.exceptions.ConnectionError:
console.print(f"\n{CONNECTION_ERROR_MSG}")
@@ -334,10 +347,144 @@ def upload_routine(
return None
def download_routine() -> Optional[requests.Response]:
"""Download a routine from the server."""
def download_routine(
auth_header: str,
name: str = "",
base_url=BASE_URL,
timeout: int = TIMEOUT,
) -> Optional[requests.Response]:
"""Download a routine from the server.
# TODO: Implement test on `test_hub_model.py` when this is implemented.
Parameters
----------
auth_header : str
The authorization header, e.g. "Bearer <token>".
name : str
The name of the routine.
base_url : str
The base url, by default BASE_URL
timeout : int
The timeout, by default TIMEOUT
console.print("[red]Not implemented yet.[/red]")
return None
Returns
-------
Optional[requests.Response]
The response from the get request.
"""
try:
response = requests.get(
headers={"Authorization": auth_header},
url=base_url + "terminal/script/" + name,
timeout=timeout,
)
if response.status_code == 404:
console.print("[red]Routine not found.[/red]")
elif response.status_code != 200:
console.print("[red]Failed to download your routine.[/red]")
return response
except requests.exceptions.ConnectionError:
console.print(f"\n{CONNECTION_ERROR_MSG}")
return None
except requests.exceptions.Timeout:
console.print(f"\n{CONNECTION_TIMEOUT_MSG}")
return None
except Exception:
console.print("[red]Failed to download your routine.[/red]")
return None
def delete_routine(
auth_header: str,
name: str = "",
base_url=BASE_URL,
timeout: int = TIMEOUT,
) -> Optional[requests.Response]:
"""Delete a routine from the server.
Parameters
----------
auth_header : str
The authorization header, e.g. "Bearer <token>".
name : str
The name of the routine.
base_url : str
The base url, by default BASE_URL
timeout : int
The timeout, by default TIMEOUT
Returns
-------
Optional[requests.Response]
The response from the delete request.
"""
try:
response = requests.delete(
headers={"Authorization": auth_header},
url=base_url + "terminal/script/" + name,
timeout=timeout,
)
if response.status_code == 200:
console.print("[green]Successfully deleted your routine.[/green]")
elif response.status_code == 404:
console.print("[red]Routine not found.[/red]")
else:
console.print("[red]Failed to delete your routine.[/red]")
return response
except requests.exceptions.ConnectionError:
console.print(f"\n{CONNECTION_ERROR_MSG}")
return None
except requests.exceptions.Timeout:
console.print(f"\n{CONNECTION_TIMEOUT_MSG}")
return None
except Exception:
console.print("[red]Failed to delete your routine.[/red]")
return None
def list_routines(
auth_header: str,
page: int = 1,
size: int = 10,
base_url=BASE_URL,
timeout: int = TIMEOUT,
) -> Optional[requests.Response]:
"""List all routines from the server.
Parameters
----------
auth_header : str
The authorization header, e.g. "Bearer <token>".
page : int
The page number.
size : int
The number of routines per page.
base_url : str
The base url, by default BASE_URL
timeout : int
The timeout, by default TIMEOUT
Returns
-------
Optional[requests.Response]
The response from the get request.
"""
fields = "name%2Cdescription"
try:
response = requests.get(
headers={"Authorization": auth_header},
url=f"{base_url}terminal/script?fields={fields}&page={page}&size={size}",
timeout=timeout,
)
if response.status_code != 200:
console.print("[red]Failed to list your routines.[/red]")
return response
except requests.exceptions.ConnectionError:
console.print(f"\n{CONNECTION_ERROR_MSG}")
return None
except requests.exceptions.Timeout:
console.print(f"\n{CONNECTION_TIMEOUT_MSG}")
return None
except Exception:
console.print("[red]Failed to list your routines.[/red]")
return None

51
openbb_terminal/session/local_model.py
View File

@@ -16,6 +16,7 @@ from openbb_terminal.core.config.paths import (
USER_ROUTINES_DIRECTORY,
)
from openbb_terminal.rich_config import console
from openbb_terminal.session.user import User
SESSION_FILE_PATH = SETTINGS_DIRECTORY / "session.json"
@@ -189,21 +190,61 @@ def cast_set_attr(obj, name, value):
def get_routine(
name: str, routines_folder: Path = USER_ROUTINES_DIRECTORY
file_name: str, folder: Path = USER_ROUTINES_DIRECTORY
) -> Optional[str]:
"""Get the routine.
Returns
-------
str
file_name : str
The routine.
routines_folder : Path
folder : Path
The routines folder.
"""
try:
with open(routines_folder / name) as f:
user_folder = USER_ROUTINES_DIRECTORY / User.get_uuid()
if os.path.exists(user_folder / file_name):
file_path = user_folder / file_name
else:
file_path = folder / file_name
with open(file_path) as f:
routine = "".join(f.readlines())
return routine
except Exception:
console.print("[red]\nFailed to find routine.[/red]")
console.print("[red]Failed to find routine.[/red]")
return None
def save_routine(
file_name: str, routine: str, folder: Path = USER_ROUTINES_DIRECTORY
) -> Optional[Path]:
"""Save the routine.
Parameters
----------
file_name : str
The routine.
routine : str
The routine.
folder : Path
The routines folder.
Returns
-------
Optional[Path]
The path to the routine or None.
"""
try:
uuid = User.get_uuid()
user_folder = folder / uuid
if not os.path.exists(user_folder):
os.makedirs(user_folder)
file_path = user_folder / file_name
with open(file_path, "w") as f:
f.write(routine)
return user_folder / file_name
except Exception:
console.print("[red]\nFailed to save routine.[/red]")
return None

152
tests/openbb_terminal/account/test_account_controller.py
View File

@@ -33,6 +33,18 @@ CONFIGS = {
},
}
ROUTINES = {
"items": [
{"name": "scrip1", "description": "abc"},
{"name": "script2", "description": "def"},
{"name": "script3", "description": "ghi"},
],
"total": 3,
"page": 1,
"size": 10,
"pages": 1,
}
@pytest.fixture(scope="module")
def vcr_config():
@@ -241,12 +253,16 @@ def test_call_func_expect_queue(expected_queue, func, queue):
"call_sync",
"call_pull",
"call_clear",
"call_list",
"call_upload",
"call_download",
"call_delete",
],
)
def test_call_func_no_parser(func, mocker):
# MOCK PARSE_KNOWN_ARGS_AND_WARN
mocker.patch(
target="openbb_terminal.account.account_controller.AccountController.parse_simple_args",
target="openbb_terminal.account.account_controller.AccountController.parse_known_args_and_warn",
return_value=None,
)
controller = account_controller.AccountController(queue=None)
@@ -254,7 +270,7 @@ def test_call_func_no_parser(func, mocker):
func_result = getattr(controller, func)(other_args=list())
assert func_result is None
assert controller.queue == []
controller.parse_simple_args.assert_called_once()
controller.parse_known_args_and_warn.assert_called_once()
@pytest.mark.vcr
@@ -395,3 +411,135 @@ def test_call_clear(mocker, input_value):
else:
mock_clear_user_configs.assert_not_called()
assert controller.queue == []
@pytest.mark.record_stdout
def test_call_list(mocker):
controller = account_controller.AccountController(queue=None)
path_controller = "openbb_terminal.account.account_controller"
mocker.patch(
target=f"{path_controller}.User.get_auth_header",
return_value="Bearer 123",
)
mock_list_routines = mocker.patch(
target=f"{path_controller}.Hub.list_routines",
)
response = Response()
response.status_code = 200
response._content = json.dumps(ROUTINES).encode( # pylint: disable=protected-access
"utf-8"
)
mock_list_routines.return_value = response
controller.call_list(other_args=["--page", "1", "--size", "10"])
mock_list_routines.assert_called_once_with(
auth_header="Bearer 123", page=1, size=10
)
def test_call_upload(mocker):
controller = account_controller.AccountController(queue=None)
path_controller = "openbb_terminal.account.account_controller"
mocker.patch(
target=f"{path_controller}.User.get_auth_header",
return_value="Bearer 123",
)
mock_get_routine = mocker.patch(
target=f"{path_controller}.Local.get_routine",
return_value="do something",
)
mock_upload_routine = mocker.patch(
target=f"{path_controller}.Hub.upload_routine",
)
controller.call_upload(
other_args=[
"--file",
"script1.openbb",
"--description",
"abc",
"--name",
"script1",
]
)
mock_get_routine.assert_called_once_with(file_name="script1.openbb")
mock_upload_routine.assert_called_once_with(
auth_header="Bearer 123",
name="script1",
description="abc",
routine="do something",
)
@pytest.mark.record_stdout
def test_call_download(mocker):
controller = account_controller.AccountController(queue=None)
path_controller = "openbb_terminal.account.account_controller"
mocker.patch(
target=f"{path_controller}.User.get_auth_header",
return_value="Bearer 123",
)
mock_download_routine = mocker.patch(
target=f"{path_controller}.Hub.download_routine",
)
response = Response()
response.status_code = 200
content = {
"name": "script1",
"description": "abc",
"script": "do something",
}
response._content = json.dumps(content).encode( # pylint: disable=protected-access
"utf-8"
)
mock_download_routine.return_value = response
mock_save_routine = mocker.patch(
target=f"{path_controller}.Local.save_routine",
return_value="path_to_file",
)
controller.call_download(
other_args=[
"--name",
"script1",
]
)
mock_download_routine.assert_called_once_with(
auth_header="Bearer 123",
name="script1",
)
mock_save_routine.assert_called_once_with(
file_name="script1.openbb",
routine="do something",
)
def test_call_delete(mocker):
controller = account_controller.AccountController(queue=None)
path_controller = "openbb_terminal.account.account_controller"
mocker.patch(
target=f"{path_controller}.User.get_auth_header",
return_value="Bearer 123",
)
mock_delete_routine = mocker.patch(
target=f"{path_controller}.Hub.delete_routine",
)
controller.call_delete(
other_args=[
"--name",
"script1",
]
)
mock_delete_routine.assert_called_once_with(
auth_header="Bearer 123",
name="script1",
)

8
tests/openbb_terminal/account/test_account_model.py
View File

@@ -1,9 +1,11 @@
from pathlib import Path
from unittest.mock import patch
from openbb_terminal import config_plot as cfg_plot
from openbb_terminal import config_terminal as cfg
from openbb_terminal import feature_flags as obbff
from openbb_terminal import (
config_plot as cfg_plot,
config_terminal as cfg,
feature_flags as obbff,
)
from openbb_terminal.account import account_model
from openbb_terminal.core.config import paths

3
tests/openbb_terminal/account/txt/test_account_controller/test_call_download.txt vendored Normal file
View File

@@ -0,0 +1,3 @@
[info]Name:[/info] script1
[info]Description:[/info] abc
[info]Location:[/info] path_to_file

4
tests/openbb_terminal/account/txt/test_account_controller/test_call_list.txt vendored Normal file
View File

@@ -0,0 +1,4 @@
name description
1 scrip1 abc
2 script2 def
3 script3 ghi

6
tests/openbb_terminal/account/txt/test_account_controller/test_print_help.txt vendored
View File

@@ -3,3 +3,9 @@ Cloud storage of keys, settings and feature flags:
pull Pull data from cloud
clear Clear data from cloud
Cloud storage of routines:
list List available routines
upload Upload routine
download Download routine
delete Delete routine

190
tests/openbb_terminal/session/test_hub_model.py
View File

@@ -435,63 +435,189 @@ def test_clear_user_configs_exception():
assert kwargs["timeout"] == hub_model.TIMEOUT
def test_upload_routine_success():
@pytest.mark.parametrize(
"auth_header, name, description, routine, override, base_url, timeout, status_code",
[
("auth_header", "name", "description", "routine", True, "base_url", 10, 200),
("auth_header", "name", "description", "routine", False, "base_url", 10, 400),
],
)
def test_upload_routine(
auth_header, name, description, routine, override, base_url, timeout, status_code
):
mock_response = MagicMock(spec=requests.Response)
mock_response.status_code = 200
mock_response.status_code = status_code
with patch(
"openbb_terminal.session.hub_model.requests.post", return_value=mock_response
) as requests_post_mock:
result = hub_model.upload_routine("auth_header", "name", "routine")
) as requests_get_mock:
result = hub_model.upload_routine(
auth_header=auth_header,
name=name,
description=description,
routine=routine,
override=override,
base_url=base_url,
timeout=timeout,
)
assert result.status_code == mock_response.status_code
requests_post_mock.assert_called_once()
_, kwargs = requests_post_mock.call_args
assert kwargs["url"] == hub_model.BASE_URL + "terminal/script"
assert kwargs["headers"] == {"Authorization": "auth_header"}
assert kwargs["json"] == {"name": "name", "script": "routine"}
assert kwargs["timeout"] == hub_model.TIMEOUT
requests_get_mock.assert_called_once()
_, kwargs = requests_get_mock.call_args
assert kwargs["url"] == base_url + "terminal/script"
assert kwargs["headers"] == {"Authorization": auth_header}
assert kwargs["timeout"] == timeout
def test_upload_routine_failure():
@pytest.mark.parametrize(
"side_effect",
[
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
Exception,
],
)
def test_upload_routine_error(side_effect):
with patch(
"openbb_terminal.session.hub_model.requests.post",
side_effect=side_effect,
):
result = hub_model.upload_routine("auth_header", "name", "routine")
assert result is None
@pytest.mark.parametrize(
"auth_header, name, base_url, timeout, status_code",
[
("auth_header", "name", "base_url", 10, 200),
("other_header", "other name", "other_url", 10, 400),
],
)
def test_download_routine(auth_header, name, base_url, timeout, status_code):
mock_response = MagicMock(spec=requests.Response)
mock_response.status_code = 400
mock_response.status_code = status_code
with patch(
"openbb_terminal.session.hub_model.requests.post", return_value=mock_response
) as requests_post_mock:
result = hub_model.upload_routine("auth_header", "name", "routine")
"openbb_terminal.session.hub_model.requests.get", return_value=mock_response
) as requests_get_mock:
result = hub_model.download_routine(
auth_header=auth_header, name=name, base_url=base_url, timeout=timeout
)
assert result.status_code == mock_response.status_code
requests_post_mock.assert_called_once()
_, kwargs = requests_post_mock.call_args
assert kwargs["url"] == hub_model.BASE_URL + "terminal/script"
assert kwargs["headers"] == {"Authorization": "auth_header"}
assert kwargs["json"] == {"name": "name", "script": "routine"}
assert kwargs["timeout"] == hub_model.TIMEOUT
requests_get_mock.assert_called_once()
_, kwargs = requests_get_mock.call_args
assert kwargs["url"] == base_url + f"terminal/script/{name}"
assert kwargs["headers"] == {"Authorization": auth_header}
assert kwargs["timeout"] == timeout
def test_upload_routine_connection_error():
@pytest.mark.parametrize(
"side_effect",
[
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
Exception,
],
)
def test_download_routine_error(side_effect):
with patch(
"openbb_terminal.session.hub_model.requests.post",
side_effect=requests.exceptions.ConnectionError,
"openbb_terminal.session.hub_model.requests.get",
side_effect=side_effect,
):
result = hub_model.upload_routine("auth_header", "name", "routine")
result = hub_model.download_routine("auth_header", "name", "routine")
assert result is None
def test_upload_routine_timeout():
@pytest.mark.parametrize(
"auth_header, name, base_url, timeout, status_code",
[
("auth_header", "name", "base_url", 10, 200),
("other_header", "other name", "other_url", 10, 400),
],
)
def test_delete_routine(auth_header, name, base_url, timeout, status_code):
mock_response = MagicMock(spec=requests.Response)
mock_response.status_code = status_code
with patch(
"openbb_terminal.session.hub_model.requests.post",
side_effect=requests.exceptions.Timeout,
"openbb_terminal.session.hub_model.requests.delete", return_value=mock_response
) as requests_get_mock:
result = hub_model.delete_routine(
auth_header=auth_header, name=name, base_url=base_url, timeout=timeout
)
assert result.status_code == mock_response.status_code
requests_get_mock.assert_called_once()
_, kwargs = requests_get_mock.call_args
assert kwargs["url"] == base_url + f"terminal/script/{name}"
assert kwargs["headers"] == {"Authorization": auth_header}
assert kwargs["timeout"] == timeout
@pytest.mark.parametrize(
"side_effect",
[
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
Exception,
],
)
def test_delete_routine_error(side_effect):
with patch(
"openbb_terminal.session.hub_model.requests.delete",
side_effect=side_effect,
):
result = hub_model.upload_routine("auth_header", "name", "routine")
result = hub_model.delete_routine("auth_header", "name")
assert result is None
def test_upload_routine_exception():
@pytest.mark.parametrize(
"auth_header, page, size, base_url, timeout, status_code",
[
("auth_header", 1, 10, "base_url", 10, 200),
("other_header", 2, 20, "other_url", 10, 400),
],
)
def test_list_routines(auth_header, page, size, base_url, timeout, status_code):
mock_response = MagicMock(spec=requests.Response)
mock_response.status_code = status_code
with patch(
"openbb_terminal.session.hub_model.requests.post", side_effect=Exception
"openbb_terminal.session.hub_model.requests.get", return_value=mock_response
) as requests_get_mock:
result = hub_model.list_routines(
auth_header=auth_header,
page=page,
size=size,
base_url=base_url,
timeout=timeout,
)
assert result.status_code == mock_response.status_code
requests_get_mock.assert_called_once()
_, kwargs = requests_get_mock.call_args
assert (
kwargs["url"]
== base_url
+ f"terminal/script?fields=name%2Cdescription&page={page}&size={size}"
)
assert kwargs["headers"] == {"Authorization": auth_header}
assert kwargs["timeout"] == timeout
@pytest.mark.parametrize(
"side_effect",
[
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
Exception,
],
)
def test_list_routines_error(side_effect):
with patch(
"openbb_terminal.session.hub_model.requests.get",
side_effect=side_effect,
):
result = hub_model.upload_routine("auth_header", "name", "routine")
result = hub_model.list_routines(auth_header="Bearer 123", page=1, size=10)
assert result is None

122
tests/openbb_terminal/session/test_local_model.py
View File

@@ -197,3 +197,125 @@ def test_apply_configs_sync(sync: str):
assert paths.USER_DATA_DIRECTORY == Path("some/path/to/user/data")
assert cfg.API_KEY_ALPHAVANTAGE == "test_av"
assert cfg.API_FRED_KEY == "test_fred"
@pytest.mark.parametrize(
"exists",
[
False,
True,
],
)
def test_get_routine(mocker, exists: bool):
file_name = "test_routine.openbb"
uuid = "test_uuid"
routine = "do something"
user_mock = mocker.patch(
"openbb_terminal.session.local_model.User.get_uuid", return_value=uuid
)
exists_mock = mocker.patch(
"openbb_terminal.session.local_model.os.path.exists", return_value=exists
)
open_mock = mocker.patch(
"openbb_terminal.session.local_model.open",
mock_open(read_data=json.dumps(routine)),
)
assert local_model.get_routine(file_name=file_name) == json.dumps(routine)
user_mock.assert_called_once()
exists_mock.assert_called_with(
local_model.USER_ROUTINES_DIRECTORY / uuid / file_name
)
if exists:
open_mock.assert_called_with(
local_model.USER_ROUTINES_DIRECTORY / uuid / file_name
)
else:
open_mock.assert_called_with(local_model.USER_ROUTINES_DIRECTORY / file_name)
def test_get_routine_exception(mocker):
file_name = "test_routine.openbb"
uuid = "test_uuid"
user_mock = mocker.patch(
"openbb_terminal.session.local_model.User.get_uuid", return_value=uuid
)
exists_mock = mocker.patch("openbb_terminal.session.local_model.os.path.exists")
open_mock = mocker.patch(
"openbb_terminal.session.local_model.open",
side_effect=Exception("test exception"),
)
assert local_model.get_routine(file_name=file_name) is None
user_mock.assert_called_once()
exists_mock.assert_called_with(
local_model.USER_ROUTINES_DIRECTORY / uuid / file_name
)
open_mock.assert_called_with(local_model.USER_ROUTINES_DIRECTORY / uuid / file_name)
@pytest.mark.parametrize(
"exists",
[
False,
True,
],
)
def test_save_routine(mocker, exists: bool):
file_name = "test_routine.openbb"
uuid = "test_uuid"
routine = "do something"
user_mock = mocker.patch(
"openbb_terminal.session.local_model.User.get_uuid", return_value=uuid
)
exists_mock = mocker.patch(
"openbb_terminal.session.local_model.os.path.exists", return_value=exists
)
makedirs_mock = mocker.patch("openbb_terminal.session.local_model.os.makedirs")
open_mock = mocker.patch(
"openbb_terminal.session.local_model.open",
)
assert (
local_model.save_routine(file_name=file_name, routine=routine)
== local_model.USER_ROUTINES_DIRECTORY / uuid / file_name
)
user_mock.assert_called_once()
exists_mock.assert_called_with(local_model.USER_ROUTINES_DIRECTORY / uuid)
if exists:
makedirs_mock.assert_not_called()
else:
makedirs_mock.assert_called_with(local_model.USER_ROUTINES_DIRECTORY / uuid)
open_mock.assert_called_with(
local_model.USER_ROUTINES_DIRECTORY / uuid / file_name, "w"
)
def test_save_routine_exception(mocker):
file_name = "test_routine.openbb"
uuid = "test_uuid"
routine = "do something"
user_mock = mocker.patch(
"openbb_terminal.session.local_model.User.get_uuid", return_value=uuid
)
exists_mock = mocker.patch("openbb_terminal.session.local_model.os.path.exists")
open_mock = mocker.patch(
"openbb_terminal.session.local_model.open",
side_effect=Exception("test exception"),
)
assert local_model.save_routine(file_name=file_name, routine=routine) is None
user_mock.assert_called_once()
exists_mock.assert_called_with(local_model.USER_ROUTINES_DIRECTORY / uuid)
open_mock.assert_called_with(
local_model.USER_ROUTINES_DIRECTORY / uuid / file_name, "w"
)

4
tests/openbb_terminal/session/test_session_model.py
View File

@@ -113,7 +113,9 @@ def test_login_success_response():
):
response = Response()
response.status_code = 200
response._content = json.dumps(CONFIGS) # pylint: disable=protected-access
response._content = json.dumps( # pylint: disable=protected-access
CONFIGS
).encode("utf-8")
mock_fetch_user_configs.return_value = response
assert session_model.login(TEST_SESSION) == session_model.LoginStatus.SUCCESS