Move quest fixtures to test mod, clean out old travis fixture

loglevel_to_testpikerd
Tyler Goodlet 2023-02-21 09:35:38 -05:00
parent e01220af14
commit cd3e9b1b2a
2 changed files with 21 additions and 66 deletions

View File

@ -45,51 +45,6 @@ def confdir(request, test_config):
return confdir
# @pytest.fixture(scope='session', autouse=True)
# def travis(confdir):
# is_travis = os.environ.get('TRAVIS', False)
# if is_travis:
# # this directory is cached, see .travis.yaml
# conf_file = config.get_broker_conf_path()
# refresh_token = os.environ['QT_REFRESH_TOKEN']
# def write_with_token(token):
# # XXX don't pass the dir path here since may be
# # written behind the scenes in the `confdir fixture`
# if not os.path.isfile(conf_file):
# open(conf_file, 'w').close()
# conf, path = config.load()
# conf.setdefault('questrade', {}).update(
# {'refresh_token': token,
# 'is_practice': 'True'}
# )
# config.write(conf, path)
# async def ensure_config():
# # try to refresh current token using cached brokers config
# # if it fails fail try using the refresh token provided by the
# # env var and if that fails stop the test run here.
# try:
# async with questrade.get_client(ask_user=False):
# pass
# except (
# FileNotFoundError, ValueError,
# questrade.BrokerError, questrade.QuestradeError,
# trio.MultiError,
# ):
# # 3 cases:
# # - config doesn't have a ``refresh_token`` k/v
# # - cache dir does not exist yet
# # - current token is expired; take it form env var
# write_with_token(refresh_token)
# async with questrade.get_client(ask_user=False):
# pass
# # XXX ``pytest_trio`` doesn't support scope or autouse
# trio.run(ensure_config)
_ci_env: bool = os.environ.get('CI', False)
@ -102,24 +57,10 @@ def ci_env() -> bool:
return _ci_env
@pytest.fixture
def us_symbols():
return ['TSLA', 'AAPL', 'CGC', 'CRON']
@pytest.fixture
def tmx_symbols():
return ['APHA.TO', 'WEED.TO', 'ACB.TO']
@pytest.fixture
def cse_symbols():
return ['TRUL.CN', 'CWEB.CN', 'SNN.CN']
@acm
async def _open_test_pikerd(
reg_addr: tuple[str, int] | None = None,
loglevel: str = 'warning',
**kwargs,
) -> tuple[
@ -140,7 +81,6 @@ async def _open_test_pikerd(
port = random.randint(6e3, 7e3)
reg_addr = ('127.0.0.1', port)
# try:
async with (
maybe_open_pikerd(
registry_addr=reg_addr,
@ -165,7 +105,7 @@ async def _open_test_pikerd(
@pytest.fixture
def open_test_pikerd():
def open_test_pikerd(request):
yield _open_test_pikerd

View File

@ -111,6 +111,21 @@ def match_packet(symbols, quotes, feed_type='stock'):
assert not quotes
@pytest.fixture
def us_symbols():
return ['TSLA', 'AAPL', 'CGC', 'CRON']
@pytest.fixture
def tmx_symbols():
return ['APHA.TO', 'WEED.TO', 'ACB.TO']
@pytest.fixture
def cse_symbols():
return ['TRUL.CN', 'CWEB.CN', 'SNN.CN']
# @tractor_test
async def test_concurrent_tokens_refresh(us_symbols, loglevel):
"""Verify that concurrent requests from mulitple tasks work alongside
@ -258,8 +273,8 @@ async def stream_option_chain(feed, symbols):
# latency arithmetic
loops = 8
period = 1/3. # 3 rps
timeout = float('inf') #loops / period
# period = 1/3. # 3 rps
timeout = float('inf') # loops / period
try:
# it'd sure be nice to have an asyncitertools here...
@ -307,8 +322,8 @@ async def stream_stocks(feed, symbols):
symbols, 'stock', rate=3, diff_cached=False)
# latency arithmetic
loops = 8
period = 1/3. # 3 rps
timeout = loops / period
# period = 1/3. # 3 rps
# timeout = loops / period
try:
# it'd sure be nice to have an asyncitertools here...