# Path: databench/analyses_packaged/dummypi/analysis.py
# class Dummypi(databench.Analysis):
#     """A dummy analysis."""
# 
#     @databench.on
#     def connected(self):
#         yield self.data.init({'samples': 100000})
# 
#     @databench.on
#     def run(self):
#         """Run when button is pressed."""
# 
#         inside = 0
#         for draws in range(1, self.data['samples']):
#             # generate points and check whether they are inside the unit circle
#             r1 = random.random()
#             r2 = random.random()
#             if r1 ** 2 + r2 ** 2 < 1.0:
#                 inside += 1
# 
#             # every 1000 iterations, update status
#             if draws % 1000 != 0:
#                 continue
# 
#             # debug
#             yield self.emit('log', {'draws': draws, 'inside': inside})
# 
#             # calculate pi and its uncertainty given the current draws
#             p = inside / draws
#             pi = {
#                 'estimate': 4.0 * p,
#                 'uncertainty': 4.0 * math.sqrt(draws * p * (1.0 - p)) / draws,
#             }
# 
#             # send status to frontend
#             yield self.set_state(pi=pi)
# 
#         yield self.emit('log', {'action': 'done'})
# 
#     @databench.on
#     def samples(self, value):
#         yield self.set_state(samples=value)
# 
# Path: databench/testing.py
# class AnalysisTest(object):
#     """Unit test wrapper for an analysis.
# 
#     :param databench.Analysis analysis: The analysis to test.
#     :param str cli_args: Command line interface arguments.
#     :param str request_args: Request arguments.
#     :param meta: An object with a `run_process` attribute.
# 
#     Trigger actions using the `~.trigger` method.
#     All outgoing messages to the frontend are captured in `emitted_messages`.
# 
#     There are two main options for constructing tests: decorating with
#     `tornado.testing.gen_test` and yielding `~tornado.concurrent.Future`
#     objects (block until
#     future is done) or to use :meth:`~tornado.testing.AsyncTestCase.wait` and
#     :meth:`~tornado.testing.AsyncTestCase.stop` in callbacks.
#     For detailed information on ioloops within the Tornado testing framework,
#     please consult `tornado.testing`.
# 
#     :ivar list cli_args: command line arguments
#     :ivar dict request_args: request arguments
#     :ivar list emitted_messages: all emitted (``signal``, ``message``) pairs
# 
#     Examples:
# 
#     .. literalinclude:: ../databench/tests/test_testing.py
#         :language: python
#     """
#     def __init__(self, analysis, cli_args=None, request_args=None, meta=None):
#         self.analysis = analysis
#         self.analysis_instance = analysis()
#         self.cli_args = cli_args
#         self.request_args = request_args
#         self.meta = meta or Meta
#         self.emitted_messages = []
# 
#         Meta.fill_action_handlers(analysis)
# 
#         # initialize
#         self.analysis_instance.init_databench()
#         self.analysis_instance.set_emit_fn(self.emulate_emit_to_frontend)
#         self.trigger('connect')
#         self.trigger('args', [cli_args, request_args])
#         self.trigger('connected')
# 
#     def emulate_emit_to_frontend(self, signal, message):
#         self.emitted_messages.append((signal, message))
# 
#     def trigger(self, action_name, message='__nomessagetoken__', **kwargs):
#         """Trigger an `on` callback.
# 
#         :param str action_name: Name of the action to trigger.
#         :param message: Message.
#         :param callback:
#             A callback function when done (e.g.
#             `~tornado.testing.AsyncTestCase.stop` in tests).
#         :rtype: tornado.concurrent.Future
#         """
#         return self.meta.run_process(
#             self.analysis_instance, action_name, message, **kwargs)


import databench
import tornado.testing
from databench.analyses_packaged.dummypi.analysis import Dummypi
from databench.testing import AnalysisTest


class Parameters(databench.Analysis):
    @databench.on
    def test_data(self, key, value):
        yield self.set_state({key: value})


class Example(tornado.testing.AsyncTestCase):
    @tornado.testing.gen_test
    def test_data(self):
        test = AnalysisTest(Parameters)
        yield test.trigger('test_data', ['light', 'red'])
        self.assertIn(('data', {'light': 'red'}), test.emitted_messages)

    @tornado.testing.gen_test
    def test_process(self):
        test = AnalysisTest(Parameters)
        yield test.trigger('test_data', {'key': 'light',
                                         'value': 'red',
                                         '__process_id': 3})
        self.assertEqual([
            ('__process', {'id': 3, 'status': 'start'}),
            ('data', {'light': 'red'}),
            ('__process', {'id': 3, 'status': 'end'}),
        ], test.emitted_messages)

    @tornado.testing.gen_test
    def test_multiple_emits(self):

