Add an initial `tractor` price streaming test

kivy_mainline_and_py3.8
Tyler Goodlet 2018-06-07 00:29:17 -04:00
parent 58f0182d8f
commit 5a997d67d4
1 changed files with 57 additions and 0 deletions

View File

@ -0,0 +1,57 @@
"""
Actor model API testing
"""
import pytest
from piker import tractor
@pytest.fixture
def us_symbols():
return ['TSLA', 'AAPL', 'CGC', 'CRON']
@pytest.mark.trio
async def test_rx_price_quotes_from_brokerd(us_symbols):
"""Verify we can spawn a daemon actor and retrieve streamed price data.
"""
async with tractor.open_nursery() as nursery:
# only one per host address, spawns an actor if None
async with tractor.find_actors('brokerd') as portals:
if not portals:
# no brokerd actor found
portal = await nursery.start_actor(
'brokerd',
['piker.brokers.core'],
statespace={
'brokers2tickersubs': {},
'clients': {},
'dtasks': set()
},
main=None,
)
# gotta expose in a broker agnostic way...
# retrieve initial symbol data
# sd = await portal.run(
# 'piker.brokers.core', 'symbol_data', symbols=us_symbols)
# assert list(sd.keys()) == us_symbols
gen = await portal.run(
'piker.brokers.core',
'_test_price_stream',
broker='robinhood',
symbols=us_symbols,
)
# it'd sure be nice to have an asyncitertools here...
async for quotes in gen:
assert quotes
for key in quotes:
assert key in us_symbols
break
# terminate far-end async-gen
# await gen.asend(None)
# break
await nursery.cancel()
# arbitter is cancelled here due to `find_actors()` internals