CTYPTO/freqtrade
1
0
Fork 0
mirror of https://github.com/freqtrade/freqtrade.git synced 2025-11-29 08:33:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

test: improve exchange_ws tests

Browse Source
This commit is contained in:
Matthias
2025-01-05 13:00:47 +01:00
parent 98e49ab187
commit 6b96d4d145
1 changed files with 34 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
tests/exchange/test_exchange_ws.py
View File

@@ -42,11 +42,19 @@ def patch_eventloop_threading(exchange):
pass pass
async def test_exchangews_ohlcv(mocker): async def test_exchangews_ohlcv(mocker, time_machine):
config = MagicMock() config = MagicMock()
ccxt_object = MagicMock() ccxt_object = MagicMock()
ccxt_object.watch_ohlcv = AsyncMock()
async def sleeper(*args, **kwargs):
# pass
await asyncio.sleep(1)
return MagicMock()
ccxt_object.watch_ohlcv = AsyncMock(side_effect=sleeper)
ccxt_object.close = AsyncMock() ccxt_object.close = AsyncMock()
time_machine.move_to("2024-11-01 01:00:00 +00:00")
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock()) mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
exchange_ws = ExchangeWS(config, ccxt_object) exchange_ws = ExchangeWS(config, ccxt_object)
@@ -56,14 +64,33 @@ async def test_exchangews_ohlcv(mocker):
assert exchange_ws._klines_scheduled == set() assert exchange_ws._klines_scheduled == set()
exchange_ws.schedule_ohlcv("ETH/BTC", "1m", CandleType.SPOT) exchange_ws.schedule_ohlcv("ETH/BTC", "1m", CandleType.SPOT)
exchange_ws.schedule_ohlcv("XRP/BTC", "1m", CandleType.SPOT)
await asyncio.sleep(0.5) await asyncio.sleep(0.5)
assert exchange_ws._klines_watching == {("ETH/BTC", "1m", CandleType.SPOT)} assert exchange_ws._klines_watching == {
assert exchange_ws._klines_scheduled == {("ETH/BTC", "1m", CandleType.SPOT)} ("ETH/BTC", "1m", CandleType.SPOT),
("XRP/BTC", "1m", CandleType.SPOT),
}
assert exchange_ws._klines_scheduled == {
("ETH/BTC", "1m", CandleType.SPOT),
("XRP/BTC", "1m", CandleType.SPOT),
}
await asyncio.sleep(0.1) await asyncio.sleep(0.1)
assert ccxt_object.watch_ohlcv.call_count == 1 assert ccxt_object.watch_ohlcv.call_count == 2
except Exception as e: ccxt_object.watch_ohlcv.reset_mock()
print(e)
time_machine.shift(timedelta(minutes=5))
await asyncio.sleep(0.1)
exchange_ws.schedule_ohlcv("ETH/BTC", "1m", CandleType.SPOT)
# XRP/BTC should be cleaned up.
assert exchange_ws._klines_watching == {
("ETH/BTC", "1m", CandleType.SPOT),
}
assert exchange_ws._klines_scheduled == {
("ETH/BTC", "1m", CandleType.SPOT),
("XRP/BTC", "1m", CandleType.SPOT),
}
finally: finally:
# Cleanup # Cleanup
exchange_ws.cleanup() exchange_ws.cleanup()