def test_PARALLEL_SPLIT01(self):
        we = GenericWorkflowEngine()
        doc = self.getDoc()

        we.setWorkflow([i('start'),
                        cf.PARALLEL_SPLIT(
            printer('p1'),
            printer('p2'),
            printer('p3'),
            printer('p4'),
            printer('p5')),
            lambda o, e: time.sleep(.1),
            a('end')
        ])
        we.process(doc)
        r = [' '.join(doc[x]) for x in range(len(doc))]

        assert doc[0][0] == 'start'
        assert doc[0][1] == 'one'
        assert doc[1][0] == 'start'
        assert doc[1][1] == 'two'

        # end must have been inserted while printers were running
        # mixed together with them
        all_pos = set()
        for x in range(len(doc)):
            pos = doc[x].index('end')
            assert pos > 2
            assert pos < len(doc[x])
            all_pos.add(pos)

    def test_workflow01(self):

        class GenericWEWithXChooser(GenericWorkflowEngine):
            def callback_chooser(self, obj):
                return self.callbacks.get('x')

        we0 = GenericWorkflowEngine()
        we1 = GenericWorkflowEngine()
        we2 = GenericWEWithXChooser()

        we0.addManyCallbacks('*', [
            obj_append('mouse'),
            [obj_append('dog'), jump_call(1), obj_append('cat'), obj_append('puppy')],
            obj_append('horse'),
        ])
        we1.setWorkflow([
            obj_append('mouse'),
            [obj_append('dog'), jump_call(1), obj_append('cat'), obj_append('puppy')],
            obj_append('horse'),
        ])
        we2.addManyCallbacks('x', [
            obj_append('mouse'),
            [obj_append('dog'), jump_call(1), obj_append('cat'), obj_append('puppy')],
            obj_append('horse'),
        ])

        we0.process(self.d0)
        we1.process(self.d1)
        we2.process(self.d2)

        assert self.d0 == self.d1
        assert self.d0 == self.d2

    def test_CHOICE03(self):
        we = GenericWorkflowEngine()
        doc = self.getDoc()[0:1]

        def arbiter(obj, eng):
            return obj[-1]

        we.setWorkflow([i('start'),
                        cf.CHOICE(arbiter,
                                  ('bam', lambda obj, eng: obj.append('bom')),
                                  ('end', lambda obj,
                                   eng: obj.append('error')),
                                  ('bim', lambda obj, eng: obj.append('bam')),
                                  bom=(lambda obj, eng: obj.append('bum')),
                                  one=(lambda obj, eng: obj.append('bim')),
                                  bum=cf.STOP(),
                                  ),
                        cf.TASK_JUMP_BWD(-1)])
        we.process(doc)

        d = ' '.join(doc[0])

        assert 'bim bam bom bum' in d
        assert 'error' not in d
        assert len(doc[0]) == 6

    def test_nested_workflow_halt(self):
        other_wfe = GenericWorkflowEngine()
        wfe = self.wfe

        other_wfe.callbacks.add_many([
            m('mouse'),
            [
                m('dog'),
                [m('cat'), m('puppy')],
                [m('python'), halt_processing()],
                m('horse'),
            ]
        ], self.key)

        wfe.callbacks.add_many([
            m('mouse'),
            [
                m('dog'),
                [m('cat'), m('puppy')],
                [m('python'), lambda o, e: other_wfe.process(self.tokens)],
                m('horse'),
            ]
        ], self.key)
        with pytest.raises(HaltProcessing):
            wfe.process(self.tokens)

        t = get_first(self.tokens)
        assert get_xth(self.tokens, 0) == 'mouse dog cat puppy python mouse dog cat puppy python'
        assert get_xth(self.tokens, 1) is None
        assert get_xth(self.tokens, 2) is None

    def test_PARALLEL_SPLIT02(self):
        """TODO: this test is failing, but that is because sometimes it does
        not take into accounts threads being executed in random mannger"""

        we = GenericWorkflowEngine()
        doc = self.getDoc()[0:1]

        we.setWorkflow([i('start'),
                        cf.PARALLEL_SPLIT(
                            [
                                cf.PARALLEL_SPLIT(
                                    printer('p0'),
                                    printer('p0a'),
                                    cf.PARALLEL_SPLIT(
                                        printer('p0b'),
                                        printer('p0c')
                                    ),
                                ),
                                printer('xx')
                            ],
                            [
                                a('AAA'),
                                printer('p2b')
                            ],
                            printer('p3'),
                            [
                                a('p4a'),
                                printer('p4b'),
                                printer('p4c')
                            ],
                            [printer('p5'), cf.PARALLEL_SPLIT(
                                printer('p6'),
                                printer('p7'),
                                [printer('p8a'), printer('p8b')],
                            )]),
                        a('end')
                        ])
        we.process(doc)

        # give threads time to finish
        time.sleep(2)

        assert doc[0][0] == 'start'
        assert doc[0][1] == 'one'

        # at least the fist object should have them all
        # print doc[0]
        for x in ['p0', 'p0a', 'p0b', 'p0c', 'xx', 'AAA', 'p2b', 'p3', 'p4a',
                  'p4b', 'p4c', 'p5', 'p6', 'p8a', 'p8b']:
            doc[0].index(x)  # will fail if not present

    def test_init(self):

        # init with empty to full parameters
        we1 = GenericWorkflowEngine()

        callbacks = [
            obj_append('mouse'),
            [obj_append('dog'), jump_call(1), obj_append('cat'), obj_append('puppy')],
            obj_append('horse'),
        ]

        we1.addManyCallbacks('*', deepcopy(callbacks))

        we1.process(self.d1)

    def test_RUN_WF02(self):
        """Test wfe is reinit=True - eng must not remember"""
        we = GenericWorkflowEngine()
        doc = self.getDoc()[0:1]

        we.callbacks.replace(
            [
                i('start'),
                ut.RUN_WF(
                    [
                        lambda obj, eng: obj.append('bom'),
                        lambda obj, eng: obj.append('bam'),
                        lambda obj, eng: obj.append('bum'),
                        lambda obj, eng: obj.append('end'),
                        lambda obj, eng: obj.append(
                            eng.store.setdefault('eng-end', '')),
                        e('eng-end', 'eng-end')
                    ],
                    data_connector=lambda obj, eng: [obj],
                    outkey='#wfe',
                    reinit=True
                ),
            ]
        )
        we.process(doc)

        d = ' '.join(doc[0])

        assert 'start' in d
        assert 'bom' in d
        assert 'bam' in d
        assert 'bum' in d
        assert 'end' in d
        assert 'eng-end' not in d

        # run the same thing again
        we.process(doc)

        d = ' '.join(doc[0])
        assert 'start' in d
        assert d.count('bom') == 2
        assert d.count('bam') == 2
        assert d.count('bum') == 2
        assert 'end' in d
        assert 'eng-end' not in d  # it must not be present if reinit=True

    def test_IF_ELSE02(self):
        we = GenericWorkflowEngine()
        doc = self.getDoc()

        we.setWorkflow([i('add'),
                        cf.IF_ELSE(lambda o, e: o[1] == 'three',
                                   a('3'),
                                   a('other'))
                        ])
        we.process(doc)

        r = [' '.join(doc[x]) for x in range(len(doc))]

        assert r[0] == 'add one other'
        assert r[1] == 'add two other'
        assert r[2] == 'add three 3'
        assert r[3] == 'add four other'
        assert r[4] == 'add five other'

    def test_PARALLEL_SPLIT03(self):
        we = GenericWorkflowEngine()
        doc = self.getDoc()

        we.setWorkflow([i('start'),
                        cf.PARALLEL_SPLIT(
            [cf.IF(lambda obj, eng: 'jump-verified' in obj, a('error')),
             cf.PARALLEL_SPLIT(
                [cf.IF(lambda obj, eng: 'nasty-jump' in obj,
                       [a('jump-ok'),
                        lambda obj, eng: ('nasty-jump' in obj and
                                          obj.append('jump-verified'))]),
                 cf.PARALLEL_SPLIT(
                    a('ok-1'),
                    a('ok-2'),
                    cf.IF(lambda obj, eng: 'ok-3' not in obj,
                          lambda obj, eng: (obj.append('ok-3') and
                                            eng.breakFromThisLoop())),
                    a('ok-4')),

                 a('xx'),
                 lambda obj, eng: ('jump-verified' in obj and
                                   eng.breakFromThisLoop()),
                 a('nasty-jump'),
                 cf.TASK_JUMP_IF(
                    lambda obj, eng: 'jump-verified' not in obj, -100)]),
             ],
            [a('AAA'), a('p2b')]),
            a('end')
        ])
        we.process(doc)
        # give threads time to finish
        time.sleep(.5)

        d = doc[0]

        # at least the fist object should have them all
        # print doc[0]
        for x in ['nasty-jump', 'jump-verified', 'ok-3']:
            d.index(x)  # will fail if not present

        assert d.count('ok-1') > 1

    def test_SIMPLE_MERGE03(self):
        we = GenericWorkflowEngine()
        doc = self.getDoc()[0:1]

        we.setWorkflow([i('start'),
                        cf.SIMPLE_MERGE(
            lambda obj, eng: obj.append('bom'),
            lambda obj, eng: obj.append('error'),
            lambda obj, eng: obj.append('bam'),
            lambda obj, eng: obj.append('bum'),
            lambda obj, eng: obj.append('end'),
        ),
        ])
        we.process(doc)

        d = ' '.join(doc[0])

        assert 'start' in d
        assert 'bom' in d
        assert 'error' not in d
        assert 'end' in d

    def test_configure(self):

        callbacks_list = [
            obj_append('mouse'),
            [obj_append('dog'), jump_call(1), obj_append('cat'), obj_append('puppy')],
            obj_append('horse'),
        ]

        we = GenericWorkflowEngine()
        we.addManyCallbacks('*', callbacks_list)

        # process using defaults
        we.process(self.d1)
        r = 'one mouse dog cat puppy horse'.split()

        we = GenericWorkflowEngine()
        we.addManyCallbacks('*', callbacks_list)
        we.process(self.d2)

        assert self.d1[0] == r
        assert self.d2[0] == r
        assert self.d1 == self.d2

    def test_IF_ELSE03(self):
        we = GenericWorkflowEngine()
        doc = self.getDoc()

        doc[3].append('4')

        def test(v):
            return lambda o, e: v in o

        we.setWorkflow([i('add'),
                        cf.IF_ELSE(
            test('three'),
            [a('xxx'), cf.IF_ELSE(test('xxx'),
                                  [a('6'), cf.IF_ELSE(
                                      test('6'),
                                      a('six'),
                                      (a('only-3s'), a('error')))],
                                  a('ok'))],
            [cf.IF_ELSE(
                test('4'),
                cf.IF_ELSE(test('four'),
                           [a('44'), [[[a('forty')]]]],
                           a('error')),
                a('not-four'))]),
            a('end'),
            cf.IF_ELSE(test('error'),
                       a('gosh!'),
                       a('OK'))
        ])
        we.process(doc)

        r = [' '.join(doc[x]) for x in range(len(doc))]

        assert r[0] == 'add one not-four end OK'
        assert r[1] == 'add two not-four end OK'
        assert r[2] == 'add three xxx 6 six end OK'
        assert r[3] == 'add four 4 44 forty end OK'
        assert r[4] == 'add five not-four end OK'

    def test_init(self):
        d1 = self.getDoc()
        d2 = self.getDoc()
        d3 = self.getDoc()

        # init with empty to full parameters
        we1 = GenericWorkflowEngine()
        we2 = GenericWorkflowEngine(callback_chooser=asterisk_chooser)

        try:
            we3 = GenericWorkflowEngine(processing_factory='x',
                                        callback_chooser='x',
                                        before_processing='x',
                                        after_processing='x')
        except Exception as msg:
            assert 'must be a callable' in str(msg)

        try:
            we3 = GenericWorkflowEngine(callback_chooser=asterisk_chooser,
                                        after_processing='x')
        except Exception as msg:
            assert 'must be a callable' in str(msg)

        we1.addManyCallbacks('*', [
            m('mouse'),
            [m('dog'), call_forward(1), m('cat'), m('puppy')],
            m('horse'),
        ])
        we2.addManyCallbacks('*', [
            m('mouse'),
            [m('dog'), call_forward(1), m('cat'), m('puppy')],
            m('horse'),
        ])

        we1.process(d1)
        we2.process(d2)

    def test_RUN_WF01(self):
        """Test wfe is reinit=False, eng must remember previous invocations"""
        we = GenericWorkflowEngine()
        doc = self.getDoc()[0:1]

        we.setWorkflow(
            [
                i('start'),
                ut.RUN_WF(
                    [
                        lambda obj, eng: obj.append('bom'),
                        lambda obj, eng: obj.append('bam'),
                        lambda obj, eng: obj.append('bum'),
                        lambda obj, eng: obj.append('end'),
                        lambda obj, eng: obj.append(
                            eng.store.setdefault('eng-end', '')),
                        e('eng-end', 'eng-end')
                    ],
                    data_connector=lambda obj, eng: [obj],
                    outkey='#wfe',
                ),
            ]
        )
        we.process(doc)

        d = ' '.join(doc[0])

        assert 'start' in d
        assert 'bom' in d
        assert 'bam' in d
        assert 'bum' in d
        assert 'end' in d
        assert 'eng-end' not in d

        # run the same thing again
        we.process(doc)

        d = ' '.join(doc[0])
        assert 'start' in d
        assert d.count('bom') == 2
        assert d.count('bam') == 2
        assert d.count('bum') == 2
        assert 'end' in d
        assert 'eng-end' in d  # now it must be present

    def test_configure(self):

        d1 = self.getDoc()
        d2 = self.getDoc()
        d3 = self.getDoc()

        we = GenericWorkflowEngine()
        we.addManyCallbacks('*', [
            m('mouse'),
            [m('dog'), call_forward(1), m('cat'), m('puppy')],
            m('horse'),
        ])

        # process using defaults
        we.process(d1)
        r = 'one mouse dog cat puppy horse'.split()

        # pass our own callback chooser
        we.configure(callback_chooser=asterisk_chooser)
        we.process(d2)

        assert d1[0] == r
        assert d2[0] == r
        assert d1 == d2

        # configure it wrongly
        we.configure(callback_chooser='')

        self.failUnlessRaises(Exception, we.process, d3)

        assert d3 == self.getDoc()

    def test_workflow01(self):
        we0 = GenericWorkflowEngine()
        we1 = GenericWorkflowEngine()
        we2 = GenericWorkflowEngine()

        d0 = self.getDoc()
        d1 = self.getDoc()
        d2 = self.getDoc()

        we0.addManyCallbacks('*', [
            m('mouse'),
            [m('dog'), call_forward(1), m('cat'), m('puppy')],
            m('horse'),
        ])
        we1.setWorkflow([
            m('mouse'),
            [m('dog'), call_forward(1), m('cat'), m('puppy')],
            m('horse'),
        ])
        we2.addManyCallbacks('x', [
            m('mouse'),
            [m('dog'), call_forward(1), m('cat'), m('puppy')],
            m('horse'),
        ])
        we2.configure(callback_chooser=lambda o, e: e.getCallbacks('x'))

        we0.process(d0)
        we1.process(d1)
        we2.process(d2)

        assert d0 == d1
        assert d0 == d2

 def setup_method(self, method):
     self.key = '*'
     self.wfe = GenericWorkflowEngine()
     self.data = ['one', 'two', 'three', 'four', 'five']
     self.tokens = [FakeToken(x, type='*') for x in self.data]

 def before_processing(objects, self):
     """Executed before processing the workflow."""
     self.save(status=WorkflowStatus.RUNNING)
     self.set_counter_initial(len(objects))
     workflow_started.send(self)
     GenericWorkflowEngine.before_processing(objects, self)

