def test_task_detail_meta_update(self):
        lb_id = uuidutils.generate_uuid()
        lb_name = 'lb-%s' % (lb_id)
        lb = logbook.LogBook(name=lb_name, uuid=lb_id)
        fd = logbook.FlowDetail('test', uuid=uuidutils.generate_uuid())
        lb.add(fd)
        td = logbook.TaskDetail("detail-1", uuid=uuidutils.generate_uuid())
        td.meta = {'test': 42}
        fd.add(td)

        with contextlib.closing(self._get_connection()) as conn:
            conn.save_logbook(lb)
            conn.update_flow_details(fd)
            conn.update_atom_details(td)

        td.meta['test'] = 43
        with contextlib.closing(self._get_connection()) as conn:
            conn.update_atom_details(td)

        with contextlib.closing(self._get_connection()) as conn:
            lb2 = conn.get_logbook(lb_id)
        fd2 = lb2.find(fd.uuid)
        td2 = fd2.find(td.uuid)
        self.assertEqual(td2.meta.get('test'), 43)
        self.assertIsInstance(td2, logbook.TaskDetail)

def create_flow_detail(flow, book=None, backend=None):
    """Creates a flow detail for the given flow and adds it to the provided
    logbook (if provided) and then uses the given backend (if provided) to
    save the logbook then returns the created flow detail.
    """
    try:
        flow_name = getattr(flow, 'name')
    except AttributeError:
        LOG.warn("Flow %s does not have a name attribute, creating one.", flow)
        flow_name = uuidutils.generate_uuid()
    try:
        flow_id = getattr(flow, 'uuid')
    except AttributeError:
        LOG.warn("Flow %s does not have a uuid attribute, creating one.", flow)
        flow_id = uuidutils.generate_uuid()
    flow_detail = logbook.FlowDetail(name=flow_name, uuid=flow_id)
    if book is not None:
        book.add(flow_detail)
        if backend is not None:
            with contextlib.closing(backend.get_connection()) as conn:
                conn.save_logbook(book)
        # Return the one from the saved logbook instead of the local one so
        # that the freshest version is given back
        return book.find(flow_id)
    else:
        if backend is not None:
            LOG.warn("Can not save %s without a provided logbook", flow)
        return flow_detail

    def test_retry_detail_save_with_task_failure(self):
        lb_id = uuidutils.generate_uuid()
        lb_name = 'lb-%s' % (lb_id)
        lb = logbook.LogBook(name=lb_name, uuid=lb_id)
        fd = logbook.FlowDetail('test', uuid=uuidutils.generate_uuid())
        lb.add(fd)
        rd = logbook.RetryDetail("retry-1", uuid=uuidutils.generate_uuid())
        fail = misc.Failure.from_exception(RuntimeError('fail'))
        rd.results.append((42, {'some-task': fail}))
        fd.add(rd)

        # save it
        with contextlib.closing(self._get_connection()) as conn:
            conn.save_logbook(lb)
            conn.update_flow_details(fd)
            conn.update_atom_details(rd)

        # now read it back
        with contextlib.closing(self._get_connection()) as conn:
            lb2 = conn.get_logbook(lb_id)
        fd2 = lb2.find(fd.uuid)
        rd2 = fd2.find(rd.uuid)
        self.assertIsInstance(rd2, logbook.RetryDetail)
        fail2 = rd2.results[0][1].get('some-task')
        self.assertIsInstance(fail2, misc.Failure)
        self.assertTrue(fail.matches(fail2))

    def test_retry_detail_save_intention(self):
        lb_id = uuidutils.generate_uuid()
        lb_name = 'lb-%s' % (lb_id)
        lb = logbook.LogBook(name=lb_name, uuid=lb_id)
        fd = logbook.FlowDetail('test', uuid=uuidutils.generate_uuid())
        lb.add(fd)
        rd = logbook.RetryDetail("retry-1", uuid=uuidutils.generate_uuid())
        fd.add(rd)

        # save it
        with contextlib.closing(self._get_connection()) as conn:
            conn.save_logbook(lb)
            conn.update_flow_details(fd)
            conn.update_atom_details(rd)

        # change intention and save
        rd.intention = states.REVERT
        with contextlib.closing(self._get_connection()) as conn:
            conn.update_atom_details(rd)

        # now read it back
        with contextlib.closing(self._get_connection()) as conn:
            lb2 = conn.get_logbook(lb_id)
        fd2 = lb2.find(fd.uuid)
        rd2 = fd2.find(rd.uuid)
        self.assertEqual(rd2.intention, states.REVERT)
        self.assertIsInstance(rd2, logbook.RetryDetail)

    def test_task_detail_with_failure(self):
        lb_id = uuidutils.generate_uuid()
        lb_name = 'lb-%s' % (lb_id)
        lb = logbook.LogBook(name=lb_name, uuid=lb_id)
        fd = logbook.FlowDetail('test', uuid=uuidutils.generate_uuid())
        lb.add(fd)
        td = logbook.TaskDetail("detail-1", uuid=uuidutils.generate_uuid())

        try:
            raise RuntimeError('Woot!')
        except Exception:
            td.failure = misc.Failure()

        fd.add(td)

        with contextlib.closing(self._get_connection()) as conn:
            conn.save_logbook(lb)
            conn.update_flow_details(fd)
            conn.update_task_details(td)

        # Read failure back
        with contextlib.closing(self._get_connection()) as conn:
            lb2 = conn.get_logbook(lb_id)
        fd2 = lb2.find(fd.uuid)
        td2 = fd2.find(td.uuid)
        failure = td2.failure
        self.assertEqual(failure.exception_str, 'Woot!')
        self.assertIs(failure.check(RuntimeError), RuntimeError)
        self.assertEqual(failure.traceback_str, td.failure.traceback_str)

 def test_logbook_add_task_detail(self):
     lb_id = uuidutils.generate_uuid()
     lb_name = 'lb-%s' % (lb_id)
     lb = logbook.LogBook(name=lb_name, uuid=lb_id)
     fd = logbook.FlowDetail('test', uuid=uuidutils.generate_uuid())
     td = logbook.TaskDetail("detail-1", uuid=uuidutils.generate_uuid())
     td.version = '4.2'
     fd.add(td)
     lb.add(fd)
     with contextlib.closing(self._get_connection()) as conn:
         conn.save_logbook(lb)
     with contextlib.closing(self._get_connection()) as conn:
         lb2 = conn.get_logbook(lb_id)
         self.assertEqual(1, len(lb2))
         tasks = 0
         for fd in lb:
             tasks += len(fd)
         self.assertEqual(1, tasks)
     with contextlib.closing(self._get_connection()) as conn:
         lb2 = conn.get_logbook(lb_id)
         fd2 = lb2.find(fd.uuid)
         td2 = fd2.find(td.uuid)
         self.assertIsNot(td2, None)
         self.assertEqual(td2.name, 'detail-1')
         self.assertEqual(td2.version, '4.2')

 def test_logbook_add_flow_detail(self):
     lb_id = uuidutils.generate_uuid()
     lb_name = "lb-%s" % (lb_id)
     lb = logbook.LogBook(name=lb_name, uuid=lb_id)
     fd = logbook.FlowDetail("test", uuid=uuidutils.generate_uuid())
     lb.add(fd)
     with contextlib.closing(self._get_connection()) as conn:
         conn.save_logbook(lb)
     with contextlib.closing(self._get_connection()) as conn:
         lb2 = conn.get_logbook(lb_id)
         self.assertEquals(1, len(lb2))
         self.assertEquals(1, len(lb))
         self.assertEquals(fd.name, lb2.find(fd.uuid).name)

    def setUpClass(cls):
        # Create a workflow for flowdetails to use
        wf_id = uuidutils.generate_uuid()
        wf_name = 'wf-%s' % (wf_id)

        wf = flow.Flow(wf_name, None, wf_id)
        cls.wfs.append(wf)

        # Create a task for taskdetails to use
        task_id = uuidutils.generate_uuid()
        task_name = 'task-%s' % (task_id)

        tsk = utils.DummyTask(task_name, task_id)
        cls.tsks.append(tsk)

    def test_logbook_add_flow_detail(self):
        lb_id = uuidutils.generate_uuid()
        lb_name = 'lb-%s' % (lb_id)
        lb = logbook.LogBook(name=lb_name, uuid=lb_id,
                             backend=self._get_backend())

        fd = flowdetail.FlowDetail('test', uuid=uuidutils.generate_uuid())
        lb.add(fd)
        lb.save()

        lb2 = logbook.load(lb_id, backend=self._get_backend())
        self.assertEquals(1, len(lb2))
        self.assertEquals(1, len(lb))

        self.assertEquals(fd.name, lb2.find(fd.uuid).name)

    def _ensure_task(self, task_name, task_version, result_mapping):
        """Ensures there is a taskdetail that corresponds to the task info.

        If task does not exist, adds a record for it. Added task will have
        PENDING state. Sets result mapping for the task from result_mapping
        argument.

        Returns uuid for the task details corresponding to the task with
        given name.
        """
        if not task_name:
            raise ValueError("Task name must be non-empty")
        with self._lock.write_lock():
            try:
                task_id = self._atom_name_to_uuid[task_name]
            except KeyError:
                task_id = uuidutils.generate_uuid()
                self._create_atom_detail(logbook.TaskDetail, task_name,
                                         task_id, task_version)
            else:
                ad = self._flowdetail.find(task_id)
                if not isinstance(ad, logbook.TaskDetail):
                    raise exceptions.Duplicate(
                        "Atom detail %s already exists in flow detail %s." %
                        (task_name, self._flowdetail.name))
            self._set_result_mapping(task_name, result_mapping)
        return task_id

    def _ensure_retry(self, retry_name, retry_version, result_mapping):
        """Ensures there is a retrydetail that corresponds to the retry info.

        If retry does not exist, adds a record for it. Added retry
        will have PENDING state. Sets result mapping for the retry from
        result_mapping argument. Initializes retry result as an empty
        collections of results and failures history.

        Returns uuid for the retry details corresponding to the retry
        with given name.
        """
        if not retry_name:
            raise ValueError("Retry name must be non-empty")
        with self._lock.write_lock():
            try:
                retry_id = self._atom_name_to_uuid[retry_name]
            except KeyError:
                retry_id = uuidutils.generate_uuid()
                self._create_atom_detail(logbook.RetryDetail, retry_name,
                                         retry_id, retry_version)
            else:
                ad = self._flowdetail.find(retry_id)
                if not isinstance(ad, logbook.RetryDetail):
                    raise exceptions.Duplicate(
                        "Atom detail %s already exists in flow detail %s." %
                        (retry_name, self._flowdetail.name))
            self._set_result_mapping(retry_name, result_mapping)
        return retry_id

    def setUpClass(cls):
        # Create a workflow to create flowdetails with
        wf_id = uuidutils.generate_uuid()
        wf_name = 'wf-%s' % (wf_id)

        wf = flow.Flow(wf_name, None, wf_id)
        cls.wfs.append(wf)

    def setUp(self):
        # Create a logbook and record its uuid and name
        lb_id = uuidutils.generate_uuid()
        lb_name = 'lb-%s' % (lb_id)

        b_api.logbook_create(lb_name, lb_id)
        self.lb_names.append(lb_name)
        self.lb_ids.append(lb_id)

        # Create a flowdetail and record its uuid and name
        fd_id = uuidutils.generate_uuid()
        fd_name = 'fd-%s' % (fd_id)

        b_api.flowdetail_create(fd_name, self.wfs[0], fd_id)
        self.fd_names.append(fd_name)
        self.fd_ids.append(fd_id)

    def compile(self):
        """Compiles the contained flow into a structure which the engine can
        use to run or if this can not be done then an exception is thrown
        indicating why this compilation could not be achieved.
        """
        if self._root is not None:
            return

        assert self._graph_action is not None, ('Graph action class must be'
                                                ' specified')
        self._change_state(states.RESUMING)  # does nothing in PENDING state
        task_graph = flow_utils.flatten(self._flow)
        if task_graph.number_of_nodes() == 0:
            raise exc.EmptyFlow("Flow %s is empty." % self._flow.name)
        self._root = self._graph_action(task_graph)
        for task in task_graph.nodes_iter():
            try:
                task_id = self.storage.get_uuid_by_name(task.name)
            except exc.NotFound:
                task_id = uuidutils.generate_uuid()
                task_version = misc.get_version_string(task)
                self.storage.add_task(task_name=task.name, uuid=task_id,
                                      task_version=task_version)

            self.storage.set_result_mapping(task_id, task.save_as)
            self._root.add(task, task_action.TaskAction(task, task_id))
        self._change_state(states.SUSPENDED)  # does nothing in PENDING state

    def test_flow_detail_save(self):
        lb_id = uuidutils.generate_uuid()
        lb_name = 'lb-%s' % (lb_id)
        lb = logbook.LogBook(name=lb_name, uuid=lb_id,
                             backend=self._get_backend())

        fd = flowdetail.FlowDetail('test', uuid=uuidutils.generate_uuid())
        lb.add(fd)

        # Ensure we can't save it since its owning logbook hasn't been
        # saved.
        self.assertRaises(exc.NotFound, fd.save)

        # Ok now we should be able to save it
        lb.save()
        fd.save()

 def __init__(self, name, parents=None, uuid=None):
     self._name = str(name)
     # The state of this flow.
     self._state = states.PENDING
     # If this flow has a parent flow/s which need to be reverted if
     # this flow fails then please include them here to allow this child
     # to call the parents...
     if parents:
         self.parents = tuple(parents)
     else:
         self.parents = tuple([])
     # Any objects that want to listen when a wf/task starts/stops/completes
     # or errors should be registered here. This can be used to monitor
     # progress and record tasks finishing (so that it becomes possible to
     # store the result of a task in some persistent or semi-persistent
     # storage backend).
     self.notifier = utils.TransitionNotifier()
     self.task_notifier = utils.TransitionNotifier()
     # Ensure that modifications and/or multiple runs aren't happening
     # at the same time in the same flow at the same time.
     self._lock = threading.RLock()
     # Assign this flow a unique identifer.
     if uuid:
         self._id = str(uuid)
     else:
         self._id = uuidutils.generate_uuid()

