<commit_before>import os
import json
import time
import logging
import uuid
import threading

import httpretty
from nose.tools import assert_equal

import ckanserviceprovider.web as web
import ckanserviceprovider.job as job
import ckanserviceprovider.util as util
import ckanserviceprovider.db as db


# The callback URL that ckanserviceprovider will post to when the
# asynchronous background job finishes. We will mock this.
RESULT_URL = "http://demo.ckan.org/ckanserviceprovider/result_url"


def configure():
    """Configure the Flask app.

    This has to be called just once per test run (not e.g. once for each test).

    """
    os.environ['JOB_CONFIG'] = os.path.join(
        os.path.dirname(__file__), 'settings_test.py')
    web.init()
configure()


def reset_db():
    """Reset the database and scheduler.

    Should be called after each test.

    """
    web.scheduler.shutdown(wait=True)
    db.drop_all()
    db.init(web.app.config.get('SQLALCHEMY_DATABASE_URI'))
    web.init_scheduler(web.app.config.get('SQLALCHEMY_DATABASE_URI'))


def test_client():
    """Return a test client for the ckanserviceprovider web app."""
    return web.app.test_client()


def login(client, username='testadmin', password='<PASSWORD>'):
    return client.post('/login', data=dict(
        username=username,
        password=password
    ), follow_redirects=True)


def _make_request_callback_function(event):
    """Return an httpretty request callback function that sets the given event.

    This is a helper function for mock_result_url() below.

    """
    def request_callback(request, uri, headers):
        event.set()
    return request_callback


def mock_result_url(result_url):
    """Mock the given CKAN Service Provider result URL.

    Returns a threading.Event object that you can use to wait for the mock URL
    to be called by doing: event.wait().

    The way it works is:

    * A test method calls this function to mock a result_url, receives a
      threading event object in return.

    * The test method posts to ckanserviceprovider passing the mocked
      result_url.

    * ckanserviceprovider kicks off an asynchronous background job.

    * The test method waits for ckanserviceprovider's asynchronous background
      job to finish by doing event.wait().

    * When the job finishes ckanserviceprovider posts to the result_url.

    * The post is intercepted and redirected to a function that sets the
      thread event.

    * event.wait() returns and the test method continues.

    """
    event = threading.Event()
    request_callback = _make_request_callback_function(event)
    httpretty.register_uri(
        httpretty.POST, result_url, body=request_callback)
    return event


@job.sync
def echo(task_id, input):
    if input['data'].startswith('>'):
        raise util.JobError('Do not start message with >')
    if input['data'].startswith('#'):
        raise Exception('Something went totally wrong')
    return '>' + input['data']


@job.sync
def echo_raw(task_id, input):
    if input['data'].startswith('>'):
        raise util.JobError('Do not start message with >')

    def raw():
        for x in sorted(input['data']):
            yield x

    return raw


@job.async
def example(task_id, input):
    if 'time' not in input['data']:
        raise util.JobError('time not in input')

    time.sleep(input['data']['time'])
    return 'Slept for ' + str(input['data']['time']) + ' seconds.'


@job.async
def failing(task_id, input):
    time.sleep(0.1)
    raise util.JobError('failed')


@job.async
def log(task_id, input):
    handler = util.StoringHandler(task_id, input)
    logger = logging.Logger(task_id)
    logger.addHandler(handler)

    logger.warn('Just a warning')


