[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]

[PATCH 07/15] Add a topological sort implementation for use in sorting device actions.



---
 pyanaconda/tsort.py |  102 +++++++++++++++++++++++++++++++++++++++++++++++++++
 tests/Makefile.am   |    2 +-
 tests/tsort_test.py |   62 +++++++++++++++++++++++++++++++
 3 files changed, 165 insertions(+), 1 deletions(-)
 create mode 100644 pyanaconda/tsort.py
 create mode 100644 tests/tsort_test.py

diff --git a/pyanaconda/tsort.py b/pyanaconda/tsort.py
new file mode 100644
index 0000000..5c6213c
--- /dev/null
+++ b/pyanaconda/tsort.py
@@ -0,0 +1,102 @@
+#!/usr/bin/python
+# tsort.py
+# Topological sorting.
+#
+# Copyright (C) 2010  Red Hat, Inc.
+#
+# This copyrighted material is made available to anyone wishing to use,
+# modify, copy, or redistribute it subject to the terms and conditions of
+# the GNU General Public License v.2, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY expressed or implied, including the implied warranties of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
+# Public License for more details.  You should have received a copy of the
+# GNU General Public License along with this program; if not, write to the
+# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
+# source code or documentation are not subject to the GNU General Public
+# License and may only be used or replicated with the express permission of
+# Red Hat, Inc.
+#
+# Red Hat Author(s): Dave Lehman <dlehman redhat com>
+#
+
+class CyclicGraphError(Exception):
+    pass
+
+def tsort(graph):
+    order = []  # sorted list of items
+
+    # determine which nodes have no incoming edges
+    roots = [n for n in graph['items'] if graph['incoming'][n] == 0]
+    if not roots:
+        raise CyclicGraphError("no root nodes")
+
+    visited = []    # list of nodes visited, for cycle detection
+    while roots:
+        # remove a root, add it to the order
+        root = roots.pop()
+        if root in visited:
+            raise CyclicGraphError("graph contains cycles")
+
+        visited.append(root)
+        i = graph['items'].index(root)
+        order.append(root)
+        # remove each edge from the root to another node
+        for (parent, child) in [e for e in graph['edges'] if e[0] == root]:
+            graph['incoming'][child] -= 1
+            graph['edges'].remove((parent, child))
+            # if destination node is now a root, add it to roots
+            if graph['incoming'][child] == 0:
+                roots.append(child)
+
+    if len(graph['items']) != len(visited):
+        raise CyclicGraphError("graph contains cycles")
+
+
+    return order
+
+def create_graph(items, edges):
+    """ Create a graph based on a list of items and a list of edges.
+
+        Arguments:
+
+            items   -   an iterable containing (hashable) items to sort
+            edges   -   an iterable containing (parent, child) edge pair tuples
+
+        Return Value:
+
+            The return value is a dictionary representing the directed graph.
+            It has three keys:
+
+                items is the same as the input argument of the same name
+                edges is the same as the input argument of the same name
+                incoming is a dict of incoming edge count hashed by item
+
+    """
+    graph = {'items': [],       # the items to sort
+             'edges': [],       # partial order info: (parent, child) pairs
+             'incoming': {}}    # incoming edge count for each item
+
+    graph['items'] = items
+    graph['edges'] = edges
+    for item in items:
+        graph['incoming'][item] = 0
+
+    for (parent, child) in edges:
+        graph['incoming'][child] += 1
+
+    return graph 
+
+
+if __name__ == "__main__":
+
+    items = [5, 2, 3, 4, 1]
+    edges = [(1, 2), (2, 4), (4, 5), (3, 2)]
+
+    print items
+    print edges
+
+    graph = create_graph(items, edges)
+    print tsort(graph)
+
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 9a938c9..5e95467 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -24,4 +24,4 @@ MAINTAINERCLEANFILES = Makefile.in
 ANACDIR = $(top_builddir)/pyanaconda
 TESTS_ENVIRONMENT = PATH=/sbin:/usr/sbin:$$PATH PYTHONPATH=$(top_builddir)/tests:$(ANACDIR)/isys/.libs:$(ANACDIR)/isys:$(ANACDIR)
 
-TESTS = fw_test.py
+TESTS = fw_test.py tsort_test.py
diff --git a/tests/tsort_test.py b/tests/tsort_test.py
new file mode 100644
index 0000000..6202971
--- /dev/null
+++ b/tests/tsort_test.py
@@ -0,0 +1,62 @@
+
+import unittest
+import tsort
+
+class TopologicalSortTestCase(unittest.TestCase):
+    def runTest(self):
+        items = [1, 2, 3, 4, 5]
+        edges = [(5, 4), (4, 3), (3, 2), (2, 1)]
+        graph = tsort.create_graph(items, edges)
+        self._tsortTest(graph)
+
+        edges = [(5, 4), (2, 3), (1, 5)]
+        graph = tsort.create_graph(items, edges)
+        self._tsortTest(graph)
+
+        edges = [(5, 4), (4, 3), (3, 2), (2, 1), (3, 5)]
+        graph = tsort.create_graph(items, edges)
+        self.failUnlessRaises(tsort.CyclicGraphError,
+                              tsort.tsort_dict,
+                              graph)
+
+        edges = [(5, 4), (4, 3), (3, 2), (2, 1), (2, 3)]
+        graph = tsort.create_graph(items, edges)
+        self.failUnlessRaises(tsort.CyclicGraphError,
+                              tsort.tsort_dict,
+                              graph)
+
+        items = ['a', 'b', 'c', 'd']
+        edges = [('a', 'c'), ('c', 'b')]
+        graph = tsort.create_graph(items, edges)
+        self._tsortTest(graph)
+
+    def _tsortTest(self, graph):
+        def check_order(order, graph):
+            # since multiple solutions can potentially exist, just verify
+            # that the ordering constraints are satisfied
+            for parent, child in graph['edges']:
+                if order.index(parent) > order.index(child):
+                    return False
+            return True
+
+        try:
+            order = tsort.tsort_dict(graph)
+        except Exception as e:
+            self.fail(e)
+
+        # verify output list is of the correct length
+        self.failIf(len(order) != len(graph['items']),
+                    "sorted list length is incorrect")
+
+        # verify that all ordering constraints are satisfied
+        self.failUnless(check_order(order, graph),
+                        "ordering constraints not satisfied")
+
+
+def suite():
+    return unittest.TestLoader().loadTestsFromTestCase(TopologicalSortTestCase)
+
+
+if __name__ == "__main__":
+    unittest.main()
+
-- 
1.7.2.3


[Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]