1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
from unittest.mock import Mock
from celery.utils.graph import DependencyGraph
from celery.utils.text import WhateverIO
class test_DependencyGraph:
def graph1(self):
res_a = self.app.AsyncResult('A')
res_b = self.app.AsyncResult('B')
res_c = self.app.GroupResult('C', [res_a])
res_d = self.app.GroupResult('D', [res_c, res_b])
node_a = (res_a, [])
node_b = (res_b, [])
node_c = (res_c, [res_a])
node_d = (res_d, [res_c, res_b])
return DependencyGraph([
node_a,
node_b,
node_c,
node_d,
])
def test_repr(self):
assert repr(self.graph1())
def test_topsort(self):
order = self.graph1().topsort()
# C must start before D
assert order.index('C') < order.index('D')
# and B must start before D
assert order.index('B') < order.index('D')
# and A must start before C
assert order.index('A') < order.index('C')
def test_edges(self):
edges = self.graph1().edges()
assert sorted(edges, key=str) == ['C', 'D']
def test_connect(self):
x, y = self.graph1(), self.graph1()
x.connect(y)
def test_valency_of_when_missing(self):
x = self.graph1()
assert x.valency_of('foobarbaz') == 0
def test_format(self):
x = self.graph1()
x.formatter = Mock()
obj = Mock()
assert x.format(obj)
x.formatter.assert_called_with(obj)
x.formatter = None
assert x.format(obj) is obj
def test_items(self):
assert dict(self.graph1().items()) == {
'A': [], 'B': [], 'C': ['A'], 'D': ['C', 'B'],
}
def test_repr_node(self):
x = self.graph1()
assert x.repr_node('fasdswewqewq')
def test_to_dot(self):
s = WhateverIO()
self.graph1().to_dot(s)
assert s.getvalue()
|
