jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-28 13:01:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Bug] Fix degiro controller (#1160)

Browse Source 
* fixed null calls

* Added exception handling

* Pep8-ed degiro-controller.py

* Added docstring to command_call

* Removed one letter variable

* removed try_except

* fixed null calls

* Added exception handling

* Added docstring to command_call

* Adapted exception handling to new project's structure

* Updating brokers : fixing cli issues

* Updating tests : portfolio/brokers/degiro

* Updating broker : degiro

* Updating tests : portfolio/brokers/degiro

* Updating brokers : degiro

* Updating tests : portfolio/brokers/degiro

Co-authored-by: Chavithra PARANA <chavithra@gmail.com>
This commit is contained in:
Juan Toca Mateo
2022-01-28 06:10:08 +01:00
committed by GitHub
parent 72de21f051
commit ad85411542
11 changed files with 745 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
gamestonk_terminal/portfolio/brokers/degiro/degiro_controller.py
View File

@@ -2,7 +2,6 @@ import argparse
from typing import List
from prompt_toolkit.completion import NestedCompleter
from gamestonk_terminal.rich_config import console
import gamestonk_terminal.config_terminal as config
from gamestonk_terminal import feature_flags as gtff
@@ -30,7 +29,7 @@ class DegiroController(BaseController):
"topnews",
"update",
]
PATH = "/portfolio/bro/derigo/"
PATH = "/portfolio/bro/degiro/"
def __init__(self, queue: List[str] = None):
"""Constructor"""
@@ -44,25 +43,9 @@ class DegiroController(BaseController):
def print_help(self):
"""Print help."""
help_text = """
login connect to degiro's api
logout disconnect from degiro's api
DegiroView.help_display()
hold view holdings
lookup view search for a product by name
create create an order
update update an order
cancel cancel an order using the id
pending view pending orders
companynews view news about a company with it's isin
lastnews view latest news
topnews view top news preview
"""
console.print(text=help_text, menu="Portfolio - Brokers - Degiro")
def cancel(self, other_args: List[str]):
def call_cancel(self, other_args: List[str]):
"""Cancel an order using the `id`."""
# PARSE ARGS
@@ -79,7 +62,7 @@ class DegiroController(BaseController):
self.__degiro_view.cancel(ns_parser=ns_parser)
def companynews(self, other_args: List[str]):
def call_companynews(self, other_args: List[str]):
"""Display news related to a company using its ISIN."""
# PARSE ARGS
@@ -96,7 +79,7 @@ class DegiroController(BaseController):
self.__degiro_view.companynews(ns_parser=ns_parser)
def create(self, other_args: List[str]):
def call_create(self, other_args: List[str]):
"""Create an order."""
# PARSE ARGS
@@ -174,7 +157,7 @@ class DegiroController(BaseController):
self.__degiro_view.create(ns_parser=ns_parser)
def hold(self, other_args):
def call_hold(self, other_args):
"""Display held products."""
# PARSE ARGS
@@ -186,7 +169,7 @@ class DegiroController(BaseController):
self.__degiro_view.hold(ns_parser=ns_parser)
def lastnews(self, other_args: List[str]):
def call_lastnews(self, other_args: List[str]):
"""Display latest news."""
# PARSE ARGS
@@ -206,7 +189,7 @@ class DegiroController(BaseController):
self.__degiro_view.lastnews(ns_parser=ns_parser)
def login(self, other_args: List[str]):
def call_login(self, other_args: List[str]):
"""Connect to Degiro's API."""
# PARSE ARGS
@@ -246,7 +229,7 @@ class DegiroController(BaseController):
self.__degiro_view.login(ns_parser=ns_parser)
def logout(self, other_args: List[str]):
def call_logout(self, other_args: List[str]):
"""Log out from Degiro's API."""
# PARSE ARGS
@@ -258,7 +241,7 @@ class DegiroController(BaseController):
self.__degiro_view.logout(ns_parser=ns_parser)
def lookup(self, other_args: List[str]):
def call_lookup(self, other_args: List[str]):
"""Search for products by their name."""
# PARSING ARGS
@@ -289,7 +272,7 @@ class DegiroController(BaseController):
self.__degiro_view.lookup(ns_parser=ns_parser)
def pending(self, other_args: List[str]):
def call_pending(self, other_args: List[str]):
"""Display pending orders."""
# PARSING ARGS
@@ -301,7 +284,7 @@ class DegiroController(BaseController):
self.__degiro_view.pending(ns_parser=ns_parser)
def topnews(self, other_args: List[str]):
def call_topnews(self, other_args: List[str]):
"""Display top news."""
# PARSING ARGS
@@ -313,7 +296,7 @@ class DegiroController(BaseController):
self.__degiro_view.topnews(ns_parser=ns_parser)
def update(self, other_args: List[str]):
def call_update(self, other_args: List[str]):
"""Update an order."""
# PARSING ARGS

