mohammadian7/freqtrade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Further telegram async tests

Browse Source
This commit is contained in:
Matthias
2023-04-15 15:54:25 +02:00
parent 4177afdf8b
commit 69f61ef767
1 changed files with 16 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/rpc/test_rpc_telegram.py
+16 -15
View File
@@ -137,7 +137,7 @@ def test_cleanup(default_conf, mocker, ) -> None:
telegram, _, _ = get_telegram_testobject(mocker, default_conf, mock=False) telegram, _, _ = get_telegram_testobject(mocker, default_conf, mock=False)
telegram.cleanup() telegram.cleanup()
assert telegram._updater.stop.call_count == 1 assert telegram._app.stop.call_count == 1
async def test_authorized_only(default_conf, mocker, caplog, update) -> None: async def test_authorized_only(default_conf, mocker, caplog, update) -> None:
@@ -2327,32 +2327,33 @@ def test__sell_emoji(default_conf, mocker, msg, expected):
async def test_telegram__send_msg(default_conf, mocker, caplog) -> None: async def test_telegram__send_msg(default_conf, mocker, caplog) -> None:
mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock()) mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock())
bot = MagicMock() bot = MagicMock()
bot.send_message = AsyncMock()
bot.edit_message_text = AsyncMock()
telegram, _, _ = get_telegram_testobject(mocker, default_conf, mock=False) telegram, _, _ = get_telegram_testobject(mocker, default_conf, mock=False)
telegram._updater = MagicMock() telegram._app = MagicMock()
telegram._updater.bot = bot telegram._app.bot = bot
telegram._config['telegram']['enabled'] = True
await telegram._send_msg('test') await telegram._send_msg('test')
assert len(bot.method_calls) == 1 assert len(bot.method_calls) == 1
# Test update # Test update
query = MagicMock() query = MagicMock()
await telegram._send_msg('test', callback_path="DeadBeef", query=query, reload_able=True) await telegram._send_msg('test', callback_path="DeadBeef", query=query, reload_able=True)
edit_message_text = telegram._updater.bot.edit_message_text edit_message_text = telegram._app.bot.edit_message_text
assert edit_message_text.call_count == 1 assert edit_message_text.call_count == 1
assert "Updated: " in edit_message_text.call_args_list[0][1]['text'] assert "Updated: " in edit_message_text.call_args_list[0][1]['text']
telegram._updater.bot.edit_message_text = MagicMock(side_effect=BadRequest("not modified")) telegram._app.bot.edit_message_text = AsyncMock(side_effect=BadRequest("not modified"))
await telegram._send_msg('test', callback_path="DeadBeef", query=query) await telegram._send_msg('test', callback_path="DeadBeef", query=query)
assert telegram._updater.bot.edit_message_text.call_count == 1 assert telegram._app.bot.edit_message_text.call_count == 1
assert not log_has_re(r"TelegramError: .*", caplog) assert not log_has_re(r"TelegramError: .*", caplog)
telegram._updater.bot.edit_message_text = MagicMock(side_effect=BadRequest("")) telegram._app.bot.edit_message_text = AsyncMock(side_effect=BadRequest(""))
await telegram._send_msg('test2', callback_path="DeadBeef", query=query) await telegram._send_msg('test2', callback_path="DeadBeef", query=query)
assert telegram._updater.bot.edit_message_text.call_count == 1 assert telegram._app.bot.edit_message_text.call_count == 1
assert log_has_re(r"TelegramError: .*", caplog) assert log_has_re(r"TelegramError: .*", caplog)
telegram._updater.bot.edit_message_text = MagicMock(side_effect=TelegramError("DeadBEEF")) telegram._app.bot.edit_message_text = AsyncMock(side_effect=TelegramError("DeadBEEF"))
await telegram._send_msg('test3', callback_path="DeadBeef", query=query) await telegram._send_msg('test3', callback_path="DeadBeef", query=query)
assert log_has_re(r"TelegramError: DeadBEEF! Giving up.*", caplog) assert log_has_re(r"TelegramError: DeadBEEF! Giving up.*", caplog)
@@ -2363,8 +2364,8 @@ async def test__send_msg_network_error(default_conf, mocker, caplog) -> None:
bot = MagicMock() bot = MagicMock()
bot.send_message = MagicMock(side_effect=NetworkError('Oh snap')) bot.send_message = MagicMock(side_effect=NetworkError('Oh snap'))
telegram, _, _ = get_telegram_testobject(mocker, default_conf, mock=False) telegram, _, _ = get_telegram_testobject(mocker, default_conf, mock=False)
telegram._updater = MagicMock() telegram._app = MagicMock()
telegram._updater.bot = bot telegram._app.bot = bot
telegram._config['telegram']['enabled'] = True telegram._config['telegram']['enabled'] = True
await telegram._send_msg('test') await telegram._send_msg('test')
@@ -2377,7 +2378,7 @@ async def test__send_msg_network_error(default_conf, mocker, caplog) -> None:
async def test__send_msg_keyboard(default_conf, mocker, caplog) -> None: async def test__send_msg_keyboard(default_conf, mocker, caplog) -> None:
mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock()) mocker.patch('freqtrade.rpc.telegram.Telegram._init', MagicMock())
bot = MagicMock() bot = MagicMock()
bot.send_message = MagicMock() bot.send_message = AsyncMock()
freqtradebot = get_patched_freqtradebot(mocker, default_conf) freqtradebot = get_patched_freqtradebot(mocker, default_conf)
rpc = RPC(freqtradebot) rpc = RPC(freqtradebot)
@@ -2393,8 +2394,8 @@ async def test__send_msg_keyboard(default_conf, mocker, caplog) -> None:
def init_telegram(freqtradebot): def init_telegram(freqtradebot):
telegram = Telegram(rpc, default_conf) telegram = Telegram(rpc, default_conf)
telegram._updater = MagicMock() telegram._app = MagicMock()
telegram._updater.bot = bot telegram._app.bot = bot
return telegram return telegram
# no keyboard in config -> default keyboard # no keyboard in config -> default keyboard