class TestWeb():

    def teardown(self):
        reset_db()

    def test_status(self):
        '''/status should return JSON with the app version, job types, etc.'''
        client = test_client()
        rv = client.get('/status')
        status_data = json.loads(rv.data)
        status_data.pop('stats')
        assert_equal(status_data, dict(version=0.1,
                                       job_types=['failing',
                                                  'example',
                                                  'log',
                                                  'echo_raw',<commit_msg>Make TestWeb into a new-style class<commit_after>import os
import json
import time
import logging
import uuid
import threading

import httpretty
from nose.tools import assert_equal

import ckanserviceprovider.web as web
import ckanserviceprovider.job as job
import ckanserviceprovider.util as util
import ckanserviceprovider.db as db


# The callback URL that ckanserviceprovider will post to when the
# asynchronous background job finishes. We will mock this.
RESULT_URL = "http://demo.ckan.org/ckanserviceprovider/result_url"


def configure():
    """Configure the Flask app.

    This has to be called just once per test run (not e.g. once for each test).

    """
    os.environ['JOB_CONFIG'] = os.path.join(
        os.path.dirname(__file__), 'settings_test.py')
    web.init()
configure()


def reset_db():
    """Reset the database and scheduler.

    Should be called after each test.

    """
    web.scheduler.shutdown(wait=True)
    db.drop_all()
    db.init(web.app.config.get('SQLALCHEMY_DATABASE_URI'))
    web.init_scheduler(web.app.config.get('SQLALCHEMY_DATABASE_URI'))


def test_client():
    """Return a test client for the ckanserviceprovider web app."""
    return web.app.test_client()


def login(client, username='testadmin', password='<PASSWORD>'):
    return client.post('/login', data=dict(
        username=username,
        password=password
    ), follow_redirects=True)


def _make_request_callback_function(event):
    """Return an httpretty request callback function that sets the given event.

    This is a helper function for mock_result_url() below.

    """
    def request_callback(request, uri, headers):
        event.set()
    return request_callback


def mock_result_url(result_url):
    """Mock the given CKAN Service Provider result URL.

    Returns a threading.Event object that you can use to wait for the mock URL
    to be called by doing: event.wait().

    The way it works is:

    * A test method calls this function to mock a result_url, receives a
      threading event object in return.

    * The test method posts to ckanserviceprovider passing the mocked
      result_url.

    * ckanserviceprovider kicks off an asynchronous background job.

    * The test method waits for ckanserviceprovider's asynchronous background
      job to finish by doing event.wait().

    * When the job finishes ckanserviceprovider posts to the result_url.

    * The post is intercepted and redirected to a function that sets the
      thread event.

    * event.wait() returns and the test method continues.

    """
    event = threading.Event()
    request_callback = _make_request_callback_function(event)
    httpretty.register_uri(
        httpretty.POST, result_url, body=request_callback)
    return event


@job.sync
def echo(task_id, input):
    if input['data'].startswith('>'):
        raise util.JobError('Do not start message with >')
    if input['data'].startswith('#'):
        raise Exception('Something went totally wrong')
    return '>' + input['data']


@job.sync
def echo_raw(task_id, input):
    if input['data'].startswith('>'):
        raise util.JobError('Do not start message with >')

    def raw():
        for x in sorted(input['data']):
            yield x

    return raw


@job.async
def example(task_id, input):
    if 'time' not in input['data']:
        raise util.JobError('time not in input')

    time.sleep(input['data']['time'])
    return 'Slept for ' + str(input['data']['time']) + ' seconds.'


@job.async
def failing(task_id, input):
    time.sleep(0.1)
    raise util.JobError('failed')


@job.async
def log(task_id, input):
    handler = util.StoringHandler(task_id, input)
    logger = logging.Logger(task_id)
    logger.addHandler(handler)

    logger.warn('Just a warning')


class TestWeb(object):

    def teardown(self):
        reset_db()

    def test_status(self):
        '''/status should return JSON with the app version, job types, etc.'''
        client = test_client()
        rv = client.get('/status')
        status_data = json.loads(rv.data)
        status_data.pop('stats')
        assert_equal(status_data, dict(version=0.1,
                                       job_types=['failing',
                                                  'example',
                                                  'log',
                                                  'echo_raw',
{'fraction_non_alphanumeric': 0.1006872852233677, 'fraction_numerical': 0.0009163802978235968, 'mean_word_length': 2.8411790585129784, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '186049', 'n_tokens_mistral': 2537, 'n_tokens_neox': 2456, 'n_words': 778}