2
gamestonk_terminal/portfolio/brokers/degiro/degiro_model.py
View File

@@ -249,7 +249,7 @@ class DegiroModel:
return news
def login(self, credentials: Credentials):
self.__trading_api = TradingAPI(credentials=credentials)
self.__trading_api.credentials.CopyFrom(credentials)
self.__trading_api.connect()
self.__setup_extra_credentials()

4
gamestonk_terminal/portfolio/brokers/degiro/degiro_view.py
View File

@@ -290,6 +290,10 @@ class DegiroView:
credentials.username = ns_parser.username
credentials.password = ns_parser.password
if "REPLACE_ME" in (credentials.username, credentials.username):
print("You didn't provide me with Degiro's credentials. Login failed.")
return
if ns_parser.otp is not None:
credentials.one_time_password = ns_parser.otp
if ns_parser.topt_secret is not None:

0
tests/gamestonk_terminal/portfolio/brokers/__init__.py Normal file
View File

0
tests/gamestonk_terminal/portfolio/brokers/degiro/test___init__.py Normal file
View File

0
tests/gamestonk_terminal/portfolio/brokers/degiro/test_degiro_api.py Normal file
View File

314
tests/gamestonk_terminal/portfolio/brokers/degiro/test_degiro_controller.py Normal file
View File

