Merge pull request #7303 from wizrds/feat/externalsignals

Producer/consumer mode

tests/conftest.py

@@ -58,6 +58,11 @@ def log_has(line, logs):
     return any(line == message for message in logs.messages)
 
 
+def log_has_when(line, logs, when):
+    """Check if line is found in caplog's messages during a specified stage"""
+    return any(line == message.message for message in logs.get_records(when))
+
+
 def log_has_re(line, logs):
     """Check if line matches some caplog's message."""
     return any(re.match(line, message) for message in logs.messages)

tests/data/test_dataprovider.py

@@ -144,6 +144,77 @@ def test_available_pairs(mocker, default_conf, ohlcv_history):
     assert dp.available_pairs == [("XRP/BTC", timeframe), ("UNITTEST/BTC", timeframe), ]
 
 
+def test_producer_pairs(mocker, default_conf, ohlcv_history):
+    dataprovider = DataProvider(default_conf, None)
+
+    producer = "default"
+    whitelist = ["XRP/BTC", "ETH/BTC"]
+    assert len(dataprovider.get_producer_pairs(producer)) == 0
+
+    dataprovider._set_producer_pairs(whitelist, producer)
+    assert len(dataprovider.get_producer_pairs(producer)) == 2
+
+    new_whitelist = ["BTC/USDT"]
+    dataprovider._set_producer_pairs(new_whitelist, producer)
+    assert dataprovider.get_producer_pairs(producer) == new_whitelist
+
+    assert dataprovider.get_producer_pairs("bad") == []
+
+
+def test_get_producer_df(mocker, default_conf, ohlcv_history):
+    dataprovider = DataProvider(default_conf, None)
+
+    pair = 'BTC/USDT'
+    timeframe = default_conf['timeframe']
+    candle_type = CandleType.SPOT
+
+    empty_la = datetime.fromtimestamp(0, tz=timezone.utc)
+    now = datetime.now(timezone.utc)
+
+    # no data has been added, any request should return an empty dataframe
+    dataframe, la = dataprovider.get_producer_df(pair, timeframe, candle_type)
+    assert dataframe.empty
+    assert la == empty_la
+
+    # the data is added, should return that added dataframe
+    dataprovider._add_external_df(pair, ohlcv_history, now, timeframe, candle_type)
+    dataframe, la = dataprovider.get_producer_df(pair, timeframe, candle_type)
+    assert len(dataframe) > 0
+    assert la > empty_la
+
+    # no data on this producer, should return empty dataframe
+    dataframe, la = dataprovider.get_producer_df(pair, producer_name='bad')
+    assert dataframe.empty
+    assert la == empty_la
+
+    # non existent timeframe, empty dataframe
+    datframe, la = dataprovider.get_producer_df(pair, timeframe='1h')
+    assert dataframe.empty
+    assert la == empty_la
+
+
+def test_emit_df(mocker, default_conf, ohlcv_history):
+    mocker.patch('freqtrade.rpc.rpc_manager.RPCManager.__init__', MagicMock())
+    rpc_mock = mocker.patch('freqtrade.rpc.rpc_manager.RPCManager', MagicMock())
+    send_mock = mocker.patch('freqtrade.rpc.rpc_manager.RPCManager.send_msg', MagicMock())
+
+    dataprovider = DataProvider(default_conf, exchange=None, rpc=rpc_mock)
+    dataprovider_no_rpc = DataProvider(default_conf, exchange=None)
+
+    pair = "BTC/USDT"
+
+    # No emit yet
+    assert send_mock.call_count == 0
+
+    # Rpc is added, we call emit, should call send_msg
+    dataprovider._emit_df(pair, ohlcv_history)
+    assert send_mock.call_count == 1
+
+    # No rpc added, emit called, should not call send_msg
+    dataprovider_no_rpc._emit_df(pair, ohlcv_history)
+    assert send_mock.call_count == 1
+
+
 def test_refresh(mocker, default_conf, ohlcv_history):
     refresh_mock = MagicMock()
     mocker.patch("freqtrade.exchange.Exchange.refresh_latest_ohlcv", refresh_mock)

tests/rpc/test_rpc_apiserver.py