def flush(client, path=None):
    # This uses the linearity guarantee of zookeeper (and associated libraries)
    # to create a temporary node, wait until a watcher notifies it's created,
    # then yield back for more work, and then at the end of that work delete
    # the created node. This ensures that the operations done in the yield
    # of this context manager will be applied and all watchers will have fired
    # before this context manager exits.
    if not path:
        path = FLUSH_PATH_TPL % uuidutils.generate_uuid()
    created = threading.Event()
    deleted = threading.Event()

    def on_created(data, stat):
        if stat is not None:
            created.set()
            return False  # cause this watcher to cease to exist

    def on_deleted(data, stat):
        if stat is None:
            deleted.set()
            return False  # cause this watcher to cease to exist

    watchers.DataWatch(client, path, func=on_created)
    client.create(path, makepath=True)
    created.wait()
    try:
        yield
    finally:
        watchers.DataWatch(client, path, func=on_deleted)
        client.delete(path, recursive=True)
        deleted.wait()

    def test_multi_message(self):
        message_count = 30
        barrier = latch.Latch(message_count)
        countdown = lambda data, message: barrier.countdown()

        on_notify = mock.MagicMock()
        on_notify.side_effect = countdown

        on_response = mock.MagicMock()
        on_response.side_effect = countdown

        on_request = mock.MagicMock()
        on_request.side_effect = countdown

        handlers = {
            pr.NOTIFY: on_notify,
            pr.RESPONSE: on_response,
            pr.REQUEST: on_request,
        }
        p = proxy.Proxy(TEST_TOPIC, TEST_EXCHANGE, handlers,
                        transport='memory',
                        transport_options={
                            'polling_interval': POLLING_INTERVAL,
                        })

        t = threading.Thread(target=p.start)
        t.daemon = True
        t.start()
        p.wait()

        for i in range(0, message_count):
            j = i % 3
            if j == 0:
                p.publish(pr.Notify(), TEST_TOPIC)
            elif j == 1:
                p.publish(pr.Response(pr.RUNNING), TEST_TOPIC)
            else:
                p.publish(pr.Request(test_utils.DummyTask("dummy_%s" % i),
                                     uuidutils.generate_uuid(),
                                     pr.EXECUTE, [], None, None), TEST_TOPIC)

        self.assertTrue(barrier.wait(test_utils.WAIT_TIMEOUT))
        self.assertEqual(0, barrier.needed)
        p.stop()
        t.join()

        self.assertTrue(on_notify.called)
        self.assertTrue(on_response.called)
        self.assertTrue(on_request.called)

        self.assertEqual(10, on_notify.call_count)
        self.assertEqual(10, on_response.call_count)
        self.assertEqual(10, on_request.call_count)

        call_count = sum([
            on_notify.call_count,
            on_response.call_count,
            on_request.call_count,
        ])
        self.assertEqual(message_count, call_count)

 def test_logbook_merge_flow_detail(self):
     lb_id = uuidutils.generate_uuid()
     lb_name = "lb-%s" % (lb_id)
     lb = logbook.LogBook(name=lb_name, uuid=lb_id)
     fd = logbook.FlowDetail("test", uuid=uuidutils.generate_uuid())
     lb.add(fd)
     with contextlib.closing(self._get_connection()) as conn:
         conn.save_logbook(lb)
     lb2 = logbook.LogBook(name=lb_name, uuid=lb_id)
     fd2 = logbook.FlowDetail("test2", uuid=uuidutils.generate_uuid())
     lb2.add(fd2)
     with contextlib.closing(self._get_connection()) as conn:
         conn.save_logbook(lb2)
     with contextlib.closing(self._get_connection()) as conn:
         lb3 = conn.get_logbook(lb_id)
         self.assertEquals(2, len(lb3))

    def setUp(self):
        # Create a flowdetail and record its uuid and name
        fd_id = uuidutils.generate_uuid()
        fd_name = 'fd-%s' % (fd_id)

        b_api.flowdetail_create(fd_name, self.wfs[0], fd_id)
        self.fd_names.append(fd_name)
        self.fd_ids.append(fd_id)

        # Create a taskdetail and record its uuid and name
        td_id = uuidutils.generate_uuid()
        td_name = 'td-%s' % (td_id)

        b_api.taskdetail_create(td_name, self.tsks[0], td_id)
        self.td_names.append(td_name)
        self.td_ids.append(td_id)