@@ -0,0 +1,314 @@
# IMPORTATION STANDARD
import os
# IMPORTATION THIRDPARTY
import pytest
# IMPORTATION INTERNAL
from gamestonk_terminal.portfolio.brokers.degiro import degiro_controller
# pylint: disable=E1101
# pylint: disable=W0603
# pylint: disable=E1111
@pytest.mark.vcr(record_mode="none")
@pytest.mark.parametrize(
"queue, expected",
[
(["load", "help"], []),
(["quit", "help"], ["help"]),
],
)
def test_menu_with_queue(expected, mocker, queue):
path_controller = "gamestonk_terminal.portfolio.brokers.degiro.degiro_controller"
# MOCK SWITCH
mocker.patch(
target=f"{path_controller}.DegiroController.switch",
return_value=["quit"],
)
result_menu = degiro_controller.DegiroController(queue=queue).menu()
assert result_menu == expected
@pytest.mark.vcr(record_mode="none")
def test_menu_without_queue_completion(mocker):
path_controller = "gamestonk_terminal.portfolio.brokers.degiro.degiro_controller"
# ENABLE AUTO-COMPLETION : HELPER_FUNCS.MENU
mocker.patch(
target="gamestonk_terminal.feature_flags.USE_PROMPT_TOOLKIT",
new=True,
)
mocker.patch(
target="gamestonk_terminal.parent_classes.session",
)
mocker.patch(
target="gamestonk_terminal.parent_classes.session.prompt",
return_value="quit",
)
# DISABLE AUTO-COMPLETION : CONTROLLER.COMPLETER
mocker.patch.object(
target=degiro_controller.gtff,
attribute="USE_PROMPT_TOOLKIT",
new=True,
)
mocker.patch(
target=f"{path_controller}.session",
)
mocker.patch(
target=f"{path_controller}.session.prompt",
return_value="quit",
)
result_menu = degiro_controller.DegiroController(queue=None).menu()
assert result_menu == []
@pytest.mark.vcr(record_mode="none")
@pytest.mark.parametrize(
"mock_input",
["help", "homee help", "home help", "mock"],
)
def test_menu_without_queue_sys_exit(mock_input, mocker):
path_controller = "gamestonk_terminal.portfolio.brokers.degiro.degiro_controller"
# DISABLE AUTO-COMPLETION
mocker.patch.object(
target=degiro_controller.gtff,
attribute="USE_PROMPT_TOOLKIT",
new=False,
)
mocker.patch(
target=f"{path_controller}.session",
return_value=None,
)
# MOCK USER INPUT
mocker.patch("builtins.input", return_value=mock_input)
# MOCK SWITCH
class SystemExitSideEffect:
def __init__(self):
self.first_call = True
def __call__(self, *args, **kwargs):
if self.first_call:
self.first_call = False
raise SystemExit()
return ["quit"]
mock_switch = mocker.Mock(side_effect=SystemExitSideEffect())
mocker.patch(
target=f"{path_controller}.DegiroController.switch",
new=mock_switch,
)
result_menu = degiro_controller.DegiroController(queue=None).menu()
assert result_menu == []
@pytest.mark.vcr(record_mode="none")
@pytest.mark.record_stdout
def test_print_help():
controller = degiro_controller.DegiroController(queue=None)
controller.print_help()
@pytest.mark.vcr(record_mode="none")
@pytest.mark.parametrize(
"an_input, expected_queue",
[
("", []),
("/help", ["quit", "quit", "quit", "help"]),
("help/help", ["help"]),
("q", ["quit"]),
("h", []),
(
"r",
["quit", "quit", "quit", "reset", "portfolio", "bro", "degiro"],
),
],
)
def test_switch(an_input, expected_queue):
controller = degiro_controller.DegiroController(queue=None)
queue = controller.switch(an_input=an_input)
assert queue == expected_queue
@pytest.mark.vcr(record_mode="none")
def test_call_cls(mocker):
mocker.patch("os.system")
controller = degiro_controller.DegiroController(queue=None)
controller.call_cls([])
assert controller.queue == []
os.system.assert_called_once_with("cls||clear")
@pytest.mark.vcr(record_mode="none")
@pytest.mark.parametrize(
"func, queue, expected_queue",
[
(
"call_exit",
[],
["quit", "quit", "quit", "quit"],
),
("call_exit", ["help"], ["quit", "quit", "quit", "quit", "help"]),
("call_home", [], ["quit", "quit", "quit"]),
("call_help", [], []),
("call_quit", [], ["quit"]),
("call_quit", ["help"], ["quit", "help"]),
(
"call_reset",
[],
["quit", "quit", "quit", "reset", "portfolio", "bro", "degiro"],
),
(
"call_reset",
["help"],
["quit", "quit", "quit", "reset", "portfolio", "bro", "degiro", "help"],
),
],
)
def test_call_func_expect_queue(expected_queue, func, queue):
controller = degiro_controller.DegiroController(queue=queue)
result = getattr(controller, func)([])
assert result is None
assert controller.queue == expected_queue
@pytest.mark.vcr(record_mode="none")
@pytest.mark.parametrize(
"tested_func, other_args, mocked_func, called_args, called_kwargs",
[
(
"call_cancel",
[
"MOCK_ORDER_ID",
],
"DegiroView.cancel",
[],
dict(),
),
(
"call_companynews",
[
"MOCK_ISIN",
],
"DegiroView.companynews",
[],
dict(),
),
(
"call_create",
[
"--action=buy",
"--symbol=MOCK_SYMBOL" "--price=1",
"--size=2",
],
"DegiroView.create",
[],
dict(),
),
(
"call_hold",
[],
"DegiroView.hold",
[],
dict(),
),
(
"call_lastnews",
[
"--limit=10",
],
"DegiroView.lastnews",
[],
dict(),
),
(
"call_login",
[
"--username=MOKC_USERNAME",
"--password=MOKC_PASSWORD",
"--topt-secret=MOCK_TOPT_SECRET",
],
"DegiroView.login",
[],
dict(),
),
(
"call_logout",
[],
"DegiroView.logout",
[],
dict(),
),
(
"call_lookup",
[
"MOCK_SEARCH_TEXT",
"--limit=1",
"--offset=2",
],
"DegiroView.lookup",
[],
dict(),
),
(
"call_pending",
[],
"DegiroView.pending",
[],
dict(),
),
(
"call_topnews",
[],
"DegiroView.topnews",
[],
dict(),
),
(
"call_update",
[
"--price=1",
],
"DegiroView.update",
[],
dict(),
),
],
)
def test_call_func(
tested_func, mocked_func, other_args, called_args, called_kwargs, mocker
):
path_controller = "gamestonk_terminal.portfolio.brokers.degiro.degiro_controller"
if mocked_func:
mock = mocker.Mock()
mocker.patch(
target=f"{path_controller}.{mocked_func}",
new=mock,
)
controller = degiro_controller.DegiroController(queue=None)
getattr(controller, tested_func)(other_args)
if called_args or called_kwargs:
mock.assert_called_once_with(*called_args, **called_kwargs)
else:
mock.assert_called_once()
else:
controller = degiro_controller.DegiroController(queue=None)
getattr(controller, tested_func)(other_args)

