from contextlib import asynccontextmanager as acm import os import pytest import tractor from piker import ( # log, config, ) 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) async with ( maybe_open_pikerd( registry_addr=reg_addr, **kwargs, ), ): 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, ) @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?