class CircularDependencyError(SQLAlchemyError):
"""Raised by topological sorts when a circular dependency is detected"""
-
+ def __init__(self, message, cycles, edges):
+ message += ": cycles: %r all edges: %r" % (cycles, edges)
+ super(CircularDependencyError, self).__init__(message)
+ self.cycles = cycles
+ self.edges = edges
class CompileError(SQLAlchemyError):
"""Raised when an error occurs during SQL compilation"""
from sqlalchemy.exc import CircularDependencyError
from sqlalchemy import util
+import itertools
__all__ = ['sort', 'sort_as_subsets', 'find_cycles']
if not output:
raise CircularDependencyError(
- "Circular dependency detected: cycles: %r all edges: %s" %
- (find_cycles(tuples, allitems), _dump_edges(edges, True)))
+ "Circular dependency detected",
+ find_cycles(tuples, allitems),
+ _gen_edges(edges)
+ )
todo.difference_update(output)
yield output
node = stack.pop()
return output
-def _dump_edges(edges, reverse):
- l = []
- for left in edges:
- for right in edges[left]:
- if reverse:
- l.append((right, left))
- else:
- l.append((left, right))
- return repr(l)
-
-
+def _gen_edges(edges):
+ return set(itertools.chain(
+ *[
+ [
+ (right, left) for right in edges[left]
+ ] for left in edges
+ ]
+ ))
(node4, node1),
]
allitems = self._nodes_from_tuples(tuples)
- assert_raises(exc.CircularDependencyError, list,
- topological.sort(tuples, allitems))
- # TODO: test find_cycles
+ try:
+ list(topological.sort(tuples, allitems))
+ assert False
+ except exc.CircularDependencyError, err:
+ eq_(err.cycles, set(['node1', 'node3', 'node2', 'node5',
+ 'node4']))
+ eq_(err.edges, set([('node3', 'node1'), ('node4', 'node1'),
+ ('node2', 'node3'), ('node1', 'node2'),
+ ('node4','node5'), ('node5', 'node4')]))
def test_raise_on_cycle_two(self):
tuples = [(node1, node2), (node3, node1), (node2, node4),
(node3, node2), (node2, node3)]
allitems = self._nodes_from_tuples(tuples)
- assert_raises(exc.CircularDependencyError, list,
- topological.sort(tuples, allitems))
- # TODO: test find_cycles
+ try:
+ list(topological.sort(tuples, allitems))
+ assert False
+ except exc.CircularDependencyError, err:
+ eq_(err.cycles, set(['node1', 'node3', 'node2']))
+ eq_(err.edges, set([('node3', 'node1'), ('node2', 'node3'),
+ ('node3', 'node2'), ('node1', 'node2'),
+ ('node2','node4')]))
def test_raise_on_cycle_three(self):
question, issue, providerservice, answer, provider = \