freqtrade_origin/tests/rpc/test_rpc_manager.py

188 lines
7.5 KiB
Python
Raw Normal View History

2018-07-29 14:09:44 +00:00
# pragma pylint: disable=missing-docstring, C0103
2018-02-13 03:45:59 +00:00
import logging
2020-09-28 17:43:15 +00:00
import time
2018-02-13 03:45:59 +00:00
from unittest.mock import MagicMock
2020-09-28 17:43:15 +00:00
from freqtrade.rpc import RPCManager, RPCMessageType
from tests.conftest import get_patched_freqtradebot, log_has
2018-02-13 03:45:59 +00:00
def test__init__(mocker, default_conf) -> None:
default_conf['telegram']['enabled'] = False
2018-02-13 03:45:59 +00:00
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
2018-02-13 03:45:59 +00:00
assert rpc_manager.registered_modules == []
def test_init_telegram_disabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
default_conf['telegram']['enabled'] = False
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
2018-02-13 03:45:59 +00:00
2019-08-11 18:17:22 +00:00
assert not log_has('Enabling rpc.telegram ...', caplog)
2018-02-13 03:45:59 +00:00
assert rpc_manager.registered_modules == []
def test_init_telegram_enabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock())
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
2018-02-13 03:45:59 +00:00
2019-08-11 18:17:22 +00:00
assert log_has('Enabling rpc.telegram ...', caplog)
2018-02-13 03:45:59 +00:00
len_modules = len(rpc_manager.registered_modules)
assert len_modules == 1
assert 'telegram' in [mod.name for mod in rpc_manager.registered_modules]
2018-02-13 03:45:59 +00:00
def test_cleanup_telegram_disabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
telegram_mock = mocker.patch('freqtrade.rpc.telegram.Telegram.cleanup', MagicMock())
default_conf['telegram']['enabled'] = False
2018-02-13 03:45:59 +00:00
freqtradebot = get_patched_freqtradebot(mocker, default_conf)
2018-02-13 03:45:59 +00:00
rpc_manager = RPCManager(freqtradebot)
rpc_manager.cleanup()
2019-08-11 18:17:22 +00:00
assert not log_has('Cleaning up rpc.telegram ...', caplog)
2018-02-13 03:45:59 +00:00
assert telegram_mock.call_count == 0
def test_cleanup_telegram_enabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock())
telegram_mock = mocker.patch('freqtrade.rpc.telegram.Telegram.cleanup', MagicMock())
freqtradebot = get_patched_freqtradebot(mocker, default_conf)
2018-02-13 03:45:59 +00:00
rpc_manager = RPCManager(freqtradebot)
# Check we have Telegram as a registered modules
assert 'telegram' in [mod.name for mod in rpc_manager.registered_modules]
2018-02-13 03:45:59 +00:00
rpc_manager.cleanup()
2019-08-11 18:17:22 +00:00
assert log_has('Cleaning up rpc.telegram ...', caplog)
assert 'telegram' not in [mod.name for mod in rpc_manager.registered_modules]
2018-02-13 03:45:59 +00:00
assert telegram_mock.call_count == 1
def test_send_msg_telegram_disabled(mocker, default_conf, caplog) -> None:
telegram_mock = mocker.patch('freqtrade.rpc.telegram.Telegram.send_msg', MagicMock())
default_conf['telegram']['enabled'] = False
2018-02-13 03:45:59 +00:00
freqtradebot = get_patched_freqtradebot(mocker, default_conf)
2018-02-13 03:45:59 +00:00
rpc_manager = RPCManager(freqtradebot)
rpc_manager.send_msg({
'type': RPCMessageType.STATUS_NOTIFICATION,
'status': 'test'
})
2018-02-13 03:45:59 +00:00
2019-08-11 18:17:22 +00:00
assert log_has("Sending rpc message: {'type': status, 'status': 'test'}", caplog)
2018-02-13 03:45:59 +00:00
assert telegram_mock.call_count == 0
def test_send_msg_telegram_enabled(mocker, default_conf, caplog) -> None:
telegram_mock = mocker.patch('freqtrade.rpc.telegram.Telegram.send_msg', MagicMock())
mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock())
freqtradebot = get_patched_freqtradebot(mocker, default_conf)
2018-02-13 03:45:59 +00:00
rpc_manager = RPCManager(freqtradebot)
rpc_manager.send_msg({
'type': RPCMessageType.STATUS_NOTIFICATION,
'status': 'test'
})
2018-02-13 03:45:59 +00:00
2019-08-11 18:17:22 +00:00
assert log_has("Sending rpc message: {'type': status, 'status': 'test'}", caplog)
2018-02-13 03:45:59 +00:00
assert telegram_mock.call_count == 1
2018-07-12 18:17:45 +00:00
def test_init_webhook_disabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
default_conf['telegram']['enabled'] = False
default_conf['webhook'] = {'enabled': False}
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
2018-07-12 18:17:45 +00:00
2019-08-11 18:17:22 +00:00
assert not log_has('Enabling rpc.webhook ...', caplog)
2018-07-12 18:17:45 +00:00
assert rpc_manager.registered_modules == []
def test_init_webhook_enabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
default_conf['telegram']['enabled'] = False
2018-07-14 11:29:50 +00:00
default_conf['webhook'] = {'enabled': True, 'url': "https://DEADBEEF.com"}
2018-07-12 18:17:45 +00:00
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
2019-08-11 18:17:22 +00:00
assert log_has('Enabling rpc.webhook ...', caplog)
2018-07-29 14:09:44 +00:00
assert len(rpc_manager.registered_modules) == 1
2018-07-12 18:17:45 +00:00
assert 'webhook' in [mod.name for mod in rpc_manager.registered_modules]
def test_send_msg_webhook_CustomMessagetype(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
default_conf['telegram']['enabled'] = False
default_conf['webhook'] = {'enabled': True, 'url': "https://DEADBEEF.com"}
mocker.patch('freqtrade.rpc.webhook.Webhook.send_msg',
MagicMock(side_effect=NotImplementedError))
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
assert 'webhook' in [mod.name for mod in rpc_manager.registered_modules]
rpc_manager.send_msg({'type': RPCMessageType.STARTUP_NOTIFICATION,
'status': 'TestMessage'})
assert log_has(
2020-09-19 18:04:12 +00:00
"Message type 'startup' not implemented by handler webhook.",
caplog)
def test_startupmessages_telegram_enabled(mocker, default_conf, caplog) -> None:
telegram_mock = mocker.patch('freqtrade.rpc.telegram.Telegram.send_msg', MagicMock())
mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock())
freqtradebot = get_patched_freqtradebot(mocker, default_conf)
rpc_manager = RPCManager(freqtradebot)
2018-12-03 19:38:15 +00:00
rpc_manager.startup_messages(default_conf, freqtradebot.pairlists)
assert telegram_mock.call_count == 3
assert "*Exchange:* `bittrex`" in telegram_mock.call_args_list[1][0][0]['status']
telegram_mock.reset_mock()
default_conf['dry_run'] = True
2018-12-03 19:31:25 +00:00
default_conf['whitelist'] = {'method': 'VolumePairList',
'config': {'number_assets': 20}
}
2018-12-03 19:38:15 +00:00
rpc_manager.startup_messages(default_conf, freqtradebot.pairlists)
assert telegram_mock.call_count == 3
assert "Dry run is enabled." in telegram_mock.call_args_list[0][0][0]['status']
2019-05-11 11:40:30 +00:00
def test_init_apiserver_disabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
run_mock = MagicMock()
mocker.patch('freqtrade.rpc.api_server.ApiServer.run', run_mock)
default_conf['telegram']['enabled'] = False
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
2019-08-11 18:17:22 +00:00
assert not log_has('Enabling rpc.api_server', caplog)
2019-05-11 11:40:30 +00:00
assert rpc_manager.registered_modules == []
assert run_mock.call_count == 0
def test_init_apiserver_enabled(mocker, default_conf, caplog) -> None:
caplog.set_level(logging.DEBUG)
run_mock = MagicMock()
mocker.patch('freqtrade.rpc.api_server.ApiServer.run', run_mock)
default_conf["telegram"]["enabled"] = False
default_conf["api_server"] = {"enabled": True,
"listen_ip_address": "127.0.0.1",
2019-11-23 14:20:53 +00:00
"listen_port": 8080,
"username": "TestUser",
"password": "TestPass",
}
2019-05-11 11:40:30 +00:00
rpc_manager = RPCManager(get_patched_freqtradebot(mocker, default_conf))
2019-11-12 12:14:43 +00:00
# Sleep to allow the thread to start
time.sleep(0.5)
2019-08-11 18:17:22 +00:00
assert log_has('Enabling rpc.api_server', caplog)
2019-05-11 11:40:30 +00:00
assert len(rpc_manager.registered_modules) == 1
assert 'apiserver' in [mod.name for mod in rpc_manager.registered_modules]
assert run_mock.call_count == 1