<commit_before>import json
import os
import random
import re
import time

import jpegtran
import mock
import pytest
from multiprocessing.pool import ThreadPool

from spreads.workflow import signals as workflow_signals
from conftest import TestPluginOutput, TestPluginProcess, TestPluginProcessB


@pytest.yield_fixture
def app(config, tmpdir):
    from spreadsplug.web import WebApplication, app
    config.load_defaults(overwrite=False)

    config['web']['mode'] = 'full'
    config['web']['project_dir'] = unicode(tmpdir.join('workflows'))
    config['web']['debug'] = False
    config['web']['standalone_device'] = True
    config['web']['postprocessing_server'] = ''

    webapp = WebApplication(config)
    webapp.setup_logging()
    with mock.patch('spreadsplug.web.task_queue') as mock_tq:
        mock_tq.task.return_value = lambda x: x
        webapp.setup_signals()
    webapp.setup_tornado()

    app.config['TESTING'] = True
    yield app


@pytest.yield_fixture
def client(app):
    yield app.test_client()


@pytest.yield_fixture
def mock_dbus(tmpdir):
    with mock.patch.multiple('dbus', SystemBus=mock.DEFAULT,
                             Interface=mock.DEFAULT) as values:
        stickdir = tmpdir.join('stick')
        stickdir.mkdir()
        mockdevs = [mock.Mock(), mock.Mock()]*2
        mockobj = mock.MagicMock()
        mockobj.get_dbus_method.return_value.return_value = unicode(
            stickdir)
        mockobj.EnumerateDevices.return_value = mockdevs
        mockobj.Get.side_effect = [True, 'usb', True]*2
        values['Interface'].return_value = mockobj
        yield mockobj


@pytest.yield_fixture
def worker():
    from spreadsplug.web.worker import ProcessingWorker
    worker = ProcessingWorker()
    worker.start()
    time.sleep(0.1)
    yield
    worker.stop()


def create_workflow(client, num_captures='random'):
    workflow = {
        'metadata': {'title': 'test{0}'.format(random.randint(0, 8192))},
    }
    data = json.loads(client.post('/api/workflow',
                      data=json.dumps(workflow)).data)
    if num_captures:
        client.post('/api/workflow/{0}/prepare_capture'.format(data['id']))
        for _ in xrange(random.randint(1, 4)
                        if num_captures == 'random' else num_captures):
            client.post('/api/workflow/{0}/capture'.format(data['id']))
        client.post('/api/workflow/{0}/finish_capture'.format(data['id']))
    return data['id']


def test_index(client):
    rv = client.get('/')
    assert "<title>spreads</title>" in rv.data
    assert "<script src=\"/static/bundle.js\"></script>" in rv.data

    cfg = json.loads(re.findall(r"window.config = ({.*});", rv.data)[0])
    assert cfg['plugins'] == ['test_output', 'test_process', 'test_process2']
    assert cfg['driver'] == 'testdriver'
    assert cfg['web']['mode'] == 'full'
    templates = json.loads(re.findall(r"window.pluginTemplates = ({.*});",
                                      rv.data)[0])
    assert 'a_boolean' in templates['test_process']
    assert 'flip_target_pages' in templates['device']
    assert 'string' in templates['test_output']


def test_get_plugins(client):
    exts = [mock.Mock(), mock.Mock(), mock.Mock()]
    exts[0].name = "test_output"
    exts[0].load.return_value = TestPluginOutput
    exts[1].name = "test_process"
    exts[1].load.return_value = TestPluginProcess
    exts[2].name = "test_process2"
    exts[2].load.return_value = TestPluginProcessB
    with mock.patch('spreadsplug.web.web.pkg_resources') as pkgr:
        pkgr.iter_entry_points.return_value = exts
        rv = client.get('/api/plugins')
    plugins = json.loads(rv.data)
    assert plugins['output'] == ["test_output"]
    assert plugins['postprocessing'] == ["test_process", "test_process2"]


def test_get_plugin_templates(client):
    rv = client.get('/api/plugins/templates')
    templates = json.loads(rv.data)
    assert 'a_boolean' in templates['test_process']
    assert 'flip_target_pages' in templates['device']
    assert 'string' in templates['test_output']