class TestWorkflowEngine(object):

    """Tests using FakeTokens in place of strings"""

    def setup_method(self, method):
        self.key = '*'
        self.wfe = GenericWorkflowEngine()
        self.data = ['one', 'two', 'three', 'four', 'five']
        self.tokens = [FakeToken(x, type='*') for x in self.data]

    def teardown_method(self, method):
        pass

    @pytest.mark.parametrize("_,tokens,exception,exception_msg", (
        ("int", 49, WorkflowError, "not an iterable"),
        ("str", "hello", WorkflowError, "not an iterable"),
        ("object", object, WorkflowError, "not an iterable"),
    ))
    def test_objects_are_of_bad_type(self, _, tokens, exception, exception_msg):
        with pytest.raises(exception) as exc_info:
            self.wfe.process(tokens)
        assert exception_msg in exc_info.value.args[0]

    def test_empty_object_list_logs_warning(self):
        assert hasattr(self.wfe, 'log')
        self.wfe.log = mock.Mock()
        self.wfe.callbacks.replace([lambda o, e: None])
        self.wfe.process([])
        self.wfe.log.warning.assert_called_once_with('List of objects is empty. Running workflow '
                                                     'on empty set has no effect.')

    def test_current_taskname_resolution(self):
        workflow = [m('test')]
        self.wfe.callbacks.replace(workflow, self.key)
        self.wfe.process(self.tokens)
        assert self.wfe.current_taskname == 'string appender'

        workflow = [lambda obj, eng: 1]
        self.wfe.callbacks.replace(workflow, self.key)
        self.wfe.process(self.tokens)
        assert self.wfe.current_taskname == '<lambda>'

        workflow = [
            IF_ELSE(
                lambda obj, eng: True,
                [lambda obj, eng: 1],
                [lambda obj, eng: 2],
            )
        ]
        self.wfe.callbacks.replace(workflow, self.key)
        # This test will break if someone changes IF_ELSE. TODO: Mock
        # Note: Python3 has much stronger introspection, thus the `.*`.
        assert re.match(r'\[<function IF_ELSE.* at 0x[0-f]+>, '
                        r'\[<function .*<lambda> at 0x[0-f]+>\], '
                        r'<function BREAK.* at 0x[0-f]+>, '
                        r'\[<function .*<lambda> at 0x[0-f]+>\]\]',
                        self.wfe.current_taskname)

    def test_current_object_returns_correct_object(self):
        self.wfe.callbacks.replace([halt_processing()])

        assert self.wfe.current_object is None
        with pytest.raises(HaltProcessing):
            self.wfe.process(self.tokens)
        assert self.wfe.current_object is self.tokens[0]
        with pytest.raises(HaltProcessing):
            self.wfe.restart('current', 'next')
        assert self.wfe.current_object is self.tokens[1]

    @pytest.mark.parametrize("_,callbacks,expected_result", (
        (
            'skips_forward_with_acceptable_increment',
            [
                m('mouse'),
                [m('dog'), jump_call(2), m('cat'), m('puppy'), m('python')],
                m('horse'),
            ],
            'mouse dog puppy python horse'
        ),

        (
            'skips_forward_with_increment_that_is_too_large',
            [
                m('mouse'),
                [m('dog'), jump_call(50), m('cat'), m('puppy'), m('python')],
                m('horse'),
            ],
            'mouse dog horse'
        ),

        (
            'jumps_forward_outside_of_nest',
            [
                jump_call(3),
                m('mouse'),
                [m('dog'), m('cat'), m('puppy'), m('python')],
                m('horse'),
            ],
            'horse'
        ),
#.........这里部分代码省略.........

def run_workflow(records, name, **kwargs):
    """Run the uploader workflow itself.

    :param records: List of tuples `(blob, json_record)` from :func:`translate`
    :param name: Name of the workflow to be run.
    :parma kwargs: Additional arguments to be used by the tasks of the workflow

    :returns: Typically the list of record Ids that has been process, although
        this value could be modify by the `post_tasks`.

    """
    def _run_pre_post_tasks(tasks):
        """Helper function to run list of functions."""
        for task in tasks:
            task(records, **kwargs)

    #FIXME: don't know why this is needed but IT IS!
    records = records[0]

    if name in cfg['UPLOADER_WORKFLOWS']:
        workflow = workflows.get(name)
    else:
        raise UploaderException("Workflow {0} not in UPLOADER_WORKFLOWS".format(name))
    _run_pre_post_tasks(workflow.pre_tasks)
    wfe = WorkflowEngine()
    wfe.setWorkflow(workflow.tasks)
    wfe.setVar('options', kwargs)
    wfe.process(records)
    _run_pre_post_tasks(workflow.post_tasks)
    signals.uploader_finished.send(uploader_workflow=name,
                                   result=records, **kwargs)
    return records