piker/tests/conftest.py

108 lines
3.1 KiB
Python

import os
import pytest
import tractor
import trio
from piker import log, config
from piker.brokers import questrade
def pytest_addoption(parser):
parser.addoption("--ll", action="store", dest='loglevel',
default=None, help="logging level to set when testing")
parser.addoption("--confdir", default=None,
help="Use a practice API account")
@pytest.fixture(scope='session', autouse=True)
def loglevel(request):
orig = tractor.log._default_loglevel
level = tractor.log._default_loglevel = request.config.option.loglevel
log.get_console_log(level)
yield level
tractor.log._default_loglevel = orig
@pytest.fixture(scope='session')
def test_config():
dirname = os.path.dirname
dirpath = os.path.abspath(
os.path.join(
dirname(os.path.realpath(__file__)),
'data'
)
)
return dirpath
@pytest.fixture(scope='session', autouse=True)
def confdir(request, test_config):
"""If the `--confdir` flag is not passed use the
broker config file found in that dir.
"""
confdir = request.config.option.confdir
if confdir is not None:
config._override_config_dir(confdir)
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)
@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']