# Path: databench/analyses_packaged/dummypi/analysis.py
# class Dummypi(databench.Analysis):
#     """A dummy analysis."""
# 
#     @databench.on
#     def connected(self):
#         yield self.data.init({'samples': 100000})
# 
#     @databench.on
#     def run(self):
#         """Run when button is pressed."""
# 
#         inside = 0
#         for draws in range(1, self.data['samples']):
#             # generate points and check whether they are inside the unit circle
#             r1 = random.random()
#             r2 = random.random()
#             if r1 ** 2 + r2 ** 2 < 1.0:
#                 inside += 1
# 
#             # every 1000 iterations, update status
#             if draws % 1000 != 0:
#                 continue
# 
#             # debug
#             yield self.emit('log', {'draws': draws, 'inside': inside})
# 
#             # calculate pi and its uncertainty given the current draws
#             p = inside / draws
#             pi = {
#                 'estimate': 4.0 * p,
#                 'uncertainty': 4.0 * math.sqrt(draws * p * (1.0 - p)) / draws,
#             }
# 
#             # send status to frontend
#             yield self.set_state(pi=pi)
# 
#         yield self.emit('log', {'action': 'done'})
# 
#     @databench.on
#     def samples(self, value):
#         yield self.set_state(samples=value)
# 
# Path: databench/testing.py
# class AnalysisTest(object):
#     """Unit test wrapper for an analysis.
# 
#     :param databench.Analysis analysis: The analysis to test.
#     :param str cli_args: Command line interface arguments.
#     :param str request_args: Request arguments.
#     :param meta: An object with a `run_process` attribute.
# 
#     Trigger actions using the `~.trigger` method.
#     All outgoing messages to the frontend are captured in `emitted_messages`.
# 
#     There are two main options for constructing tests: decorating with
#     `tornado.testing.gen_test` and yielding `~tornado.concurrent.Future`
#     objects (block until
#     future is done) or to use :meth:`~tornado.testing.AsyncTestCase.wait` and
#     :meth:`~tornado.testing.AsyncTestCase.stop` in callbacks.
#     For detailed information on ioloops within the Tornado testing framework,
#     please consult `tornado.testing`.
# 
#     :ivar list cli_args: command line arguments
#     :ivar dict request_args: request arguments
#     :ivar list emitted_messages: all emitted (``signal``, ``message``) pairs
# 
#     Examples:
# 
#     .. literalinclude:: ../databench/tests/test_testing.py
#         :language: python
#     """
#     def __init__(self, analysis, cli_args=None, request_args=None, meta=None):
#         self.analysis = analysis
#         self.analysis_instance = analysis()
#         self.cli_args = cli_args
#         self.request_args = request_args
#         self.meta = meta or Meta
#         self.emitted_messages = []
# 
#         Meta.fill_action_handlers(analysis)
# 
#         # initialize
#         self.analysis_instance.init_databench()
#         self.analysis_instance.set_emit_fn(self.emulate_emit_to_frontend)
#         self.trigger('connect')
#         self.trigger('args', [cli_args, request_args])
#         self.trigger('connected')
# 
#     def emulate_emit_to_frontend(self, signal, message):
#         self.emitted_messages.append((signal, message))
# 
#     def trigger(self, action_name, message='__nomessagetoken__', **kwargs):
#         """Trigger an `on` callback.
# 
#         :param str action_name: Name of the action to trigger.
#         :param message: Message.
#         :param callback:
#             A callback function when done (e.g.
#             `~tornado.testing.AsyncTestCase.stop` in tests).
#         :rtype: tornado.concurrent.Future
#         """
#         return self.meta.run_process(
#             self.analysis_instance, action_name, message, **kwargs)


# Path: databench/tests/test_analysistest.py
import databench
import tornado.testing
from databench.analyses_packaged.dummypi.analysis import Dummypi
from databench.testing import AnalysisTest



class Parameters(databench.Analysis):
    @databench.on
    def test_data(self, key, value):
        yield self.set_state({key: value})


class Example(tornado.testing.AsyncTestCase):
    @tornado.testing.gen_test
    def test_data(self):
        test = AnalysisTest(Parameters)
        yield test.trigger('test_data', ['light', 'red'])
        self.assertIn(('data', {'light': 'red'}), test.emitted_messages)

    @tornado.testing.gen_test
    def test_process(self):
        test = AnalysisTest(Parameters)
        yield test.trigger('test_data', {'key': 'light',
                                         'value': 'red',
                                         '__process_id': 3})
        self.assertEqual([
            ('__process', {'id': 3, 'status': 'start'}),
            ('data', {'light': 'red'}),
            ('__process', {'id': 3, 'status': 'end'}),
        ], test.emitted_messages)

    @tornado.testing.gen_test
    def test_multiple_emits(self):

        test = AnalysisTest(Dummypi)