def test_create_workflow(client):
    wf = json.dumps({<commit_msg>web: Fix asserts in templates tests<commit_after>import json
import os
import random
import re
import time

import jpegtran
import mock
import pytest
from multiprocessing.pool import ThreadPool

from spreads.workflow import signals as workflow_signals
from conftest import TestPluginOutput, TestPluginProcess, TestPluginProcessB


@pytest.yield_fixture
def app(config, tmpdir):
    from spreadsplug.web import WebApplication, app
    config.load_defaults(overwrite=False)

    config['web']['mode'] = 'full'
    config['web']['project_dir'] = unicode(tmpdir.join('workflows'))
    config['web']['debug'] = False
    config['web']['standalone_device'] = True
    config['web']['postprocessing_server'] = ''

    webapp = WebApplication(config)
    webapp.setup_logging()
    with mock.patch('spreadsplug.web.task_queue') as mock_tq:
        mock_tq.task.return_value = lambda x: x
        webapp.setup_signals()
    webapp.setup_tornado()

    app.config['TESTING'] = True
    yield app


@pytest.yield_fixture
def client(app):
    yield app.test_client()


@pytest.yield_fixture
def mock_dbus(tmpdir):
    with mock.patch.multiple('dbus', SystemBus=mock.DEFAULT,
                             Interface=mock.DEFAULT) as values:
        stickdir = tmpdir.join('stick')
        stickdir.mkdir()
        mockdevs = [mock.Mock(), mock.Mock()]*2
        mockobj = mock.MagicMock()
        mockobj.get_dbus_method.return_value.return_value = unicode(
            stickdir)
        mockobj.EnumerateDevices.return_value = mockdevs
        mockobj.Get.side_effect = [True, 'usb', True]*2
        values['Interface'].return_value = mockobj
        yield mockobj


@pytest.yield_fixture
def worker():
    from spreadsplug.web.worker import ProcessingWorker
    worker = ProcessingWorker()
    worker.start()
    time.sleep(0.1)
    yield
    worker.stop()


def create_workflow(client, num_captures='random'):
    workflow = {
        'metadata': {'title': 'test{0}'.format(random.randint(0, 8192))},
    }
    data = json.loads(client.post('/api/workflow',
                      data=json.dumps(workflow)).data)
    if num_captures:
        client.post('/api/workflow/{0}/prepare_capture'.format(data['id']))
        for _ in xrange(random.randint(1, 4)
                        if num_captures == 'random' else num_captures):
            client.post('/api/workflow/{0}/capture'.format(data['id']))
        client.post('/api/workflow/{0}/finish_capture'.format(data['id']))
    return data['id']


def test_index(client):
    rv = client.get('/')
    assert "<title>spreads</title>" in rv.data
    assert "<script src=\"/static/bundle.js\"></script>" in rv.data

    cfg = json.loads(re.findall(r"window.config = ({.*});", rv.data)[0])
    assert cfg['plugins'] == ['test_output', 'test_process', 'test_process2']
    assert cfg['driver'] == 'testdriver'
    assert cfg['web']['mode'] == 'full'
    templates = json.loads(re.findall(r"window.configTemplates = ({.*});",
                                      rv.data)[0])
    assert 'a_boolean' in templates['test_process']
    assert 'flip_target_pages' in templates['device']
    assert 'string' in templates['test_output']


def test_get_plugins(client):
    exts = [mock.Mock(), mock.Mock(), mock.Mock()]
    exts[0].name = "test_output"
    exts[0].load.return_value = TestPluginOutput
    exts[1].name = "test_process"
    exts[1].load.return_value = TestPluginProcess
    exts[2].name = "test_process2"
    exts[2].load.return_value = TestPluginProcessB
    with mock.patch('spreadsplug.web.web.pkg_resources') as pkgr:
        pkgr.iter_entry_points.return_value = exts
        rv = client.get('/api/plugins')
    plugins = json.loads(rv.data)
    assert plugins['output'] == ["test_output"]
    assert plugins['postprocessing'] == ["test_process", "test_process2"]


def test_get_plugin_templates(client):
    rv = client.get('/api/templates')
    templates = json.loads(rv.data)
    assert 'a_boolean' in templates['test_process']
    assert 'flip_target_pages' in templates['device']
    assert 'string' in templates['test_output']


def test_create_workflow(client):
    wf = json.dumps({
{'fraction_non_alphanumeric': 0.14258048898571773, 'fraction_numerical': 0.006293875574921327, 'mean_word_length': 3.4689021092482424, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '871207', 'n_tokens_mistral': 2658, 'n_tokens_neox': 2588, 'n_words': 488}