You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
85 lines
2.6 KiB
85 lines
2.6 KiB
import pytest |
|
|
|
import cryptocom.exchange as cro |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_get_symbols(exchange: cro.Exchange): |
|
data = await exchange.get_symbols() |
|
symbols = [d['symbol'] for d in data] |
|
assert sorted(symbols) == sorted(s.value for s in cro.Symbol) |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_get_tickers(exchange: cro.Exchange): |
|
tickers = await exchange.get_tickers() |
|
assert sorted(tickers) == sorted(s.value for s in cro.Symbol) |
|
keys = sorted(['amount', 'close', 'high', 'open', 'low', 'rose', 'vol']) |
|
for data in tickers.values(): |
|
assert keys == sorted(data) |
|
ticker = await exchange.get_ticker(cro.Symbol.BTCUSDT) |
|
assert keys == sorted(ticker) |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_get_candles(exchange: cro.Exchange): |
|
candles = [candle async for candle in exchange.get_candles( |
|
cro.Symbol.CROUSDT, cro.Period.D1)] |
|
assert len(candles) > 10 |
|
assert isinstance(candles[-1], cro.Candle) |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_get_trades(exchange: cro.Exchange): |
|
trades = await exchange.get_trades(cro.Symbol.CROUSDT) |
|
keys = sorted(['amount', 'price', 'ctime', 'id', 'type']) |
|
for trade in trades: |
|
assert sorted(trade) == keys |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_get_prices(exchange: cro.Exchange): |
|
tickers = await exchange.get_prices() |
|
assert sorted(tickers) == sorted(s.value for s in cro.Symbol) |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_get_orderbook(exchange: cro.Exchange): |
|
data = await exchange.get_orderbook(cro.Symbol.CROUSDT, cro.Depth.HIGH) |
|
asks = data['asks'] |
|
bids = data['bids'] |
|
assert asks and bids |
|
assert len(asks[0]) == 2 |
|
assert len(bids[0]) == 2 |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_listen_candles(exchange: cro.Exchange): |
|
candles = [] |
|
async for candle in exchange.listen_candles( |
|
cro.Symbol.CROUSDT, cro.Period.MIN1): |
|
candles.append(candle) |
|
break |
|
assert isinstance(candles[-1], cro.Candle) |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_listen_trades(exchange: cro.Exchange): |
|
all_trades = [] |
|
async for trades in exchange.listen_trades(cro.Symbol.CROUSDT): |
|
all_trades.extend(trades) |
|
break |
|
keys = sorted(['amount', 'price', 'ds', 'id', 'side', 'ts', 'vol']) |
|
assert sorted(all_trades[0]) == keys |
|
|
|
|
|
@pytest.mark.asyncio |
|
async def test_listen_orderbook(exchange: cro.Exchange): |
|
async for orders in exchange.listen_order_book( |
|
cro.Symbol.CROUSDT, cro.Depth.HIGH): |
|
asks = orders['asks'] |
|
bids = orders['bids'] |
|
assert asks and bids |
|
assert len(asks[0]) == 2 |
|
assert len(bids[0]) == 2 |
|
break
|
|
|