<commit_before># -*- coding: utf-8 -*-
u"""Support for unit tests

:copyright: Copyright (c) 2016 RadiaSoft LLC.  All Rights Reserved.
:license: http://www.apache.org/licenses/LICENSE-2.0.html
"""
from __future__ import absolute_import, division, print_function
import flask
import flask.testing
import json
import re
from pykern.pkcollections import PKDict


#: Default "app"
MYAPP = 'myapp'

#: import sirepo.server
server = None

#: app result from server.init
app = None

#: Matches javascript-redirect.html
_JAVASCRIPT_REDIRECT_RE = re.compile(r'window.location = "([^"]+)"')

#: set by conftest.py
CONFTEST_ALL_CODES = None


def flask_client(cfg=None, sim_types=None, job_run_mode=None):
    """Return FlaskClient with easy access methods.

    Creates a new run directory every test file so can assume
    sharing of state on the server within a file (module).

    Two methods of interest: `sr_post` and `sr_get`.

    Args:
        cfg (dict): extra configuration for reset_state_for_testing
        sim_types (str): value for SIREPO_FEATURE_CONFIG_SIM_TYPES [CONFTEST_ALL_CODES]

    Returns:
        FlaskClient: for local requests to Flask server
    """
    global server, app

    a = 'srunit_flask_client'
    if not cfg:
        cfg = PKDict()
    t = sim_types or CONFTEST_ALL_CODES
    if t:
        if isinstance(t, (tuple, list)):
            t = ':'.join(t)
        cfg['SIREPO_FEATURE_CONFIG_SIM_TYPES'] = t
    if not (server and hasattr(app, a)):
        from pykern import pkconfig

        # initialize pkdebug with correct values
        pkconfig.reset_state_for_testing(cfg)

        from pykern import pkunit
        with pkunit.save_chdir_work() as wd:
            from pykern import pkio
            cfg['SIREPO_SRDB_ROOT'] = str(pkio.mkdir_parent(wd.join('db')))
            pkconfig.reset_state_for_testing(cfg)
            from sirepo import server as s

            server = s
            app = server.init()
            app.config['TESTING'] = True
            app.test_client_class = _TestClient
            setattr(app, a, app.test_client(job_run_mode=job_run_mode))
    return getattr(app, a)


def init_auth_db():
    """Force a request that creates a user in db with just myapp"""
    fc = flask_client(sim_types=MYAPP)
    fc.sr_login_as_guest()
    return fc, fc.sr_post('listSimulations', {'simulationType': fc.sr_sim_type})


def sim_data(sim_name=None, sim_type=None, sim_types=CONFTEST_ALL_CODES, cfg=None):
    """Get simulation data

    Args:
        sim_name (str): full name of simulation
        sim_type (str): app [defaults to myapp]
        sim_types (str): `SIREPO_FEATURE_CONFIG_SIM_TYPES` value
    Returns:
        PKDict: simulation data
        object: flask client
    """
    fc = flask_client(sim_types=sim_types or [sim_type or MYAPP], cfg=cfg)
    fc.sr_login_as_guest()
    return fc.sr_sim_data(sim_name=sim_name, sim_type=sim_type), fc


def test_in_request(op, cfg=None, before_request=None, headers=None, want_cookie=True, want_user=True, **kwargs):
    fc = flask_client(cfg, **kwargs)
    try:
        from pykern import pkunit
        from pykern import pkcollections

        if before_request:
            before_request(fc)
        setattr(
            server._app,
            server.SRUNIT_TEST_IN_REQUEST,
            PKDict(op=op, want_cookie=want_cookie, want_user=want_user),
        )
        from sirepo import uri_router
        r = fc.get(
            uri_router.srunit_uri,
            headers=headers,
        )
        pkunit.pkeq(200, r.status_code, 'FAIL: status={}', r.status)
        if r.mimetype == 'text/html':
            m = _JAVASCRIPT_REDIRECT_RE.search(r.data)
            if m:
                pkunit.pkfail('redirect={}', m.group(1))
            pkunit.pkfail('other html response={}', r.data)
        d = pkcollections.json_load_any(r.data)
        pkunit.pkeq('ok', d.get('state'), 'FAIL: data={}', d)
    finally:
        try:
            delattr(server._app, server.SRUNIT_TEST_IN_REQUEST)
        except AttributeError:
            pass
    return r


def wrap_in_request(*args, **kwargs):
    """Decorator for calling functions in `test_in_request`

    Examples:
        # note that the parens are required
        @srunit.wrap_in_request()
        def test_simple():
            inside request context

        @srunit.wrap_in_request(cfg={'SIREPO_AUTH_METHODS': 'github:guest'})
        def test_myapp():
            inside a request context here

    Args:<commit_msg>document failures better<commit_after># -*- coding: utf-8 -*-
u"""Support for unit tests