342
tests/gamestonk_terminal/portfolio/brokers/degiro/test_degiro_model.py Normal file
View File

@@ -0,0 +1,342 @@
# IMPORTATION STANDARD
# IMPORTATION THIRDPARTY
import pandas as pd
import pytest
from degiro_connector.trading.models import trading_pb2
# IMPORTATION INTERNAL
from gamestonk_terminal.portfolio.brokers.degiro.degiro_model import DegiroModel
# pylint: disable=E1101
# pylint: disable=W0603
# pylint: disable=E1111
@pytest.fixture(scope="module")
def vcr_config():
return {
"filter_headers": [
("User-Agent", None),
("cookie", None),
],
"filter_query_parameters": [
("intAccount", "MOCK_INT_ACCOUNT"),
("sessionId", "MOCK_SESSION_ID"),
],
"filter_post_data_parameters": [
("username", "MOCK_USERNAME"),
("password", "MOCK_PASSWORD"),
("oneTimePassword", "MOCK_ONE_TIME_PASSWORD"),
],
}
@pytest.mark.vcr(record_mode="none")
@pytest.mark.parametrize(
"func, kwargs, mocked_func, mock_return",
[
(
"cancel",
dict(order_id="MOCK_ORDER_ID"),
"delete_order",
True,
),
(
"companynews",
dict(isin="MOCK_ISIN"),
"get_news_by_company",
trading_pb2.NewsByCompany(),
),
(
"create_check",
dict(order=trading_pb2.Order()),
"check_order",
trading_pb2.Order.CheckingResponse(),
),
(
"create_confirm",
dict(confirmation_id="MOCK_CONFIRMATION_ID", order=trading_pb2.Order()),
"confirm_order",
trading_pb2.Order.ConfirmationResponse(),
),
(
"lastnews",
dict(limit=1),
"get_latest_news",
trading_pb2.LatestNews(),
),
(
"logout",
dict(),
"logout",
True,
),
(
"lookup",
dict(limit=1, offset=2, search_text="MOCK_SEARCH_TXT"),
"product_search",
trading_pb2.ProductSearch(),
),
(
"topnews",
dict(),
"get_top_news_preview",
trading_pb2.TopNewsPreview(),
),
(
"update",
dict(order=trading_pb2.Order()),
"update_order",
trading_pb2.Update.Orders(),
),
],
)
def test_call_func(func, kwargs, mocked_func, mock_return, mocker):
model = DegiroModel()
# MOCK MOCKED_FUNC
mock_func = mocker.Mock(return_value=mock_return)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute=mocked_func,
new=mock_func,
)
result = getattr(model, func)(**kwargs)
mock_func.assert_called_once()
assert isinstance(result, type(mock_return))
@pytest.mark.vcr(record_mode="none")
def test_pending(mocker):
model = DegiroModel()
# MOCK API FUNCTION
mock_func = mocker.Mock(return_value=trading_pb2.Update())
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="get_update",
new=mock_func,
)
result = model.pending()
mock_func.assert_called_once()
assert isinstance(result, trading_pb2.Update.Orders)
@pytest.mark.vcr(record_mode="none")
def test_update_pending_order(mocker):
model = DegiroModel()
# MOCK API FUNCTION
mock_update = trading_pb2.Update()
mock_update.orders.values.extend(
[
trading_pb2.Order(
id="MOCK_ORDER_ID",
product_id=1,
size=2,
price=3,
),
]
)
mock_func = mocker.Mock(return_value=mock_update)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="get_update",
new=mock_func,
)
result = model.update_pending_order(order_id="MOCK_ORDER_ID")
mock_func.assert_called_once()
assert isinstance(result, trading_pb2.Order)
@pytest.mark.vcr(record_mode="none")
def test_update_pending_order_no_order_id(mocker):
model = DegiroModel()
# MOCK API FUNCTION
mock_update = trading_pb2.Update()
mock_update.orders.values.extend(
[
trading_pb2.Order(
product_id=1,
size=2,
price=3,
),
]
)
mock_func = mocker.Mock(return_value=mock_update)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="get_update",
new=mock_func,
)
result = model.update_pending_order(order_id="MOCK_ORDER_ID")
mock_func.assert_called_once()
assert result is None
@pytest.mark.vcr(record_mode="none")
def test_update_pending_order_none(mocker):
model = DegiroModel()
# MOCK API FUNCTION
mock_update = trading_pb2.Update()
mock_func = mocker.Mock(return_value=mock_update)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="get_update",
new=mock_func,
)
result = model.update_pending_order(order_id="MOCK_ORDER_ID")
mock_func.assert_called_once()
assert result is None
@pytest.mark.vcr(record_mode="none")
def test_hold_positions_empty_update(mocker):
model = DegiroModel()
# MOCK API FUNCTION
mock_update = trading_pb2.Update()
mock_func = mocker.Mock(return_value=mock_update)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="get_update",
new=mock_func,
)
result = model.hold_positions()
assert isinstance(result, pd.DataFrame)
@pytest.mark.vcr(record_mode="none")
def test_login_default_credentials():
model = DegiroModel()
credentials = model.login_default_credentials()
assert isinstance(credentials, trading_pb2.Credentials)
@pytest.mark.vcr(record_mode="none")
def test_login(mocker):
model = DegiroModel()
# MOCK CONNECT
mock_connect = mocker.Mock()
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="connect",
new=mock_connect,
)
# MOCK GET_CLIENT_DETAILS
mock_client_details_table = {
"data": {
"intAccount": 12345,
}
}
mock_get_client_details = mocker.Mock(return_value=mock_client_details_table)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="get_client_details",
new=mock_get_client_details,
)
model.login(credentials=trading_pb2.Credentials())
mock_connect.assert_called_once()
mock_get_client_details.assert_called_once()
@pytest.mark.vcr(record_mode="none")
@pytest.mark.parametrize(
"kwargs, expected",
[
(
dict(price=10, size=None, up_to=22),
2,
),
(
dict(price=10, size=2, up_to=None),
2,
),
],
)
def test_create_calculate_size(expected, kwargs):
model = DegiroModel()
result = model.create_calculate_size(**kwargs)
assert result == expected
@pytest.mark.vcr(record_mode="none")
def test_create_calculate_product_id_product_id():
model = DegiroModel()
result = model.create_calculate_product_id(product=1, symbol=None)
assert result == 1
@pytest.mark.vcr(record_mode="none")
def test_create_calculate_product_id_product_symbol(mocker):
model = DegiroModel()
result = model.create_calculate_product_id(product=1, symbol=None)
assert result == 1
# MOCK CONNECT
product_search = trading_pb2.ProductSearch()
product = product_search.products.add()
product["id"] = 1
product["symbol"] = "MOCK_SYMBOL"
mock_product_search = mocker.Mock(return_value=product_search)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="product_search",
new=mock_product_search,
)
result = model.create_calculate_product_id(product=None, symbol="MOCK_SYMBOL")
assert result == 1
@pytest.mark.vcr(record_mode="none")
def test_create_calculate_product_id_product_symbol_no_result(mocker):
model = DegiroModel()
result = model.create_calculate_product_id(product=1, symbol=None)
assert result == 1
# MOCK CONNECT
product_search = trading_pb2.ProductSearch()
product = product_search.products.add()
product["id"] = 1
product["symbol"] = "MOCK_WRONG_SYMBOL"
mock_product_search = mocker.Mock(return_value=product_search)
mocker.patch.object(
target=model.__dict__["_DegiroModel__trading_api"],
attribute="product_search",
new=mock_product_search,
)
result = model.create_calculate_product_id(product=None, symbol="MOCK_SYMBOL")
assert result is None

