fix: threadApiManager (#1449)

* fix theadedapimanager

* pr comments

* add test

* rename test

* add test_ticker_socket

* get loop as dependenc yinjection

---------

Co-authored-by: carlosmiei <[email protected]>

Author: pcriadoperez

binance/streams.py

@@ -1306,9 +1306,10 @@ class ThreadedWebsocketManager(ThreadedApiManager):
     def __init__(
         self, api_key: Optional[str] = None, api_secret: Optional[str] = None,
         requests_params: Optional[Dict[str, Any]] = None, tld: str = 'com',
-        testnet: bool = False, session_params: Optional[Dict[str, Any]] = None
+        testnet: bool = False, session_params: Optional[Dict[str, Any]] = None,
+        loop: Optional[asyncio.AbstractEventLoop] = None
     ):
-        super().__init__(api_key, api_secret, requests_params, tld, testnet, session_params)
+        super().__init__(api_key, api_secret, requests_params, tld, testnet, session_params, loop)
         self._bsm: Optional[BinanceSocketManager] = None
 
     async def _before_socket_listener_start(self):

binance/threaded_stream.py

@@ -11,13 +11,14 @@ class ThreadedApiManager(threading.Thread):
     def __init__(
         self, api_key: Optional[str] = None, api_secret: Optional[str] = None,
         requests_params: Optional[Dict[str, Any]] = None, tld: str = 'com',
-        testnet: bool = False, session_params: Optional[Dict[str, Any]] = None
+        testnet: bool = False, session_params: Optional[Dict[str, Any]] = None,
+        _loop: Optional[asyncio.AbstractEventLoop] = None
     ):
         """Initialise the BinanceSocketManager
 
         """
         super().__init__()
-        self._loop: asyncio.AbstractEventLoop = asyncio.get_event_loop() if asyncio.get_event_loop().is_running() else asyncio.new_event_loop()
+        self._loop: asyncio.AbstractEventLoop = get_loop() if _loop is None else _loop
         self._client: Optional[AsyncClient] = None
         self._running: bool = True
         self._socket_running: Dict[str, bool] = {}

tests/test_socket_manager.py

@@ -0,0 +1,23 @@
+from binance import BinanceSocketManager, AsyncClient
+import pytest
+
+
+def assert_message(msg):
+    assert msg['stream'] == '!ticker@arr'
+    assert len(msg['data']) > 0
+
+@pytest.mark.asyncio()
+async def test_ticker_socket():
+    client = await AsyncClient.create(testnet=True)
+    bm = BinanceSocketManager(client)
+
+    ts = bm.futures_ticker_socket()
+
+    async with ts as tscm:
+        try:
+            res = await tscm.recv()
+            assert_message(res)
+        except Exception as e:
+            print(f"An error occurred: {e}")
+
+    await client.close_connection()

tests/test_threaded_socket_manager.py

@@ -0,0 +1,34 @@
+
+
+from binance import ThreadedWebsocketManager
+
+
+received_ohlcv = False
+received_depth = False
+
+twm: ThreadedWebsocketManager
+
+def handle_socket_message(msg):
+    global received_ohlcv, received_depth
+    print(msg)
+    if 'e' in msg:
+        if msg['e'] == 'depthUpdate':
+            received_depth = True
+        if msg['e'] == 'kline':
+            received_ohlcv = True
+    if received_depth and received_ohlcv:
+        twm.stop()
+
+def test_threaded_socket_manager():
+    global twm
+    twm = ThreadedWebsocketManager(api_key="", api_secret="", testnet=True)
+
+    symbol = "BTCUSDT"
+
+    twm.start()
+
+    twm.start_kline_socket(callback=handle_socket_message, symbol=symbol)
+
+    twm.start_depth_socket(callback=handle_socket_message, symbol=symbol)
+
+    twm.join()