<commit_before>'''py.test fixtures for Pyramid.

http://pyramid.readthedocs.org/en/latest/narr/testing.html
'''
import pytest
from pytest import fixture

app_settings = {
    'multiauth.policies': 'authtkt remoteuser edwkey',
    'multiauth.groupfinder': 'encoded.authorization.groupfinder',
    'multiauth.policy.authtkt.use': 'pyramid.authentication.AuthTktAuthenticationPolicy',
    'multiauth.policy.authtkt.hashalg': 'sha512',
    'multiauth.policy.authtkt.secret': 'GLIDING LIKE A WHALE',
    'multiauth.policy.remoteuser.use': 'encoded.authentication.NamespacedAuthenticationPolicy',
    'multiauth.policy.remoteuser.namespace': 'remoteuser',
    'multiauth.policy.remoteuser.base': 'pyramid.authentication.RemoteUserAuthenticationPolicy',
    'multiauth.policy.accesskey.use': 'encoded.authentication.NamespacedAuthenticationPolicy',
    'multiauth.policy.accesskey.namespace': 'accesskey',
    'multiauth.policy.accesskey.base': 'encoded.authentication.BasicAuthAuthenticationPolicy',
    'multiauth.policy.accesskey.check': 'encoded.authentication.basic_auth_check',
    'multiauth.policy.edwkey.use': 'encoded.authentication.NamespacedAuthenticationPolicy',
    'multiauth.policy.edwkey.namespace': 'edwkey',
    'multiauth.policy.edwkey.base': 'encoded.authentication.BasicAuthAuthenticationPolicy',
    'multiauth.policy.edwkey.check': 'encoded.edw_key.basic_auth_check',
    'persona.audiences': 'http://localhost:6543',
    'persona.siteName': 'ENCODE DCC Submission',
    'allow.view': 'Everyone',
    'allow.list': 'Everyone',
    'allow.traverse': 'Everyone',
    'allow.ALL_PERMISSIONS': 'group:admin',
    'allow.edw_key_create': 'edwkey:edw',
    'allow.edw_key_update': 'edwkey:edw',
    'load_test_only': True,
    'load_sample_data': False,
    'testing': True,
}


def pytest_configure():
    import logging
    logging.basicConfig()
    logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
    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()


@fixture
def dummy_request():
    from pyramid.testing import DummyRequest
    return DummyRequest()


@fixture(scope='session')
def app(request, check_constraints, zsa_savepoints):
    '''WSGI application level functional testing.
    '''
    from encoded import main
    return main({}, **app_settings)


@pytest.datafixture
def workbook(app):
    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
    insertbook = resource_filename('encoded', 'tests/data/test_encode3_interface_inserts.xlsx')
    updatebook = resource_filename('encoded', 'tests/data/test_encode3_interface_updates.xlsx')
    docsdir = [resource_filename('encoded', 'tests/data/documents/')]
    load_test_only = app_settings.get('load_test_only', False)
    assert load_test_only

    load_all(testapp, insertbook, docsdir, test=load_test_only)
    load_all(testapp, updatebook, docsdir, test=load_test_only)


@fixture
def htmltestapp(request, app, external_tx, zsa_savepoints):
    from webtest import TestApp
    return TestApp(app)


@fixture
def testapp(request, app, external_tx, zsa_savepoints):
    '''TestApp with JSON accept header.
    '''
    from webtest import TestApp
    environ = {
        'HTTP_ACCEPT': 'application/json',
        'REMOTE_USER': 'TEST',
    }
    return TestApp(app, environ)


@fixture
def anontestapp(request, app, external_tx, zsa_savepoints):
    '''TestApp with JSON accept header.
    '''
    from webtest import TestApp
    environ = {
        'HTTP_ACCEPT': 'application/json',
    }
    return TestApp(app, environ)


@fixture(scope='session')
def _server(request, app, zsa_savepoints):
    from webtest.http import StopableWSGIServer
    import threading
    from webtest.http import get_free_port

    def create(cls, application, **kwargs):
        """Start a server to serve ``application``. Return a server
        instance."""
        host, port = get_free_port()
        kwargs['port'] = port
        if 'host' not in kwargs:
            kwargs['host'] = host
        if 'expose_tracebacks' not in kwargs:
            kwargs['expose_tracebacks'] = True
        server = cls(application, **kwargs)
        server.runner = threading.Thread(target=server.run)
        server.runner.daemon = True
        server.runner.start()
        return server

    server = create(
        StopableWSGIServer,
        app,
        threads=1,
        channel_timeout=60,
        cleanup_interval=10,
        expose_tracebacks=True,
    )
    assert server.wait()

    @request.addfinalizer
    def shutdown():
        server.shutdown()

    return server


@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<commit_msg>Calculate the persona assertion as this will be needed when bdd tests are added.<commit_after>'''py.test fixtures for Pyramid.

http://pyramid.readthedocs.org/en/latest/narr/testing.html
'''
import pytest
from pytest import fixture