37
tests/gamestonk_terminal/portfolio/brokers/degiro/test_degiro_view.py Normal file
View File

@@ -0,0 +1,37 @@
# IMPORTATION STANDARD
# IMPORTATION THIRDPARTY
import pytest
# IMPORTATION INTERNAL
from gamestonk_terminal.portfolio.brokers.degiro.degiro_view import DegiroView
# pylint: disable=E1101
# pylint: disable=W0603
# pylint: disable=E1111
@pytest.fixture(scope="module")
def vcr_config():
return {
"filter_headers": [
("User-Agent", None),
("cookie", None),
],
"filter_query_parameters": [
("intAccount", "MOCK_INT_ACCOUNT"),
("sessionId", "MOCK_SESSION_ID"),
],
"filter_post_data_parameters": [
("username", "MOCK_USERNAME"),
("password", "MOCK_PASSWORD"),
("oneTimePassword", "MOCK_ONE_TIME_PASSWORD"),
],
}
@pytest.mark.vcr(record_mode="none")
@pytest.mark.record_stdout
def test_help_display():
DegiroView.help_display()

17
tests/gamestonk_terminal/portfolio/brokers/degiro/txt/test_degiro_controller/test_print_help.txt vendored Normal file
View File

@@ -0,0 +1,17 @@
Degiro:
login connect to degiro's api
logout disconnect from degiro's api
hold view holdings
lookup view search for a product by name
create create an order
update update an order
cancel cancel an order using the id
pending view pending orders
companynews view news about a company with it's isin
lastnews view latest news
topnews view top news preview

17
tests/gamestonk_terminal/portfolio/brokers/degiro/txt/test_degiro_view/test_help_display.txt vendored Normal file
View File

@@ -0,0 +1,17 @@
Degiro:
login connect to degiro's api
logout disconnect from degiro's api
hold view holdings
lookup view search for a product by name
create create an order
update update an order
cancel cancel an order using the id
pending view pending orders
companynews view news about a company with it's isin
lastnews view latest news
topnews view top news preview