176 lines
4.5 KiB
Python
176 lines
4.5 KiB
Python
from contextlib import asynccontextmanager as acm
|
|
import os
|
|
|
|
import pytest
|
|
import tractor
|
|
from piker import (
|
|
# log,
|
|
config,
|
|
)
|
|
from piker._daemon import (
|
|
Services,
|
|
)
|
|
|
|
|
|
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')
|
|
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)
|
|
|
|
|
|
_ci_env: bool = os.environ.get('CI', False)
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def ci_env() -> bool:
|
|
'''
|
|
Detect CI envoirment.
|
|
|
|
'''
|
|
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,
|
|
**kwargs,
|
|
|
|
) -> tuple[
|
|
str,
|
|
int,
|
|
tractor.Portal
|
|
]:
|
|
'''
|
|
Testing helper to startup the service tree and runtime on
|
|
a different port then the default to allow testing alongside
|
|
a running stack.
|
|
|
|
'''
|
|
import random
|
|
from piker._daemon import maybe_open_pikerd
|
|
|
|
if reg_addr is None:
|
|
port = random.randint(6e3, 7e3)
|
|
reg_addr = ('127.0.0.1', port)
|
|
|
|
# try:
|
|
async with (
|
|
maybe_open_pikerd(
|
|
registry_addr=reg_addr,
|
|
**kwargs,
|
|
) as service_manager,
|
|
):
|
|
# this proc/actor is the pikerd
|
|
assert service_manager is Services
|
|
|
|
async with tractor.wait_for_actor(
|
|
'pikerd',
|
|
arbiter_sockaddr=reg_addr,
|
|
) as portal:
|
|
raddr = portal.channel.raddr
|
|
assert raddr == reg_addr
|
|
yield (
|
|
raddr[0],
|
|
raddr[1],
|
|
portal,
|
|
service_manager,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def open_test_pikerd():
|
|
|
|
yield _open_test_pikerd
|
|
|
|
# TODO: teardown checks such as,
|
|
# - no leaked subprocs or shm buffers
|
|
# - all requested container service are torn down
|
|
# - certain ``tractor`` runtime state?
|