def test_flatten_checks_for_dups_globally(self):
     flo = gf.Flow("test").add(
         gf.Flow("int1").add(t_utils.DummyTask(name="a")),
         gf.Flow("int2").add(t_utils.DummyTask(name="a")))
     with self.assertRaisesRegexp(exc.InvariantViolationException,
                                  '^Tasks with duplicate names'):
         f_utils.flatten(flo)

    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 _translate_flow_to_action(self):
     # Flatten the flow into just 1 graph.
     task_graph = flow_utils.flatten(self._flow)
     ga = graph_action.SequentialGraphAction(task_graph)
     for n in task_graph.nodes_iter():
         ga.add(n, task_action.TaskAction(n, self))
     return ga

    def test_graph_flatten(self):
        a, b, c, d = _make_many(4)
        flo = gf.Flow("test")
        flo.add(a, b, c, d)

        g = f_utils.flatten(flo)
        self.assertEquals(4, len(g))
        self.assertEquals(0, g.number_of_edges())

 def _translate_flow_to_action(self):
     assert self._graph_action is not None, ('Graph action class must be'
                                             ' specified')
     task_graph = flow_utils.flatten(self._flow)
     ga = self._graph_action(task_graph)
     for n in task_graph.nodes_iter():
         ga.add(n, task_action.TaskAction(n, self))
     return ga

 def test_targeted_flow_one_node(self):
     wf = gw.TargetedFlow("test")
     test_1 = utils.ProvidesRequiresTask('test-1',
                                         provides=['a'], requires=[])
     wf.add(test_1)
     wf.set_target(test_1)
     g = fu.flatten(wf)
     self.assertEqual(1, len(g))
     self.assertTrue(g.has_node(test_1))

 def test_unordered_flatten(self):
     a, b, c, d = _make_many(4)
     flo = uf.Flow("test")
     flo.add(a, b, c, d)
     g = f_utils.flatten(flo)
     self.assertEquals(4, len(g))
     self.assertEquals(0, g.number_of_edges())
     self.assertEquals(set([a, b, c, d]),
                       set(g_utils.get_no_successors(g)))
     self.assertEquals(set([a, b, c, d]),
                       set(g_utils.get_no_predecessors(g)))

    def test_graph_flatten_nested_graph(self):
        a, b, c, d, e, f, g = _make_many(7)
        flo = gf.Flow("test")
        flo.add(a, b, c, d)

        flo2 = gf.Flow('test2')
        flo2.add(e, f, g)
        flo.add(flo2)

        g = f_utils.flatten(flo)
        self.assertEquals(7, len(g))
        self.assertEquals(0, g.number_of_edges())

    def test_graph_flatten_links(self):
        a, b, c, d = _make_many(4)
        flo = gf.Flow("test")
        flo.add(a, b, c, d)
        flo.link(a, b)
        flo.link(b, c)
        flo.link(c, d)

        g = f_utils.flatten(flo)
        self.assertEquals(4, len(g))
        self.assertEquals(3, g.number_of_edges())
        self.assertEquals(set([a]),
                          set(g_utils.get_no_predecessors(g)))
        self.assertEquals(set([d]),
                          set(g_utils.get_no_successors(g)))

 def test_flatten_attribute(self):
     wf = gw.Flow("the-test-action")
     test_1 = utils.ProvidesRequiresTask('test-1',
                                         requires=[],
                                         provides=[])
     test_2 = utils.ProvidesRequiresTask('test-2',
                                         provides=[],
                                         requires=[])
     wf.add(test_1, test_2)
     wf.link(test_1, test_2)
     g = fu.flatten(wf)
     self.assertEqual(2, len(g))
     edge_attrs = gu.get_edge_attrs(g, test_1, test_2)
     self.assertTrue(edge_attrs.get('manual'))
     self.assertTrue(edge_attrs.get('flatten'))

    def test_linear_flatten(self):
        a, b, c, d = _make_many(4)
        flo = lf.Flow("test")
        flo.add(a, b, c)
        sflo = lf.Flow("sub-test")
        sflo.add(d)
        flo.add(sflo)

        g = f_utils.flatten(flo)
        self.assertEquals(4, len(g))

        order = nx.topological_sort(g)
        self.assertEquals([a, b, c, d], order)
        self.assertTrue(g.has_edge(c, d))
        self.assertEquals([d], list(g_utils.get_no_successors(g)))
        self.assertEquals([a], list(g_utils.get_no_predecessors(g)))

 def test_targeted_flow_reset(self):
     wf = gw.TargetedFlow("test")
     test_1 = utils.ProvidesRequiresTask('test-1',
                                         provides=['a'], requires=[])
     test_2 = utils.ProvidesRequiresTask('test-2',
                                         provides=['b'], requires=['a'])
     test_3 = utils.ProvidesRequiresTask('test-3',
                                         provides=[], requires=['b'])
     test_4 = utils.ProvidesRequiresTask('test-4',
                                         provides=['c'], requires=['b'])
     wf.add(test_1, test_2, test_3, test_4)
     wf.set_target(test_3)
     wf.reset_target()
     g = fu.flatten(wf)
     self.assertEqual(4, len(g))
     self.assertTrue(g.has_node(test_4))

 def compile(self):
     if self._compiled:
         return
     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._analyzer = self._graph_analyzer_cls(task_graph, self.storage)
     if self._task_executor is None:
         self._task_executor = self._task_executor_cls()
     if self._task_action is None:
         self._task_action = self._task_action_cls(self.storage, self._task_executor, self.task_notifier)
     self._root = self._graph_action_cls(self._analyzer, self.storage, self._task_action)
     # NOTE(harlowja): Perform initial state manipulation and setup.
     #
     # TODO(harlowja): This doesn't seem like it should be in a compilation
     # function since compilation seems like it should not modify any
     # external state.
     self._ensure_storage_for(task_graph)
     self._compiled = True

    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)
        self._root = self._graph_action(task_graph)
        loaded_failures = {}

        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)
            try:
                result = self.storage.get(task_id)
            except exc.NotFound:
                result = None

            if isinstance(result, misc.Failure):
                # NOTE(imelnikov): old failure may have exc_info which
                # might get lost during serialization, so we preserve
                # old failure object if possible.
                old_failure = self._failures.get(task_id, None)
                if result.matches(old_failure):
                    loaded_failures[task_id] = old_failure
                else:
                    loaded_failures[task_id] = result

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

    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_cls 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_cls(task_graph)
        for task in task_graph.nodes_iter():
            task_version = misc.get_version_string(task)
            self.storage.ensure_task(task.name, task_version, task.save_as)

        self._change_state(states.SUSPENDED)  # does nothing in PENDING state

    def test_unordered_nested_flatten(self):
        a, b, c, d = _make_many(4)
        flo = uf.Flow("test")
        flo.add(a, b)
        flo2 = lf.Flow("test2")
        flo2.add(c, d)
        flo.add(flo2)

        g = f_utils.flatten(flo)
        self.assertEquals(4, len(g))
        for n in [a, b]:
            self.assertFalse(g.has_edge(n, c))
            self.assertFalse(g.has_edge(n, d))
        self.assertTrue(g.has_edge(c, d))
        self.assertFalse(g.has_edge(d, c))

        ub = g.subgraph([a, b])
        self.assertEquals(0, ub.number_of_edges())
        lb = g.subgraph([c, d])
        self.assertEquals(1, lb.number_of_edges())

    def test_linear_nested_flatten(self):
        a, b, c, d = _make_many(4)
        flo = lf.Flow("test")
        flo.add(a, b)
        flo2 = uf.Flow("test2")
        flo2.add(c, d)
        flo.add(flo2)
        g = f_utils.flatten(flo)
        self.assertEquals(4, len(g))

        lb = g.subgraph([a, b])
        self.assertTrue(lb.has_edge(a, b))
        self.assertFalse(lb.has_edge(b, a))

        ub = g.subgraph([c, d])
        self.assertEquals(0, ub.number_of_edges())

        # This ensures that c and d do not start executing until after b.
        self.assertTrue(g.has_edge(b, c))
        self.assertTrue(g.has_edge(b, d))

    def translate(self, flow):
        graph = flow_utils.flatten(flow)

        #TODO(jlucci): Logic to be re-written once task_id logic decided on
        for node in graph.nodes():
            if getattr(node, '_id', None) is None:
                node._id = "%s.%s" % (node.name, uuidutils.generate_uuid())
            self.engine.tasks[node._id] = (self.HybridTask(node,
                                           self.engine.celery_app))
            self.engine.storage.add_task(node._id, node.name)

        for (u, v) in graph.edges_iter():
            self._add(self.engine.tasks[u._id])
            self._add(self.engine.tasks[v._id])
            if not (u.provides).intersection(v.requires):
                self._link(self.engine.tasks[u._id],
                           self.engine.tasks[v._id])

        roots = graph_utils.get_no_predecessors(graph)
        for root in roots:
            self.engine.roots.append(self.engine.tasks[root._id])
        return graph