65 lines
2.8 KiB
Python
65 lines
2.8 KiB
Python
import os
|
|
import sys
|
|
|
|
root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
|
|
sys.path.append(root)
|
|
|
|
# ----------------------------------------------------------------------------
|
|
|
|
# PLEASE DO NOT EDIT THIS FILE, IT IS GENERATED AND WILL BE OVERWRITTEN:
|
|
# https://github.com/ccxt/ccxt/blob/master/CONTRIBUTING.md#how-to-contribute-code
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import asyncio
|
|
from ccxt.base.errors import ArgumentsRequired # noqa E402
|
|
from ccxt.test.exchange.base import test_ticker # noqa E402
|
|
from ccxt.test.exchange.base import test_shared_methods # noqa E402
|
|
|
|
async def test_watch_tickers(exchange, skipped_properties, symbol):
|
|
without_symbol = test_watch_tickers_helper(exchange, skipped_properties, None)
|
|
with_symbol = test_watch_tickers_helper(exchange, skipped_properties, [symbol])
|
|
await asyncio.gather(*[with_symbol, without_symbol])
|
|
|
|
|
|
async def test_watch_tickers_helper(exchange, skipped_properties, arg_symbols, arg_params={}):
|
|
method = 'watchTickers'
|
|
now = exchange.milliseconds()
|
|
ends = now + 15000
|
|
while now < ends:
|
|
response = None
|
|
success = True
|
|
should_return = False
|
|
try:
|
|
response = await exchange.watch_tickers(arg_symbols, arg_params)
|
|
except Exception as e:
|
|
# for some exchanges, specifically watchTickers method not subscribe
|
|
# to "all tickers" itself, and it requires symbols to be set
|
|
# so, in such case, if it's arguments-required exception, we don't
|
|
# mark tests as failed, but just skip them
|
|
if (isinstance(e, ArgumentsRequired)) and (arg_symbols is None or len(arg_symbols) == 0):
|
|
# todo: provide random symbols to try
|
|
# return;
|
|
# return false;
|
|
should_return = True
|
|
elif not test_shared_methods.is_temporary_failure(e):
|
|
raise e
|
|
now = exchange.milliseconds()
|
|
# continue;
|
|
success = False
|
|
if should_return:
|
|
return False
|
|
if success:
|
|
assert isinstance(response, dict), exchange.id + ' ' + method + ' ' + exchange.json(arg_symbols) + ' must return an object. ' + exchange.json(response)
|
|
values = list(response.values())
|
|
checked_symbol = None
|
|
if arg_symbols is not None and len(arg_symbols) == 1:
|
|
checked_symbol = arg_symbols[0]
|
|
test_shared_methods.assert_non_emtpy_array(exchange, skipped_properties, method, values, checked_symbol)
|
|
for i in range(0, len(values)):
|
|
ticker = values[i]
|
|
test_ticker(exchange, skipped_properties, method, ticker, checked_symbol)
|
|
now = exchange.milliseconds()
|
|
return True
|