:copyright: Copyright (c) 2016 RadiaSoft LLC.  All Rights Reserved.
:license: http://www.apache.org/licenses/LICENSE-2.0.html
"""
from __future__ import absolute_import, division, print_function
import flask
import flask.testing
import json
import re
from pykern.pkcollections import PKDict


#: Default "app"
MYAPP = 'myapp'

#: import sirepo.server
server = None

#: app result from server.init
app = None

#: Matches javascript-redirect.html
_JAVASCRIPT_REDIRECT_RE = re.compile(r'window.location = "([^"]+)"')

#: set by conftest.py
CONFTEST_ALL_CODES = None


def flask_client(cfg=None, sim_types=None, job_run_mode=None):
    """Return FlaskClient with easy access methods.

    Creates a new run directory every test file so can assume
    sharing of state on the server within a file (module).

    Two methods of interest: `sr_post` and `sr_get`.

    Args:
        cfg (dict): extra configuration for reset_state_for_testing
        sim_types (str): value for SIREPO_FEATURE_CONFIG_SIM_TYPES [CONFTEST_ALL_CODES]

    Returns:
        FlaskClient: for local requests to Flask server
    """
    global server, app

    a = 'srunit_flask_client'
    if not cfg:
        cfg = PKDict()
    t = sim_types or CONFTEST_ALL_CODES
    if t:
        if isinstance(t, (tuple, list)):
            t = ':'.join(t)
        cfg['SIREPO_FEATURE_CONFIG_SIM_TYPES'] = t
    if not (server and hasattr(app, a)):
        from pykern import pkconfig

        # initialize pkdebug with correct values
        pkconfig.reset_state_for_testing(cfg)

        from pykern import pkunit
        with pkunit.save_chdir_work() as wd:
            from pykern import pkio
            cfg['SIREPO_SRDB_ROOT'] = str(pkio.mkdir_parent(wd.join('db')))
            pkconfig.reset_state_for_testing(cfg)
            from sirepo import server as s

            server = s
            app = server.init()
            app.config['TESTING'] = True
            app.test_client_class = _TestClient
            setattr(app, a, app.test_client(job_run_mode=job_run_mode))
    return getattr(app, a)


def init_auth_db():
    """Force a request that creates a user in db with just myapp"""
    fc = flask_client(sim_types=MYAPP)
    fc.sr_login_as_guest()
    return fc, fc.sr_post('listSimulations', {'simulationType': fc.sr_sim_type})


def sim_data(sim_name=None, sim_type=None, sim_types=CONFTEST_ALL_CODES, cfg=None):
    """Get simulation data

    Args:
        sim_name (str): full name of simulation
        sim_type (str): app [defaults to myapp]
        sim_types (str): `SIREPO_FEATURE_CONFIG_SIM_TYPES` value
    Returns:
        PKDict: simulation data
        object: flask client
    """
    fc = flask_client(sim_types=sim_types or [sim_type or MYAPP], cfg=cfg)
    fc.sr_login_as_guest()
    return fc.sr_sim_data(sim_name=sim_name, sim_type=sim_type), fc


def test_in_request(op, cfg=None, before_request=None, headers=None, want_cookie=True, want_user=True, **kwargs):
    fc = flask_client(cfg, **kwargs)
    try:
        from pykern import pkunit
        from pykern import pkcollections

        if before_request:
            before_request(fc)
        setattr(
            server._app,
            server.SRUNIT_TEST_IN_REQUEST,
            PKDict(op=op, want_cookie=want_cookie, want_user=want_user),
        )
        from sirepo import uri_router
        r = fc.get(
            uri_router.srunit_uri,
            headers=headers,
        )
        pkunit.pkeq(200, r.status_code, 'FAIL: unexpected status={}', r.status)
        if r.mimetype == 'text/html':
            m = _JAVASCRIPT_REDIRECT_RE.search(r.data)
            if m:
                pkunit.pkfail('unexpected redirect={}', m.group(1))
            pkunit.pkfail('unexpected html response={}', r.data)
        d = pkcollections.json_load_any(r.data)
        pkunit.pkeq('ok', d.get('state'), 'FAIL: expecting state=ok, but got data={}', d)
    finally:
        try:
            delattr(server._app, server.SRUNIT_TEST_IN_REQUEST)
        except AttributeError:
            pass
    return r


def wrap_in_request(*args, **kwargs):
    """Decorator for calling functions in `test_in_request`

    Examples:
        # note that the parens are required
        @srunit.wrap_in_request()
        def test_simple():
            inside request context

        @srunit.wrap_in_request(cfg={'SIREPO_AUTH_METHODS': 'github:guest'})
        def test_myapp():
            inside a request context here

    Args:
{'fraction_non_alphanumeric': 0.09918087226034979, 'fraction_numerical': 0.0024352446313925173, 'mean_word_length': 2.763015410245731, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1019761', 'n_tokens_mistral': 2988, 'n_tokens_neox': 2903, 'n_words': 731}