@@ -3,6 +3,8 @@ Unit test file for rpc/api_server.py
 """
 
 import json
+import logging
+import time
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from unittest.mock import ANY, MagicMock, PropertyMock
@@ -10,7 +12,7 @@ from unittest.mock import ANY, MagicMock, PropertyMock
 import pandas as pd
 import pytest
 import uvicorn
-from fastapi import FastAPI
+from fastapi import FastAPI, WebSocketDisconnect
 from fastapi.exceptions import HTTPException
 from fastapi.testclient import TestClient
 from requests.auth import _basic_auth_str
@@ -31,6 +33,7 @@ from tests.conftest import (CURRENT_TEST_STRATEGY, create_mock_trades, get_mock_
 BASE_URI = "/api/v1"
 _TEST_USER = "FreqTrader"
 _TEST_PASS = "SuperSecurePassword1!"
+_TEST_WS_TOKEN = "secret_Ws_t0ken"
 
 
 @pytest.fixture
@@ -44,17 +47,21 @@ def botclient(default_conf, mocker):
                                         "CORS_origins": ['http://example.com'],
                                         "username": _TEST_USER,
                                         "password": _TEST_PASS,
+                                        "ws_token": _TEST_WS_TOKEN
                                         }})
 
     ftbot = get_patched_freqtradebot(mocker, default_conf)
     rpc = RPC(ftbot)
     mocker.patch('freqtrade.rpc.api_server.ApiServer.start_api', MagicMock())
+    apiserver = None
     try:
         apiserver = ApiServer(default_conf)
         apiserver.add_rpc_handler(rpc)
         yield ftbot, TestClient(apiserver.app)
         # Cleanup ... ?
     finally:
+        if apiserver:
+            apiserver.cleanup()
         ApiServer.shutdown()
 
 
@@ -154,6 +161,25 @@ def test_api_auth():
         get_user_from_token(b'not_a_token', 'secret1234')
 
 
+def test_api_ws_auth(botclient):
+    ftbot, client = botclient
+    def url(token): return f"/api/v1/message/ws?token={token}"
+
+    bad_token = "bad-ws_token"
+    with pytest.raises(WebSocketDisconnect):
+        with client.websocket_connect(url(bad_token)) as websocket:
+            websocket.receive()
+
+    good_token = _TEST_WS_TOKEN
+    with client.websocket_connect(url(good_token)) as websocket:
+        pass
+
+    jwt_secret = ftbot.config['api_server'].get('jwt_secret_key', 'super-secret')
+    jwt_token = create_token({'identity': {'u': 'Freqtrade'}}, jwt_secret)
+    with client.websocket_connect(url(jwt_token)) as websocket:
+        pass
+
+
 def test_api_unauthorized(botclient):
     ftbot, client = botclient
     rc = client.get(f"{BASE_URI}/ping")
@@ -261,6 +287,7 @@ def test_api__init__(default_conf, mocker):
     with pytest.raises(OperationalException, match="RPC Handler already attached."):
         apiserver.add_rpc_handler(RPC(get_patched_freqtradebot(mocker, default_conf)))
 
+    apiserver.cleanup()
     ApiServer.shutdown()
 
 
@@ -388,6 +415,7 @@ def test_api_run(default_conf, mocker, caplog):
                  MagicMock(side_effect=Exception))
     apiserver.start_api()
     assert log_has("Api server failed to start.", caplog)
+    apiserver.cleanup()
     ApiServer.shutdown()
 
 
@@ -410,6 +438,7 @@ def test_api_cleanup(default_conf, mocker, caplog):
     apiserver.cleanup()
     assert apiserver._server.cleanup.call_count == 1
     assert log_has("Stopping API Server", caplog)
+    assert log_has("Stopping API Server background tasks", caplog)
     ApiServer.shutdown()
 
 
@@ -1663,3 +1692,93 @@ def test_health(botclient):
     ret = rc.json()
     assert ret['last_process_ts'] == 0
     assert ret['last_process'] == '1970-01-01T00:00:00+00:00'
+
+
+def test_api_ws_subscribe(botclient, mocker):
+    ftbot, client = botclient
+    ws_url = f"/api/v1/message/ws?token={_TEST_WS_TOKEN}"
+
+    sub_mock = mocker.patch('freqtrade.rpc.api_server.ws.WebSocketChannel.set_subscriptions')
+
+    with client.websocket_connect(ws_url) as ws:
+        ws.send_json({'type': 'subscribe', 'data': ['whitelist']})
+
+    # Check call count is now 1 as we sent a valid subscribe request
+    assert sub_mock.call_count == 1
+
+    with client.websocket_connect(ws_url) as ws:
+        ws.send_json({'type': 'subscribe', 'data': 'whitelist'})
+
+    # Call count hasn't changed as the subscribe request was invalid
+    assert sub_mock.call_count == 1
+
+
+def test_api_ws_requests(botclient, mocker, caplog):
+    caplog.set_level(logging.DEBUG)
+
+    ftbot, client = botclient
+    ws_url = f"/api/v1/message/ws?token={_TEST_WS_TOKEN}"
+
+    # Test whitelist request
+    with client.websocket_connect(ws_url) as ws:
+        ws.send_json({"type": "whitelist", "data": None})
+        response = ws.receive_json()
+
+    assert log_has_re(r"Request of type whitelist from.+", caplog)
+    assert response['type'] == "whitelist"
+
+    # Test analyzed_df request
+    with client.websocket_connect(ws_url) as ws:
+        ws.send_json({"type": "analyzed_df", "data": {}})
+        response = ws.receive_json()
+
+    assert log_has_re(r"Request of type analyzed_df from.+", caplog)
+    assert response['type'] == "analyzed_df"
+
+    caplog.clear()
+    # Test analyzed_df request with data
+    with client.websocket_connect(ws_url) as ws:
+        ws.send_json({"type": "analyzed_df", "data": {"limit": 100}})
+        response = ws.receive_json()
+
+    assert log_has_re(r"Request of type analyzed_df from.+", caplog)
+    assert response['type'] == "analyzed_df"
+
+
+def test_api_ws_send_msg(default_conf, mocker, caplog):
+    try:
+        caplog.set_level(logging.DEBUG)
+
+        default_conf.update({"api_server": {"enabled": True,
+                                            "listen_ip_address": "127.0.0.1",
+                                            "listen_port": 8080,
+                                            "CORS_origins": ['http://example.com'],
+                                            "username": _TEST_USER,
+                                            "password": _TEST_PASS,
+                                            "ws_token": _TEST_WS_TOKEN
+                                            }})
+        mocker.patch('freqtrade.rpc.telegram.Updater')
+        mocker.patch('freqtrade.rpc.api_server.ApiServer.start_api')
+        apiserver = ApiServer(default_conf)
+        apiserver.add_rpc_handler(RPC(get_patched_freqtradebot(mocker, default_conf)))
+        apiserver.start_message_queue()
+        # Give the queue thread time to start
+        time.sleep(0.2)
+
+        # Test message_queue coro receives the message
+        test_message = {"type": "status", "data": "test"}
+        apiserver.send_msg(test_message)
+        time.sleep(0.1)  # Not sure how else to wait for the coro to receive the data
+        assert log_has("Found message of type: status", caplog)
+
+        # Test if exception logged when error occurs in sending
+        mocker.patch('freqtrade.rpc.api_server.ws.channel.ChannelManager.broadcast',
+                     side_effect=Exception)
+
+        apiserver.send_msg(test_message)
+        time.sleep(0.1)  # Not sure how else to wait for the coro to receive the data
+        assert log_has_re(r"Exception happened in background task.*", caplog)
+
+    finally:
+        apiserver.cleanup()
+        ApiServer.shutdown()

tests/rpc/test_rpc_emc.py

@@ -0,0 +1,460 @@
+"""
+Unit test file for rpc/external_message_consumer.py
+"""
+import asyncio
+import functools
+import logging
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+import pytest
+import websockets
+
+from freqtrade.data.dataprovider import DataProvider
+from freqtrade.exceptions import OperationalException
+from freqtrade.rpc.external_message_consumer import ExternalMessageConsumer
+from tests.conftest import log_has, log_has_re, log_has_when
+
+
+_TEST_WS_TOKEN = "secret_Ws_t0ken"
+_TEST_WS_HOST = "127.0.0.1"
+_TEST_WS_PORT = 9989
+
+
+@pytest.fixture
+def patched_emc(default_conf, mocker):
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": "null",
+                    "port": 9891,
+                    "ws_token": _TEST_WS_TOKEN
+                }
+            ]
+        }
+    })
+    dataprovider = DataProvider(default_conf, None, None, None)
+    emc = ExternalMessageConsumer(default_conf, dataprovider)
+
+    try:
+        yield emc
+    finally:
+        emc.shutdown()
+
+
+def test_emc_start(patched_emc, caplog):
+    # Test if the message was printed
+    assert log_has_when("Starting ExternalMessageConsumer", caplog, "setup")
+    # Test if the thread and loop objects were created
+    assert patched_emc._thread and patched_emc._loop
+
+    # Test we call start again nothing happens
+    prev_thread = patched_emc._thread
+    patched_emc.start()
+    assert prev_thread == patched_emc._thread
+
+
+def test_emc_shutdown(patched_emc, caplog):
+    patched_emc.shutdown()
+
+    assert log_has("Stopping ExternalMessageConsumer", caplog)
+    # Test the loop has stopped
+    assert patched_emc._loop is None
+    # Test if the thread has stopped
+    assert patched_emc._thread is None
+
+    caplog.clear()
+    patched_emc.shutdown()
+
+    # Test func didn't run again as it was called once already
+    assert not log_has("Stopping ExternalMessageConsumer", caplog)
+
+
+def test_emc_init(patched_emc, default_conf):
+    # Test the settings were set correctly
+    assert patched_emc.initial_candle_limit <= 1500
+    assert patched_emc.wait_timeout > 0
+    assert patched_emc.sleep_time > 0
+
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": []
+        }
+    })
+    dataprovider = DataProvider(default_conf, None, None, None)
+    with pytest.raises(OperationalException,
+                       match="You must specify at least 1 Producer to connect to."):
+        ExternalMessageConsumer(default_conf, dataprovider)
+
+
+# Parametrize this?
+def test_emc_handle_producer_message(patched_emc, caplog, ohlcv_history):
+    test_producer = {"name": "test", "url": "ws://test", "ws_token": "test"}
+    producer_name = test_producer['name']
+
+    caplog.set_level(logging.DEBUG)
+
+    # Test handle whitelist message
+    whitelist_message = {"type": "whitelist", "data": ["BTC/USDT"]}
+    patched_emc.handle_producer_message(test_producer, whitelist_message)
+
+    assert log_has(f"Received message of type `whitelist` from `{producer_name}`", caplog)
+    assert log_has(
+        f"Consumed message from `{producer_name}` of type `RPCMessageType.WHITELIST`", caplog)
+
+    # Test handle analyzed_df message
+    df_message = {
+        "type": "analyzed_df",
+        "data": {
+            "key": ("BTC/USDT", "5m", "spot"),
+            "df": ohlcv_history,
+            "la": datetime.now(timezone.utc)
+        }
+    }
+    patched_emc.handle_producer_message(test_producer, df_message)
+
+    assert log_has(f"Received message of type `analyzed_df` from `{producer_name}`", caplog)
+    assert log_has(
+        f"Consumed message from `{producer_name}` of type `RPCMessageType.ANALYZED_DF`", caplog)
+
+    # Test unhandled message
+    unhandled_message = {"type": "status", "data": "RUNNING"}
+    patched_emc.handle_producer_message(test_producer, unhandled_message)
+
+    assert log_has_re(r"Received unhandled message\: .*", caplog)
+
+    # Test malformed messages
+    caplog.clear()
+    malformed_message = {"type": "whitelist", "data": {"pair": "BTC/USDT"}}
+    patched_emc.handle_producer_message(test_producer, malformed_message)
+
+    assert log_has_re(r"Invalid message .+", caplog)
+
+    malformed_message = {
+        "type": "analyzed_df",
+        "data": {
+            "key": "BTC/USDT",
+            "df": ohlcv_history,
+            "la": datetime.now(timezone.utc)
+        }
+    }
+    patched_emc.handle_producer_message(test_producer, malformed_message)
+
+    assert log_has(f"Received message of type `analyzed_df` from `{producer_name}`", caplog)
+    assert log_has_re(r"Invalid message .+", caplog)
+
+    caplog.clear()
+    malformed_message = {"some": "stuff"}
+    patched_emc.handle_producer_message(test_producer, malformed_message)
+
+    assert log_has_re(r"Invalid message .+", caplog)
+
+    caplog.clear()
+    malformed_message = {"type": "whitelist", "data": None}
+    patched_emc.handle_producer_message(test_producer, malformed_message)
+
+    assert log_has_re(r"Empty message .+", caplog)
+
+
+async def test_emc_create_connection_success(default_conf, caplog, mocker):
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": _TEST_WS_HOST,
+                    "port": _TEST_WS_PORT,
+                    "ws_token": _TEST_WS_TOKEN
+                }
+            ],
+            "wait_timeout": 60,
+            "ping_timeout": 60,
+            "sleep_timeout": 60
+        }
+    })
+
+    mocker.patch('freqtrade.rpc.external_message_consumer.ExternalMessageConsumer.start',
+                 MagicMock())
+    dp = DataProvider(default_conf, None, None, None)
+    emc = ExternalMessageConsumer(default_conf, dp)
+
+    test_producer = default_conf['external_message_consumer']['producers'][0]
+    lock = asyncio.Lock()
+
+    emc._running = True
+
+    async def eat(websocket):
+        emc._running = False
+
+    try:
+        async with websockets.serve(eat, _TEST_WS_HOST, _TEST_WS_PORT):
+            await emc._create_connection(test_producer, lock)
+
+        assert log_has_re(r"Producer connection success.+", caplog)
+    finally:
+        emc.shutdown()
+
+
+# async def test_emc_create_connection_invalid(default_conf, caplog, mocker):
+#     default_conf.update({
+#         "external_message_consumer": {
+#             "enabled": True,
+#             "producers": [
+#                 {
+#                     "name": "default",
+#                     "host": _TEST_WS_HOST,
+#                     "port": _TEST_WS_PORT,
+#                     "ws_token": _TEST_WS_TOKEN
+#                 }
+#             ],
+#             "wait_timeout": 60,
+#             "ping_timeout": 60,
+#             "sleep_timeout": 60
+#         }
+#     })
+#
+#     mocker.patch('freqtrade.rpc.external_message_consumer.ExternalMessageConsumer.start',
+#                  MagicMock())
+#
+#     test_producer = default_conf['external_message_consumer']['producers'][0]
+#     lock = asyncio.Lock()
+#
+#     dp = DataProvider(default_conf, None, None, None)
+#     emc = ExternalMessageConsumer(default_conf, dp)
+#
+#     try:
+#         # Test invalid URL
+#         test_producer['url'] = "tcp://null:8080/api/v1/message/ws"
+#         emc._running = True
+#         await emc._create_connection(test_producer, lock)
+#         emc._running = False
+#
+#         assert log_has_re(r".+is an invalid WebSocket URL.+", caplog)
+#     finally:
+#         emc.shutdown()
+
+
+async def test_emc_create_connection_error(default_conf, caplog, mocker):
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": _TEST_WS_HOST,
+                    "port": _TEST_WS_PORT,
+                    "ws_token": _TEST_WS_TOKEN
+                }
+            ],
+            "wait_timeout": 60,
+            "ping_timeout": 60,
+            "sleep_timeout": 60
+        }
+    })
+
+    # Test unexpected error
+    mocker.patch('websockets.connect', side_effect=RuntimeError)
+
+    dp = DataProvider(default_conf, None, None, None)
+    emc = ExternalMessageConsumer(default_conf, dp)
+
+    try:
+        await asyncio.sleep(0.01)
+        assert log_has("Unexpected error has occurred:", caplog)
+    finally:
+        emc.shutdown()
+
+
+async def test_emc_receive_messages_valid(default_conf, caplog, mocker):
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": _TEST_WS_HOST,
+                    "port": _TEST_WS_PORT,
+                    "ws_token": _TEST_WS_TOKEN
+                }
+            ],
+            "wait_timeout": 1,
+            "ping_timeout": 60,
+            "sleep_time": 60
+        }
+    })
+
+    mocker.patch('freqtrade.rpc.external_message_consumer.ExternalMessageConsumer.start',
+                 MagicMock())
+
+    lock = asyncio.Lock()
+    test_producer = default_conf['external_message_consumer']['producers'][0]
+
+    dp = DataProvider(default_conf, None, None, None)
+    emc = ExternalMessageConsumer(default_conf, dp)
+
+    loop = asyncio.get_event_loop()
+    def change_running(emc): emc._running = not emc._running
+
+    class TestChannel:
+        async def recv(self, *args, **kwargs):
+            return {"type": "whitelist", "data": ["BTC/USDT"]}
+
+        async def ping(self, *args, **kwargs):
+            return asyncio.Future()
+
+    try:
+        change_running(emc)
+        loop.call_soon(functools.partial(change_running, emc=emc))
+        await emc._receive_messages(TestChannel(), test_producer, lock)
+
+        assert log_has_re(r"Received message of type `whitelist`.+", caplog)
+    finally:
+        emc.shutdown()
+
+
+async def test_emc_receive_messages_invalid(default_conf, caplog, mocker):
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": _TEST_WS_HOST,
+                    "port": _TEST_WS_PORT,
+                    "ws_token": _TEST_WS_TOKEN
+                }
+            ],
+            "wait_timeout": 1,
+            "ping_timeout": 60,
+            "sleep_time": 60
+        }
+    })
+
+    mocker.patch('freqtrade.rpc.external_message_consumer.ExternalMessageConsumer.start',
+                 MagicMock())
+
+    lock = asyncio.Lock()
+    test_producer = default_conf['external_message_consumer']['producers'][0]
+
+    dp = DataProvider(default_conf, None, None, None)
+    emc = ExternalMessageConsumer(default_conf, dp)
+
+    loop = asyncio.get_event_loop()
+    def change_running(emc): emc._running = not emc._running
+
+    class TestChannel:
+        async def recv(self, *args, **kwargs):
+            return {"type": ["BTC/USDT"]}
+
+        async def ping(self, *args, **kwargs):
+            return asyncio.Future()
+
+    try:
+        change_running(emc)
+        loop.call_soon(functools.partial(change_running, emc=emc))
+        await emc._receive_messages(TestChannel(), test_producer, lock)
+
+        assert log_has_re(r"Invalid message from.+", caplog)
+    finally:
+        emc.shutdown()
+
+
+async def test_emc_receive_messages_timeout(default_conf, caplog, mocker):
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": _TEST_WS_HOST,
+                    "port": _TEST_WS_PORT,
+                    "ws_token": _TEST_WS_TOKEN
+                }
+            ],
+            "wait_timeout": 1,
+            "ping_timeout": 1,
+            "sleep_time": 1
+        }
+    })
+
+    mocker.patch('freqtrade.rpc.external_message_consumer.ExternalMessageConsumer.start',
+                 MagicMock())
+
+    lock = asyncio.Lock()
+    test_producer = default_conf['external_message_consumer']['producers'][0]
+
+    dp = DataProvider(default_conf, None, None, None)
+    emc = ExternalMessageConsumer(default_conf, dp)
+
+    loop = asyncio.get_event_loop()
+    def change_running(emc): emc._running = not emc._running
+
+    class TestChannel:
+        async def recv(self, *args, **kwargs):
+            await asyncio.sleep(10)
+
+        async def ping(self, *args, **kwargs):
+            return asyncio.Future()
+
+    try:
+        change_running(emc)
+        loop.call_soon(functools.partial(change_running, emc=emc))
+        await emc._receive_messages(TestChannel(), test_producer, lock)
+
+        assert log_has_re(r"Ping error.+", caplog)
+    finally:
+        emc.shutdown()
+
+
+async def test_emc_receive_messages_handle_error(default_conf, caplog, mocker):
+    default_conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": _TEST_WS_HOST,
+                    "port": _TEST_WS_PORT,
+                    "ws_token": _TEST_WS_TOKEN
+                }
+            ],
+            "wait_timeout": 1,
+            "ping_timeout": 1,
+            "sleep_time": 1
+        }
+    })
+
+    mocker.patch('freqtrade.rpc.external_message_consumer.ExternalMessageConsumer.start',
+                 MagicMock())
+
+    lock = asyncio.Lock()
+    test_producer = default_conf['external_message_consumer']['producers'][0]
+
+    dp = DataProvider(default_conf, None, None, None)
+    emc = ExternalMessageConsumer(default_conf, dp)
+
+    emc.handle_producer_message = MagicMock(side_effect=Exception)
+
+    loop = asyncio.get_event_loop()
+    def change_running(emc): emc._running = not emc._running
+
+    class TestChannel:
+        async def recv(self, *args, **kwargs):
+            return {"type": "whitelist", "data": ["BTC/USDT"]}
+
+        async def ping(self, *args, **kwargs):
+            return asyncio.Future()
+
+    try:
+        change_running(emc)
+        loop.call_soon(functools.partial(change_running, emc=emc))
+        await emc._receive_messages(TestChannel(), test_producer, lock)
+
+        assert log_has_re(r"Error handling producer message.+", caplog)
+    finally:
+        emc.shutdown()

tests/test_freqtradebot.py

@@ -1319,9 +1319,9 @@ def test_create_stoploss_order_invalid_order(
     assert create_order_mock.call_args[1]['amount'] == trade.amount
 
     # Rpc is sending first buy, then sell
-    assert rpc_mock.call_count == 2
-    assert rpc_mock.call_args_list[1][0][0]['sell_reason'] == ExitType.EMERGENCY_EXIT.value
-    assert rpc_mock.call_args_list[1][0][0]['order_type'] == 'market'
+    assert rpc_mock.call_count == 3
+    assert rpc_mock.call_args_list[2][0][0]['sell_reason'] == ExitType.EMERGENCY_EXIT.value
+    assert rpc_mock.call_args_list[2][0][0]['order_type'] == 'market'
 
 
 @pytest.mark.parametrize("is_short", [False, True])
@@ -2439,7 +2439,7 @@ def test_manage_open_orders_entry_usercustom(
     # Trade should be closed since the function returns true
     freqtrade.manage_open_orders()
     assert cancel_order_wr_mock.call_count == 1
-    assert rpc_mock.call_count == 1
+    assert rpc_mock.call_count == 2
     trades = Trade.query.filter(Trade.open_order_id.is_(open_trade.open_order_id)).all()
     nb_trades = len(trades)
     assert nb_trades == 0
@@ -2478,7 +2478,7 @@ def test_manage_open_orders_entry(
     # check it does cancel buy orders over the time limit
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 1
-    assert rpc_mock.call_count == 1
+    assert rpc_mock.call_count == 2
     trades = Trade.query.filter(Trade.open_order_id.is_(open_trade.open_order_id)).all()
     nb_trades = len(trades)
     assert nb_trades == 0
@@ -2608,7 +2608,7 @@ def test_check_handle_cancelled_buy(
     # check it does cancel buy orders over the time limit
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 0
-    assert rpc_mock.call_count == 1
+    assert rpc_mock.call_count == 2
     trades = Trade.query.filter(Trade.open_order_id.is_(open_trade.open_order_id)).all()
     assert len(trades) == 0
     assert log_has_re(
@@ -2639,7 +2639,7 @@ def test_manage_open_orders_buy_exception(
     # check it does cancel buy orders over the time limit
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 0
-    assert rpc_mock.call_count == 0
+    assert rpc_mock.call_count == 1
     trades = Trade.query.filter(Trade.open_order_id.is_(open_trade.open_order_id)).all()
     nb_trades = len(trades)
     assert nb_trades == 1
@@ -2686,7 +2686,7 @@ def test_manage_open_orders_exit_usercustom(
     # Return false - No impact
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 0
-    assert rpc_mock.call_count == 0
+    assert rpc_mock.call_count == 1
     assert open_trade_usdt.is_open is False
     assert freqtrade.strategy.check_exit_timeout.call_count == 1
     assert freqtrade.strategy.check_entry_timeout.call_count == 0
@@ -2696,7 +2696,7 @@ def test_manage_open_orders_exit_usercustom(
     # Return Error - No impact
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 0
-    assert rpc_mock.call_count == 0
+    assert rpc_mock.call_count == 1
     assert open_trade_usdt.is_open is False
     assert freqtrade.strategy.check_exit_timeout.call_count == 1
     assert freqtrade.strategy.check_entry_timeout.call_count == 0
@@ -2706,7 +2706,7 @@ def test_manage_open_orders_exit_usercustom(
     freqtrade.strategy.check_entry_timeout = MagicMock(return_value=True)
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 1
-    assert rpc_mock.call_count == 1
+    assert rpc_mock.call_count == 2
     assert open_trade_usdt.is_open is True
     assert freqtrade.strategy.check_exit_timeout.call_count == 1
     assert freqtrade.strategy.check_entry_timeout.call_count == 0
@@ -2766,7 +2766,7 @@ def test_manage_open_orders_exit(
     # check it does cancel sell orders over the time limit
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 1
-    assert rpc_mock.call_count == 1
+    assert rpc_mock.call_count == 2
     assert open_trade_usdt.is_open is True
     # Custom user sell-timeout is never called
     assert freqtrade.strategy.check_exit_timeout.call_count == 0
@@ -2805,7 +2805,7 @@ def test_check_handle_cancelled_exit(
     # check it does cancel sell orders over the time limit
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 0
-    assert rpc_mock.call_count == 1
+    assert rpc_mock.call_count == 2
     assert open_trade_usdt.is_open is True
     exit_name = 'Buy' if is_short else 'Sell'
     assert log_has_re(f"{exit_name} order cancelled on exchange for Trade.*", caplog)
@@ -2843,7 +2843,7 @@ def test_manage_open_orders_partial(
     # note this is for a partially-complete buy order
     freqtrade.manage_open_orders()
     assert cancel_order_mock.call_count == 1
-    assert rpc_mock.call_count == 2
+    assert rpc_mock.call_count == 3
     trades = Trade.query.filter(Trade.open_order_id.is_(open_trade.open_order_id)).all()
     assert len(trades) == 1
     assert trades[0].amount == 23.0
@@ -2890,7 +2890,7 @@ def test_manage_open_orders_partial_fee(
     assert log_has_re(r"Applying fee on amount for Trade.*", caplog)
 
     assert cancel_order_mock.call_count == 1
-    assert rpc_mock.call_count == 2
+    assert rpc_mock.call_count == 3
     trades = Trade.query.filter(Trade.open_order_id.is_(open_trade.open_order_id)).all()
     assert len(trades) == 1
     # Verify that trade has been updated
@@ -2940,7 +2940,7 @@ def test_manage_open_orders_partial_except(
     assert log_has_re(r"Could not update trade amount: .*", caplog)
 
     assert cancel_order_mock.call_count == 1
-    assert rpc_mock.call_count == 2
+    assert rpc_mock.call_count == 3
     trades = Trade.query.filter(Trade.open_order_id.is_(open_trade.open_order_id)).all()
     assert len(trades) == 1
     # Verify that trade has been updated
@@ -3155,7 +3155,7 @@ def test_handle_cancel_exit_limit(mocker, default_conf_usdt, fee) -> None:
     reason = CANCEL_REASON['TIMEOUT']
     assert freqtrade.handle_cancel_exit(trade, order, reason)
     assert cancel_order_mock.call_count == 1
-    assert send_msg_mock.call_count == 1
+    assert send_msg_mock.call_count == 2
     assert trade.close_rate is None
     assert trade.exit_reason is None
 
@@ -3592,7 +3592,7 @@ def test_execute_trade_exit_with_stoploss_on_exchange(
     trade.is_short = is_short
     assert trade
     assert cancel_order.call_count == 1
-    assert rpc_mock.call_count == 3
+    assert rpc_mock.call_count == 4
 
 
 @pytest.mark.parametrize("is_short", [False, True])
@@ -3662,11 +3662,11 @@ def test_may_execute_trade_exit_after_stoploss_on_exchange_hit(
     assert trade.stoploss_order_id is None
     assert trade.is_open is False
     assert trade.exit_reason == ExitType.STOPLOSS_ON_EXCHANGE.value
-    assert rpc_mock.call_count == 3
-    assert rpc_mock.call_args_list[0][0][0]['type'] == RPCMessageType.ENTRY
-    assert rpc_mock.call_args_list[0][0][0]['amount'] > 20
-    assert rpc_mock.call_args_list[1][0][0]['type'] == RPCMessageType.ENTRY_FILL
-    assert rpc_mock.call_args_list[2][0][0]['type'] == RPCMessageType.EXIT_FILL
+    assert rpc_mock.call_count == 4
+    assert rpc_mock.call_args_list[1][0][0]['type'] == RPCMessageType.ENTRY
+    assert rpc_mock.call_args_list[1][0][0]['amount'] > 20
+    assert rpc_mock.call_args_list[2][0][0]['type'] == RPCMessageType.ENTRY_FILL
+    assert rpc_mock.call_args_list[3][0][0]['type'] == RPCMessageType.EXIT_FILL
 
 
 @pytest.mark.parametrize(

tests/test_misc.py

@@ -7,10 +7,11 @@ from unittest.mock import MagicMock
 
 import pytest
 
-from freqtrade.misc import (decimals_per_coin, deep_merge_dicts, file_dump_json, file_load_json,
-                            format_ms_time, pair_to_filename, parse_db_uri_for_logging, plural,
-                            render_template, render_template_with_fallback, round_coin_value,
-                            safe_value_fallback, safe_value_fallback2, shorten_date)
+from freqtrade.misc import (dataframe_to_json, decimals_per_coin, deep_merge_dicts, file_dump_json,
+                            file_load_json, format_ms_time, json_to_dataframe, pair_to_filename,
+                            parse_db_uri_for_logging, plural, render_template,
+                            render_template_with_fallback, round_coin_value, safe_value_fallback,
+                            safe_value_fallback2, shorten_date)
 
 
 def test_decimals_per_coin():
@@ -219,3 +220,14 @@ def test_deep_merge_dicts():
 
     res2['first']['rows']['test'] = 'asdf'
     assert deep_merge_dicts(a, deepcopy(b), allow_null_overrides=False) == res2
+
+
+def test_dataframe_json(ohlcv_history):
+    from pandas.testing import assert_frame_equal
+    json = dataframe_to_json(ohlcv_history)
+    dataframe = json_to_dataframe(json)
+
+    assert list(ohlcv_history.columns) == list(dataframe.columns)
+    assert len(ohlcv_history) == len(dataframe)
+
+    assert_frame_equal(ohlcv_history, dataframe)