pyrox.dev
1import unittest
2from .pipe import pipe
3
4from . import test_helpers as th
5
6from .lib import (
7 directed_graph,
8)
9
10
11if __name__ == "__main__":
12 unittest.main()
13
14
15def make_test_graph():
16 edges = [
17 ("Root1", "A"),
18 ("A", "B"),
19 ("A", "C"),
20 ("B", "D"),
21 ("B", "E"),
22 ("E", "F"),
23 ("B", "G"),
24 ("Root2", "B"),
25 ("Root3", "C"),
26 ]
27
28 return directed_graph(edges)
29
30
31class CustomAssertions:
32 def runAndAssertResult(self, graph, pipeline, expected_graph_args):
33 result = list(pipe(pipeline, graph))
34
35 for (index, expected_graph_arg) in enumerate(expected_graph_args):
36
37 self.assertGraphEqual(
38 directed_graph(*expected_graph_arg),
39 result[index]
40 )
41
42
43if __name__ == "__main__":
44 unittest.main()
45
46
47class Test(
48 unittest.TestCase,
49 CustomAssertions,
50 th.CustomAssertions
51):
52
53 def test_1(self):
54 pipeline = [
55 ["split_paths", ["B"]],
56 [
57 "over",
58 "main",
59 [
60 "pipe",
61 [
62 ["subcomponent_in", ["B"]],
63 [
64 "over",
65 "rest",
66 ["popularity_contest"]
67 ]
68 ]
69 ]
70 ],
71 ["flatten"],
72 ["map", ["remove_paths", "Root3"]],
73 ["limit_layers", 5],
74 ]
75
76 expected_graph_args = [
77 # "B"" separated from the rest by "split_paths" and
78 # "subcomponent_in' stages.
79 ([], ["B"]),
80 # Deps of "B", split into individual layers by "popularity_contest",
81 # with "F" being most popular
82 ([], ["F"]),
83 ([], ["D"]),
84 ([], ["E"]),
85 # "rest" output of "split_paths" stage with "G" merged into it by
86 # "limit_layers" stage.
87 (
88 [
89 ("Root1", "A"),
90 ("A", "C")
91 ],
92 ["Root2", "G"]
93 )
94 ]
95
96 self.runAndAssertResult(
97 make_test_graph(),
98 pipeline,
99 expected_graph_args
100 )
101
102 def test_2(self):
103 graph = directed_graph(
104 [
105 ("Root1", "A"),
106 ("A", "B"),
107 ],
108 ["Root2"]
109 )
110 self.runAndAssertResult(
111 graph,
112 [
113 ["popularity_contest"],
114 ],
115 [
116 # Ordered from most to least popular
117 ([], ["B"]),
118 ([], ["A"]),
119 ([], ["Root1"]),
120 ([], ["Root2"])
121 ]
122 )
123
124 self.runAndAssertResult(
125 graph,
126 [
127 ["popularity_contest"],
128 ["limit_layers", 3],
129 ],
130 [
131 # Most popular first
132 ([], ["B"]),
133 ([], ["A"]),
134 # Least popular combined
135 ([], ["Root1", "Root2"]),
136 ]
137 )
138
139 self.runAndAssertResult(
140 graph,
141 [
142 ["popularity_contest"],
143 ["reverse"],
144 ["limit_layers", 3],
145 ],
146 [
147 # Least popular first
148 ([], ["Root2"]),
149 ([], ["Root1"]),
150 # Most popular first
151 ([], ["A", "B"])
152 ]
153 )