def test_posting_with_book(self):
        backend = impl_dir.DirBackend(conf={
            'path': self.makeTmpDir(),
        })
        backend.get_connection().upgrade()
        book, flow_detail = p_utils.temporary_flow_detail(backend)
        self.assertEqual(1, len(book))

        client, board = self.create_board(persistence=backend)
        with connect_close(board):
            with self.flush(client):
                board.post('test', book)

            possible_jobs = list(board.iterjobs(only_unclaimed=True))
            self.assertEqual(1, len(possible_jobs))
            j = possible_jobs[0]
            self.assertEqual(1, len(j.book))
            self.assertEqual(book.name, j.book.name)
            self.assertEqual(book.uuid, j.book.uuid)
            self.assertEqual(book.name, j.book_name)
            self.assertEqual(book.uuid, j.book_uuid)

            flow_details = list(j.book)
            self.assertEqual(flow_detail.uuid, flow_details[0].uuid)
            self.assertEqual(flow_detail.name, flow_details[0].name)

 def test_no_meta(self):
     _lb, flow_detail = p_utils.temporary_flow_detail()
     self.assertEqual({}, flow_detail.meta)
     self.assertRaisesRegexp(ValueError,
                             '^Cannot .* no factory information saved.$',
                             taskflow.engines.flow_from_detail,
                             flow_detail)

    def test_fail_run(self):
        components = self.make_components()
        components.conductor.connect()
        consumed_event = threading_utils.Event()

        def on_consume(state, details):
            consumed_event.set()

        components.board.notifier.register(jobboard.REMOVAL, on_consume)
        with close_many(components.conductor, components.client):
            t = threading_utils.daemon_thread(components.conductor.run)
            t.start()
            lb, fd = pu.temporary_flow_detail(components.persistence)
            engines.save_factory_details(fd, test_factory,
                                         [True], {},
                                         backend=components.persistence)
            components.board.post('poke', lb,
                                  details={'flow_uuid': fd.uuid})
            self.assertTrue(consumed_event.wait(test_utils.WAIT_TIMEOUT))
            self.assertTrue(components.conductor.stop(test_utils.WAIT_TIMEOUT))
            self.assertFalse(components.conductor.dispatching)

        persistence = components.persistence
        with contextlib.closing(persistence.get_connection()) as conn:
            lb = conn.get_logbook(lb.uuid)
            fd = lb.find(fd.uuid)
        self.assertIsNotNone(fd)
        self.assertEqual(st.REVERTED, fd.state)

    def test_sequential_flow_iter_suspend_resume(self):
        flow = lf.Flow('flow-2').add(
            utils.ProgressingTask(name='task1'),
            utils.ProgressingTask(name='task2')
        )
        lb, fd = p_utils.temporary_flow_detail(self.backend)

        engine = self._make_engine(flow, flow_detail=fd)
        with utils.CaptureListener(engine, capture_flow=False) as capturer:
            it = engine.run_iter()
            gathered_states = []
            suspend_it = None
            while True:
                try:
                    s = it.send(suspend_it)
                    gathered_states.append(s)
                    if s == states.WAITING:
                        # Stop it before task2 runs/starts.
                        suspend_it = True
                except StopIteration:
                    break
        self.assertTrue(len(gathered_states) > 0)
        expected = ['task1.t RUNNING', 'task1.t SUCCESS(5)']
        self.assertEqual(expected, capturer.values)
        self.assertEqual(states.SUSPENDED, engine.storage.get_flow_state())

        # Attempt to resume it and see what runs now...
        with utils.CaptureListener(engine, capture_flow=False) as capturer:
            gathered_states = list(engine.run_iter())
        self.assertTrue(len(gathered_states) > 0)
        expected = ['task2.t RUNNING', 'task2.t SUCCESS(5)']
        self.assertEqual(expected, capturer.values)
        self.assertEqual(states.SUCCESS, engine.storage.get_flow_state())

 def _get_storage(self, flow_detail=None, threaded=False):
     if flow_detail is None:
         _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
     storage_cls = storage.SingleThreadedStorage
     if threaded:
         storage_cls = storage.MultiThreadedStorage
     return storage_cls(flow_detail=flow_detail, backend=self.backend)

    def test_run_max_dispatches(self):
        components = self.make_components()
        components.conductor.connect()
        consumed_event = threading.Event()

        def on_consume(state, details):
            consumed_event.set()

        components.board.notifier.register(base.REMOVAL, on_consume)
        with close_many(components.client, components.conductor):
            t = threading_utils.daemon_thread(
                lambda: components.conductor.run(max_dispatches=5))
            t.start()
            lb, fd = pu.temporary_flow_detail(components.persistence)
            engines.save_factory_details(fd, test_factory,
                                         [False], {},
                                         backend=components.persistence)
            for _ in range(5):
                components.board.post('poke', lb,
                                      details={'flow_uuid': fd.uuid})
                self.assertTrue(consumed_event.wait(
                    test_utils.WAIT_TIMEOUT))
            components.board.post('poke', lb,
                                  details={'flow_uuid': fd.uuid})
            components.conductor.stop()
            self.assertTrue(components.conductor.wait(test_utils.WAIT_TIMEOUT))
            self.assertFalse(components.conductor.dispatching)

    def test_get_without_save(self):
        _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
        td = logbook.TaskDetail(name='my_task', uuid='42')
        flow_detail.add(td)

        s = storage.Storage(backend=self.backend, flow_detail=flow_detail)
        self.assertEquals('42', s.get_task_uuid('my_task'))

    def test_combined_store(self):
        components = self.make_components()
        components.conductor.connect()
        consumed_event = threading.Event()

        def on_consume(state, details):
            consumed_event.set()

        flow_store = {'x': True, 'y': False}
        job_store = {'z': None}

        components.board.notifier.register(base.REMOVAL, on_consume)
        with close_many(components.conductor, components.client):
            t = threading_utils.daemon_thread(components.conductor.run)
            t.start()
            lb, fd = pu.temporary_flow_detail(components.persistence,
                                              meta={'store': flow_store})
            engines.save_factory_details(fd, test_store_factory,
                                         [], {},
                                         backend=components.persistence)
            components.board.post('poke', lb,
                                  details={'flow_uuid': fd.uuid,
                                           'store': job_store})
            self.assertTrue(consumed_event.wait(test_utils.WAIT_TIMEOUT))
            components.conductor.stop()
            self.assertTrue(components.conductor.wait(test_utils.WAIT_TIMEOUT))
            self.assertFalse(components.conductor.dispatching)

        persistence = components.persistence
        with contextlib.closing(persistence.get_connection()) as conn:
            lb = conn.get_logbook(lb.uuid)
            fd = lb.find(fd.uuid)
        self.assertIsNotNone(fd)
        self.assertEqual(st.SUCCESS, fd.state)

 def test_ensure_existing_task(self):
     _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
     td = logbook.TaskDetail(name="my_task", uuid="42")
     flow_detail.add(td)
     s = self._get_storage(flow_detail)
     s.ensure_atom(test_utils.NoopTask("my_task"))
     self.assertEqual("42", s.get_atom_uuid("my_task"))

 def test_ensure_existing_task(self):
     _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
     td = models.TaskDetail(name='my_task', uuid='42')
     flow_detail.add(td)
     s = self._get_storage(flow_detail)
     s.ensure_atom(test_utils.NoopTask('my_task'))
     self.assertEqual('42', s.get_atom_uuid('my_task'))

 def test_get_flow_state(self):
     _lb, fd = p_utils.temporary_flow_detail(backend=self.backend)
     fd.state = states.FAILURE
     with contextlib.closing(self.backend.get_connection()) as conn:
         fd.update(conn.update_flow_details(fd))
     s = storage.Storage(flow_detail=fd, backend=self.backend)
     self.assertEquals(s.get_flow_state(), states.FAILURE)

    def test_sequential_flow_iter_suspend_resume(self):
        flow = lf.Flow('flow-2').add(
            utils.SaveOrderTask(name='task1'),
            utils.SaveOrderTask(name='task2')
        )
        _lb, fd = p_utils.temporary_flow_detail(self.backend)
        e = self._make_engine(flow, flow_detail=fd)
        it = e.run_iter()
        gathered_states = []
        suspend_it = None
        while True:
            try:
                s = it.send(suspend_it)
                gathered_states.append(s)
                if s == states.WAITING:
                    # Stop it before task2 runs/starts.
                    suspend_it = True
            except StopIteration:
                break
        self.assertTrue(len(gathered_states) > 0)
        self.assertEqual(self.values, ['task1'])
        self.assertEqual(states.SUSPENDED, e.storage.get_flow_state())

        # Attempt to resume it and see what runs now...
        #
        # NOTE(harlowja): Clear all the values, but don't reset the reference.
        while len(self.values):
            self.values.pop()
        gathered_states = list(e.run_iter())
        self.assertTrue(len(gathered_states) > 0)
        self.assertEqual(self.values, ['task2'])
        self.assertEqual(states.SUCCESS, e.storage.get_flow_state())

 def _get_storage(self, threaded=False):
     _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
     if threaded:
         return storage.MultiThreadedStorage(backend=self.backend,
                                             flow_detail=flow_detail)
     else:
         return storage.SingleThreadedStorage(backend=self.backend,
                                              flow_detail=flow_detail)

    def test_ensure_existing_task(self):
        _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
        td = logbook.TaskDetail(name='my_task', uuid='42')
        flow_detail.add(td)

        s = storage.Storage(backend=self.backend, flow_detail=flow_detail)
        s.ensure_task('my_task')
        self.assertEqual('42', s.get_task_uuid('my_task'))

 def test_no_importable_function(self):
     _lb, flow_detail = p_utils.temporary_flow_detail()
     flow_detail.meta = dict(factory=dict(
         name='you can not import me, i contain spaces'
     ))
     expected_msg = '^Could not import factory'
     with self.assertRaisesRegexp(ImportError, expected_msg):
         taskflow.engines.flow_from_detail(flow_detail)

 def test_ensure_task_fd(self):
     _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
     s = storage.Storage(backend=self.backend, flow_detail=flow_detail)
     s.ensure_task('my task', '3.11')
     td = flow_detail.find(s.get_task_uuid('my task'))
     self.assertIsNotNone(td)
     self.assertEqual(td.name, 'my task')
     self.assertEqual(td.version, '3.11')
     self.assertEqual(td.state, states.PENDING)

 def test_ensure_task_flow_detail(self):
     _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
     s = self._get_storage(flow_detail)
     s.ensure_task("my task", "3.11")
     td = flow_detail.find(s.get_task_uuid("my task"))
     self.assertIsNotNone(td)
     self.assertEqual(td.name, "my task")
     self.assertEqual(td.version, "3.11")
     self.assertEqual(td.state, states.PENDING)

    def test_transient_storage_restore(self):
        _lb, flow_detail = p_utils.temporary_flow_detail(self.backend)
        s = self._get_storage(flow_detail=flow_detail)
        s.inject([("a", "b")], transient=True)
        s.inject([("b", "c")])

        s2 = self._get_storage(flow_detail=flow_detail)
        results = s2.fetch_all()
        self.assertEqual({"b": "c"}, results)

    def test_no_arg_factory(self):
        name = 'some.test.factory'
        _lb, flow_detail = p_utils.temporary_flow_detail()
        flow_detail.meta = dict(factory=dict(name=name))

        with mock.patch('oslo_utils.importutils.import_class',
                        return_value=lambda: 'RESULT') as mock_import:
            result = taskflow.engines.flow_from_detail(flow_detail)
            mock_import.assert_called_once_with(name)
        self.assertEqual('RESULT', result)

 def test_deregister(self):
     """Verify that register and deregister don't blow up"""
     with contextlib.closing(impl_memory.MemoryBackend()) as be:
         flow = lf.Flow("test")
         flow.add(SleepyTask("test-1", sleep_for=0.1))
         (lb, fd) = persistence_utils.temporary_flow_detail(be)
         e = self._make_engine(flow, fd, be)
         l = timing.DurationListener(e)
         l.register()
         l.deregister()