text: '''py.test fixtures for Pyramid.
  
  http://pyramid.readthedocs.org/en/latest/narr/testing.html
  '''
  import pkg_resources
  import pytest
  from pytest import fixture
  import random
  
  pytest_plugins = [
      'clincoded.tests.datafixtures',
      'clincoded.tests.layers',
  ]
  
  _app_settings = {
      'collection_datastore': 'database',
      'item_datastore': 'database',
      'multiauth.policies': 'auth0 session remoteuser accesskey',
      'multiauth.groupfinder': 'clincoded.authorization.groupfinder',
      'multiauth.policy.auth0.use': 'clincoded.authentication.NamespacedAuthenticationPolicy',
      'multiauth.policy.auth0.base': 'clincoded.auth0.Auth0AuthenticationPolicy',
      'multiauth.policy.auth0.namespace': 'auth0',
      'multiauth.policy.session.use': 'clincoded.authentication.NamespacedAuthenticationPolicy',
      'multiauth.policy.session.base': 'pyramid.authentication.SessionAuthenticationPolicy',
      'multiauth.policy.session.namespace': 'mailto',
      'multiauth.policy.remoteuser.use': 'clincoded.authentication.NamespacedAuthenticationPolicy',
      'multiauth.policy.remoteuser.namespace': 'remoteuser',
      'multiauth.policy.remoteuser.base': 'pyramid.authentication.RemoteUserAuthenticationPolicy',
      'multiauth.policy.accesskey.use': 'clincoded.authentication.NamespacedAuthenticationPolicy',
      'multiauth.policy.accesskey.namespace': 'accesskey',
      'multiauth.policy.accesskey.base': 'clincoded.authentication.BasicAuthAuthenticationPolicy',
      'multiauth.policy.accesskey.check': 'clincoded.authentication.basic_auth_check',
      'auth0.verifier': 'browserid.LocalVerifier',
      'auth0.siteName': 'ClinGen Submission',
      'load_test_only': True,
      'testing': True,
      'pyramid.debug_authorization': True,
      'postgresql.statement_timeout': 20,
      'ontology_path': pkg_resources.resource_filename('clincoded', '../../ontology.json'),
  }
  
  
  @pytest.mark.fixture_cost(10)
  @pytest.yield_fixture(scope='session')
  def engine_url(request):
      engine_url = request.session.config.option.engine_url
      if engine_url is not None:
          yield engine_url
          return
  
      # Ideally this would use a different database on the same postgres server
      from urllib.parse import quote
      from .postgresql_fixture import initdb, server_process
      tmpdir = request.config._tmpdirhandler.mktemp('postgresql-engine', numbered=True)
      tmpdir = str(tmpdir)
      initdb(tmpdir)
      process = server_process(tmpdir)
  
      yield 'postgresql://postgres@:5432/postgres?host=%s' % quote(tmpdir)
  
      if process.poll() is None:
          process.terminate()
          process.wait()
  
  
  @fixture(scope='session')
  def app_settings(request, server_host_port, connection):
      settings = _app_settings.copy()
      settings['persona.audiences'] = 'http://%s:%s' % server_host_port
      return settings
  
  
  def pytest_configure():
      import logging
      logging.basicConfig()
      logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING)
      logging.getLogger('selenium').setLevel(logging.DEBUG)
  
      class Shorten(logging.Filter):
          max_len = 500
  
          def filter(self, record):
              if record.msg == '%r':
                  record.msg = record.msg % record.args
                  record.args = ()
              if len(record.msg) > self.max_len:
                  record.msg = record.msg[:self.max_len] + '...'
              return True
  
      logging.getLogger('sqlalchemy.engine.base.Engine').addFilter(Shorten())
  
  
  @fixture
  def config(request):
      from pyramid.testing import setUp, tearDown
      request.addfinalizer(tearDown)
      return setUp()
  
  
  @pytest.yield_fixture
  def threadlocals(request, dummy_request, registry):
      from pyramid.threadlocal import manager
      manager.push({'request': dummy_request, 'registry': registry})
      yield dummy_request
      manager.pop()
  
  
  from pyramid.testing import DummyRequest
  
  
  class MyDummyRequest(DummyRequest):
      def remove_conditional_headers(self):
          pass
  
      def _get_registry(self):
          from pyramid.threadlocal import get_current_registry
          if self._registry is None:
              return get_current_registry()
          return self._registry
  
      def _set_registry(self, registry):
          self.__dict__['registry'] = registry
  
      def _del_registry(self):
          self._registry = None
  
      registry = property(_get_registry, _set_registry, _del_registry)
  
  
  @fixture
  def dummy_request(root, registry, app):
      from pyramid.request import apply_request_extensions
      request = app.request_factory.blank('/dummy')
      request.root = root
      request.registry = registry
      request._stats = {}
      request.invoke_subrequest = app.invoke_subrequest
      apply_request_extensions(request)
      return request
  
  
  @fixture(scope='session')
  def app(zsa_savepoints, check_constraints, app_settings):
      '''WSGI application level functional testing.
      '''
      from clincoded import main
      return main({}, **app_settings)
  
  
  @fixture
  def registry(app):
      return app.registry
  
  
  @fixture
  def elasticsearch(registry):
      from contentbase.elasticsearch import ELASTIC_SEARCH
      return registry[ELASTIC_SEARCH]
  
  
  @fixture
  def root(app):
      return app.root_factory(app)
  
  
  @pytest.mark.fixture_cost(500)
  @pytest.yield_fixture(scope='session')
  def workbook(connection, app, app_settings):
      tx = connection.begin_nested()
      try:
          from webtest import TestApp
          environ = {
              'HTTP_ACCEPT': 'application/json',
              'REMOTE_USER': 'TEST',
          }
          testapp = TestApp(app, environ)
  
          from ..loadxl import load_all
          from pkg_resources import resource_filename
          inserts = resource_filename('clincoded', 'tests/data/inserts/')
          docsdir = [resource_filename('clincoded', 'tests/data/documents/')]
          load_all(testapp, inserts, docsdir)
  
          yield
      finally:
          tx.rollback()
  
  
  @fixture
  def anonhtmltestapp(app, external_tx):
      from webtest import TestApp
      return TestApp(app)
  
  
  @fixture
  def htmltestapp(app, external_tx):
      from webtest import TestApp
      environ = {
          'REMOTE_USER': 'TEST',
      }
      return TestApp(app, environ)
  
  
  @fixture
  def testapp(app, external_tx):
      '''TestApp with JSON accept header.
      '''
      from webtest import TestApp
      environ = {
          'HTTP_ACCEPT': 'application/json',
          'REMOTE_USER': 'TEST',
      }
      return TestApp(app, environ)
  
  
  @fixture
  def anontestapp(app, external_tx):
      '''TestApp with JSON accept header.
      '''
      from webtest import TestApp
      environ = {
          'HTTP_ACCEPT': 'application/json',
      }
      return TestApp(app, environ)
  
  
  @fixture
  def authenticated_testapp(app, external_tx):
      '''TestApp with JSON accept header for non-admin user.
      '''
      from webtest import TestApp
      environ = {
          'HTTP_ACCEPT': 'application/json',
          'REMOTE_USER': 'TEST_AUTHENTICATED',
      }
      return TestApp(app, environ)
  
  
  @fixture
  def submitter_testapp(app, external_tx):
      '''TestApp with JSON accept header for non-admin user.
      '''
      from webtest import TestApp
      environ = {
          'HTTP_ACCEPT': 'application/json',
          'REMOTE_USER': 'TEST_SUBMITTER',
      }
      return TestApp(app, environ)
  
  
  @pytest.fixture
  def indexer_testapp(app, external_tx):
      from webtest import TestApp
      environ = {
          'HTTP_ACCEPT': 'application/json',
          'REMOTE_USER': 'INDEXER',
      }
      return TestApp(app, environ)
  
  
  @pytest.fixture
  def embed_testapp(app, external_tx):
      from webtest import TestApp
      environ = {
          'HTTP_ACCEPT': 'application/json',
          'REMOTE_USER': 'EMBED',
      }
      return TestApp(app, environ)
  
  
  @fixture(scope='session')
  def server_host_port():
      # we have specified a range of ports for login testing at Auth0 for clincoded (65525-65534)
      port = random.randrange(65525, 65535)
      # TODO: test if socket can be made
  
      # the port will only be printed if errors occur
      print("**** login port ***", port)
      return ("127.0.0.1", port)
  
  
  @fixture(scope='session')
  def authenticated_app(app):
      from http.cookies import SimpleCookie
  
      def wsgi_filter(environ, start_response):
          # set REMOTE_USER from cookie
          cookies = SimpleCookie()
          cookies.load(environ.get('HTTP_COOKIE', ''))
          if 'REMOTE_USER' in cookies:
              user = cookies['REMOTE_USER'].value
          else:
              user = 'TEST_AUTHENTICATED'
          environ['REMOTE_USER'] = user
          return app(environ, start_response)
      return wsgi_filter
  
  
  @pytest.mark.fixture_cost(100)
  @fixture(scope='session')
  def _server(request, authenticated_app, server_host_port):
      from webtest.http import StopableWSGIServer
      host, port = server_host_port
  
      server = StopableWSGIServer.create(
          authenticated_app,
          host=host,
          port=port,
          threads=1,
          channel_timeout=60,
          cleanup_interval=10,
          expose_tracebacks=True,
      )
      assert server.wait()
  
      @request.addfinalizer
      def shutdown():
          server.shutdown()
  
      return 'http://%s:%s' % server_host_port
  
  
  @fixture
  def server(_server, external_tx):
      return _server
  
  
  # http://docs.sqlalchemy.org/en/rel_0_8/orm/session.html#joining-a-session-into-an-external-transaction
  # By binding the SQLAlchemy Session to an external transaction multiple testapp
  # requests can be rolled back at the end of the test.
  
  @pytest.mark.fixture_lock('contentbase.storage.DBSession')
  @pytest.yield_fixture(scope='session')
  def connection(request, engine_url):
      from clincoded import configure_engine
      from contentbase.storage import Base, DBSession
      from sqlalchemy.orm.scoping import ScopedRegistry
  
      # ``server`` thread must be in same scope
      if type(DBSession.registry) is not ScopedRegistry:
          DBSession.registry = ScopedRegistry(DBSession.session_factory, lambda: 0)
  
      engine_settings = {
          'sqlalchemy.url': engine_url,
      }
  
      engine = configure_engine(engine_settings, test_setup=True)
      connection = engine.connect()
      tx = connection.begin()
      try:
          Base.metadata.create_all(bind=connection)
          session = DBSession(scope=None, bind=connection)
          DBSession.registry.set(session)
          yield connection
      finally:
          tx.rollback()
          connection.close()
          engine.dispose()
  
  
  @fixture
  def external_tx(request, connection):
      print('BEGIN external_tx')
      tx = connection.begin_nested()
      request.addfinalizer(tx.rollback)
      # # The database should be empty unless a data fixture was loaded
      # from contentbase.storage import Base
      # for table in Base.metadata.sorted_tables:
      #     assert connection.execute(table.count()).scalar() == 0
      return tx
  
  
  @fixture
  def transaction(request, external_tx, zsa_savepoints, check_constraints):
      import transaction
      transaction.begin()
      request.addfinalizer(transaction.abort)
      return transaction
  
  
  @fixture(scope='session')
  def zsa_savepoints(request, connection):
      """ Place a savepoint at the start of the zope transaction
  
      This means failed requests rollback to the db state when they began rather
      than that at the start of the test.
      """
      from transaction.interfaces import ISynchronizer
      from zope.interface import implementer
  
      @implementer(ISynchronizer)
      class Savepoints(object):
          def __init__(self, connection):
              self.connection = connection
              self.sp = None
              self.state = None
  
          def beforeCompletion(self, transaction):
              pass
  
          def afterCompletion(self, transaction):
              # txn be aborted a second time in manager.begin()
              if self.sp is None:
                  return
              if self.state == 'commit':
                  self.state = 'completion'
                  self.sp.commit()
              else:
                  self.state = 'abort'
                  self.sp.rollback()
              self.sp = None
              self.state = 'done'
  
          def newTransaction(self, transaction):
              self.state = 'new'
              self.sp = self.connection.begin_nested()
              self.state = 'begun'
              transaction.addBeforeCommitHook(self._registerCommit)
  
          def _registerCommit(self):
              self.state = 'commit'
  
      zsa_savepoints = Savepoints(connection)
  
      import transaction
      transaction.manager.registerSynch(zsa_savepoints)
  
      @request.addfinalizer
      def unregister():
          transaction.manager.unregisterSynch(zsa_savepoints)
  
      return zsa_savepoints
  
  
  @fixture
  def session(transaction):
      """ Returns a setup session
  
      Depends on transaction as storage relies on some interaction there.
      """
      from contentbase.storage import DBSession
      return DBSession()
  
  
  @fixture(scope='session')
  def check_constraints(request, connection):
      '''Check deffered constraints on zope transaction commit.
  
      Deferred foreign key constraints are only checked at the outer transaction
      boundary, not at a savepoint. With the Pyramid transaction bound to a
      subtransaction check them manually.
      '''
      from contentbase.storage import DBSession
      from transaction.interfaces import ISynchronizer
      from zope.interface import implementer
  
      @implementer(ISynchronizer)
      class CheckConstraints(object):
          def __init__(self, connection):
              self.connection = connection
              self.state = None
  
          def beforeCompletion(self, transaction):
              pass
  
          def afterCompletion(self, transaction):
              pass
  
          def newTransaction(self, transaction):
  
              @transaction.addBeforeCommitHook
              def set_constraints():
                  self.state = 'checking'
                  session = DBSession()
                  session.flush()
                  sp = self.connection.begin_nested()
                  try:
                      self.connection.execute('SET CONSTRAINTS ALL IMMEDIATE')
                  except:
                      sp.rollback()
                      raise
                  else:
                      self.connection.execute('SET CONSTRAINTS ALL DEFERRED')
                  finally:
                      sp.commit()
                      self.state = None
  
      check_constraints = CheckConstraints(connection)
  
      import transaction
      transaction.manager.registerSynch(check_constraints)
  
      @request.addfinalizer
      def unregister():
          transaction.manager.unregisterSynch(check_constraints)
  
      return check_constraints
  
  
  @fixture
  def execute_counter(request, connection, zsa_savepoints, check_constraints):
      """ Count calls to execute
      """
      from contextlib import contextmanager
      from sqlalchemy import event
  
      class Counter(object):
          def __init__(self):
              self.reset()
              self.conn = connection
  
          def reset(self):
              self.count = 0
  
          @contextmanager
          def expect(self, count):
              start = self.count
              yield
              difference = self.count - start
              assert difference == count
  
      counter = Counter()
  
      @event.listens_for(connection, 'after_cursor_execute')
      def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
          # Ignore the testing savepoints
          if zsa_savepoints.state != 'begun' or check_constraints.state == 'checking':
              return
          counter.count += 1
  
      @request.addfinalizer
      def remove():
          event.remove(connection, 'after_cursor_execute', after_cursor_execute)
  
      return counter
  
  
  @fixture
  def no_deps(request, connection):
      from contentbase.storage import DBSession
      from sqlalchemy import event
  
      session = DBSession()
  
      @event.listens_for(session, 'after_flush')
      def check_dependencies(session, flush_context):
          assert not flush_context.cycles
  
      @event.listens_for(connection, "before_execute", retval=True)
      def before_execute(conn, clauseelement, multiparams, params):
          return clauseelement, multiparams, params
  
      @request.addfinalizer
      def remove():
          event.remove(session, 'before_flush', check_dependencies)
  
  
  @pytest.mark.fixture_cost(10)
  @pytest.yield_fixture(scope='session')
  def postgresql_server(request):
      from urllib.parse import quote
      from .postgresql_fixture import initdb, server_process
      tmpdir = request.config._tmpdirhandler.mktemp('postgresql', numbered=True)
      tmpdir = str(tmpdir)
      initdb(tmpdir)
      process = server_process(tmpdir)
  
      yield 'postgresql://postgres@:5432/postgres?host=%s' % quote(tmpdir)
  
      if process.poll() is None:
          process.terminate()
          process.wait()
  
  
  @pytest.fixture(scope='session')
  def elasticsearch_host_port():
      from webtest.http import get_free_port
      return get_free_port()
  
  
  @pytest.mark.fixture_cost(10)
  @pytest.yield_fixture(scope='session')
  def elasticsearch_server(request, elasticsearch_host_port):
      from .elasticsearch_fixture import server_process
      host, port = elasticsearch_host_port
      tmpdir = request.config._tmpdirhandler.mktemp('elasticsearch', numbered=True)
      tmpdir = str(tmpdir)
      process = server_process(str(tmpdir), host=host, port=port)
  
      yield 'http://%s:%d' % (host, port)
  
      if process.poll() is None:
          process.terminate()
          process.wait()
  