_app_settings = {
    'multiauth.policies': 'authtkt remoteuser edwkey',
    'multiauth.groupfinder': 'encoded.authorization.groupfinder',
    'multiauth.policy.authtkt.use': 'pyramid.authentication.AuthTktAuthenticationPolicy',
    'multiauth.policy.authtkt.hashalg': 'sha512',
    'multiauth.policy.authtkt.secret': 'GLIDING LIKE A WHALE',
    'multiauth.policy.remoteuser.use': 'encoded.authentication.NamespacedAuthenticationPolicy',
    'multiauth.policy.remoteuser.namespace': 'remoteuser',
    'multiauth.policy.remoteuser.base': 'pyramid.authentication.RemoteUserAuthenticationPolicy',
    'multiauth.policy.accesskey.use': 'encoded.authentication.NamespacedAuthenticationPolicy',
    'multiauth.policy.accesskey.namespace': 'accesskey',
    'multiauth.policy.accesskey.base': 'encoded.authentication.BasicAuthAuthenticationPolicy',
    'multiauth.policy.accesskey.check': 'encoded.authentication.basic_auth_check',
    'multiauth.policy.edwkey.use': 'encoded.authentication.NamespacedAuthenticationPolicy',
    'multiauth.policy.edwkey.namespace': 'edwkey',
    'multiauth.policy.edwkey.base': 'encoded.authentication.BasicAuthAuthenticationPolicy',
    'multiauth.policy.edwkey.check': 'encoded.edw_key.basic_auth_check',
    'persona.audiences': 'http://localhost:6543',
    'persona.siteName': 'ENCODE DCC Submission',
    'allow.view': 'Everyone',
    'allow.list': 'Everyone',
    'allow.traverse': 'Everyone',
    'allow.ALL_PERMISSIONS': 'group:admin',
    'allow.edw_key_create': 'edwkey:edw',
    'allow.edw_key_update': 'edwkey:edw',
    'load_test_only': True,
    'load_sample_data': False,
    'testing': True,
}


@fixture(scope='session')
def app_settings(server_host_port):
    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.INFO)
    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()


@fixture
def dummy_request():
    from pyramid.testing import DummyRequest
    return DummyRequest()


@fixture(scope='session')
def app(request, check_constraints, zsa_savepoints, app_settings):
    '''WSGI application level functional testing.
    '''
    from encoded import main
    return main({}, **app_settings)


@pytest.datafixture
def workbook(app, app_settings):
    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
    insertbook = resource_filename('encoded', 'tests/data/test_encode3_interface_inserts.xlsx')
    updatebook = resource_filename('encoded', 'tests/data/test_encode3_interface_updates.xlsx')
    docsdir = [resource_filename('encoded', 'tests/data/documents/')]
    load_test_only = app_settings.get('load_test_only', False)
    assert load_test_only

    load_all(testapp, insertbook, docsdir, test=load_test_only)
    load_all(testapp, updatebook, docsdir, test=load_test_only)


@fixture
def htmltestapp(request, app, external_tx, zsa_savepoints):
    from webtest import TestApp
    return TestApp(app)


@fixture
def testapp(request, app, external_tx, zsa_savepoints):
    '''TestApp with JSON accept header.
    '''
    from webtest import TestApp
    environ = {
        'HTTP_ACCEPT': 'application/json',
        'REMOTE_USER': 'TEST',
    }
    return TestApp(app, environ)


@fixture
def anontestapp(request, app, external_tx, zsa_savepoints):
    '''TestApp with JSON accept header.
    '''
    from webtest import TestApp
    environ = {
        'HTTP_ACCEPT': 'application/json',
    }
    return TestApp(app, environ)


@fixture(scope='session')
def server_host_port():
    from webtest.http import get_free_port
    return get_free_port()


@fixture(scope='session')
def _server(request, app, zsa_savepoints, server_host_port):
    from webtest.http import StopableWSGIServer
    import threading
    from webtest.http import get_free_port

    def create(cls, application, **kwargs):
        """Start a server to serve ``application``. Return a server
        instance."""
        host, port = get_free_port()
        kwargs['port'] = port
        if 'host' not in kwargs:
            kwargs['host'] = host
        if 'expose_tracebacks' not in kwargs:
            kwargs['expose_tracebacks'] = True
        server = cls(application, **kwargs)
        server.runner = threading.Thread(target=server.run)
        server.runner.daemon = True
        server.runner.start()
        return server

    host, port = server_host_port

    server = create(
        StopableWSGIServer,
        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 server


@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
{'fraction_non_alphanumeric': 0.11771792444712159, 'fraction_numerical': 0.0032699423457533774, 'mean_word_length': 3.6674698795180722, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2204091', 'n_tokens_mistral': 3696, 'n_tokens_neox': 3590, 'n_words': 703}