TestPathsEdges.test_cspf_delay_between_u2_u3()   A
last analyzed

Complexity

Conditions 2

Size

Total Lines 20
Code Lines 15

Duplication

Lines 20
Ratio 100 %

Code Coverage

Tests 13
CRAP Score 2

Importance

Changes 0
Metric Value
cc 2
eloc 15
nop 1
dl 20
loc 20
ccs 13
cts 13
cp 1
crap 2
rs 9.65
c 0
b 0
f 0
1
"""Module to test the KytosGraph in graph.py."""
2 1
import pytest
3 1
from itertools import combinations
4
5
# pylint: disable=import-error
6 1
from tests.integration.edges_settings import EdgesSettings
7
8
9 1
class TestPathsEdges(EdgesSettings):
10
    """TestPathsEdges."""
11
12 1
    @pytest.mark.parametrize("source,destination",
13
                             combinations(["User1", "User2", "User3", "User4"], 2))
14 1
    def test_k_shortest_paths_among_users(self, source, destination):
15
        """Tests paths between all users using unconstrained path algorithm."""
16 1
        self.initializer()
17 1
        paths = self.graph.k_shortest_paths(source, destination)
18 1
        assert paths
19 1
        for path in paths:
20 1
            assert path[0] == source
21 1
            assert path[-1] == destination
22
23 1
    @pytest.mark.parametrize("source,destination",
24
                             combinations(["User1", "User2", "User3", "User4"], 2))
25 1
    def test_constrained_k_shortest_paths_among_users(self, source, destination):
26
        """Tests paths between all users using constrained path algorithm,
27
        with no constraints set.
28
        """
29 1
        self.initializer()
30 1
        paths = self.graph.constrained_k_shortest_paths(
31
            source, destination
32
        )
33 1
        assert paths
34 1
        for path in paths:
35 1
            assert path["hops"][0] == source
36 1
            assert path["hops"][-1] == destination
37
38 1
    def test_cspf_delay_spf_attribute_between_u1_u4(self):
39
        """Test CSPF delay spf attribute between user1 and user4."""
40 1
        self.initializer()
41 1
        source = "User1"
42 1
        destination = "User4"
43 1
        spf_attribute = "delay"
44 1
        paths = self.graph.constrained_k_shortest_paths(
45
            source, destination, weight=spf_attribute
46
        )
47 1
        assert paths
48 1
        for path in paths:
49 1
            assert path["hops"][0] == source
50 1
            assert path["hops"][-1] == destination
51 1
        paths = self.graph.path_cost_builder(paths, weight=spf_attribute)
52 1
        assert paths[0]["cost"] == 105 + 1 + 1
53
54 1 View Code Duplication
    def test_cspf_reliability_between_u1_u2(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
55
        """Test CSPF reliability constraint between user1 and user2."""
56 1
        self.initializer()
57 1
        source = "User1"
58 1
        destination = "User2"
59 1
        paths = self.graph.constrained_k_shortest_paths(
60
            source, destination, mandatory_metrics={"reliability": 10}
61
        )
62 1
        assert not paths
63
64 1
        paths = self.graph.constrained_k_shortest_paths(
65
            source, destination, mandatory_metrics={"reliability": 3}
66
        )
67 1
        assert paths
68
69 1
        for path in paths:
70 1
            assert path["hops"][0] == source
71 1
            assert path["hops"][-1] == destination
72 1
            assert path["metrics"] == {"reliability": 3}
73 1
        paths = self.graph.path_cost_builder(paths)
74 1
        assert paths[0]["cost"] == 12
75
76 1
    def test_cspf_bandwidth_between_u1_u4(self):
77
        """Test CSPF bandwidth constraint between user1 and user4."""
78 1
        self.initializer()
79 1
        source = "User1"
80 1
        destination = "User4"
81 1
        spf_attribute = "delay"
82 1
        paths = self.graph.constrained_k_shortest_paths(
83
            source,
84
            destination,
85
            weight=spf_attribute,
86
            mandatory_metrics={"bandwidth": 200},
87
        )
88 1
        assert not paths
89
90 1
        paths = self.graph.constrained_k_shortest_paths(
91
            source,
92
            destination,
93
            weight=spf_attribute,
94
            mandatory_metrics={"bandwidth": 100},
95
        )
96 1
        assert paths
97
98 1
        for path in paths:
99 1
            assert path["hops"][0] == source
100 1
            assert path["hops"][-1] == destination
101 1
            assert path["metrics"] == {"bandwidth": 100}
102 1
        paths = self.graph.path_cost_builder(paths, weight=spf_attribute)
103 1
        assert paths[0]["cost"] == 122
104
105 1 View Code Duplication
    def test_cspf_delay_between_u2_u3(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
106
        """Test CSPF delay constraint between user2 and user3."""
107 1
        self.initializer()
108 1
        source = "User2"
109 1
        destination = "User3"
110
111 1
        paths = self.graph.constrained_k_shortest_paths(
112
            source, destination, mandatory_metrics={"delay": 1}
113
        )
114 1
        assert not paths
115
116 1
        paths = self.graph.constrained_k_shortest_paths(
117
            source, destination, mandatory_metrics={"delay": 50}
118
        )
119 1
        assert paths
120 1
        for path in paths:
121 1
            assert path["hops"][0] == source
122 1
            assert path["hops"][-1] == destination
123 1
        paths = self.graph.path_cost_builder(paths)
124 1
        assert paths[0]["cost"] >= 3
125
126 1 View Code Duplication
    def test_cspf_ownership_between_s4_s6(self):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
127
        """Test CSPF ownership constraint between switch4 and switch6."""
128 1
        self.initializer()
129 1
        source = "S6:2"
130 1
        destination = "S4:2"
131
132 1
        paths = self.graph.constrained_k_shortest_paths(
133
            source, destination, mandatory_metrics={"ownership": "B"}
134
        )
135 1
        assert not paths
136
137 1
        paths = self.graph.constrained_k_shortest_paths(
138
            source, destination, mandatory_metrics={"ownership": "A"}
139
        )
140 1
        assert paths
141 1
        for path in paths:
142 1
            assert path["hops"][0] == source
143 1
            assert path["hops"][-1] == destination
144 1
        paths = self.graph.path_cost_builder(paths)
145 1
        assert paths[0]["cost"] >= 3
146
147 1
    def test_cspf_flexible_between_s4_s6(self):
148
        """Test CSPF flexible constraint between switch4 and switch6."""
149 1
        self.initializer()
150 1
        source = "S6:2"
151 1
        destination = "S4:2"
152
153 1
        paths = self.graph.constrained_k_shortest_paths(
154
            source,
155
            destination,
156
            mandatory_metrics={"reliability": 2},
157
            flexible_metrics={"bandwidth": 60},
158
            minimium_hits=1,
159
        )
160 1
        assert paths
161 1
        for path in paths:
162 1
            assert path["hops"][0] == source
163 1
            assert path["hops"][-1] == destination
164 1
        paths = self.graph.path_cost_builder(paths)
165 1
        assert paths[0]["cost"] >= 3
166
167 1
    def test_cspf_paths_mandatory_with_flexible(self):
168
        """Tests paths between all users using constrained path algorithm,
169
        with the delay constraint set to 50, the bandwidth constraint
170
        set to 100, the reliability constraint set to 3, and the ownership
171
        constraint set to 'B'
172
173
        Tests conducted with all but ownership flexible
174
        """
175 1
        combos = combinations(["User1", "User2", "User3", "User4"], 2)
176 1
        self.initializer()
177
178 1
        for source, destination in combos:
179 1
            paths = self.graph.constrained_k_shortest_paths(
180
                source,
181
                destination,
182
                mandatory_metrics={"ownership": "B"},
183
                flexible_metrics={
184
                    "delay": 50,
185
                    "bandwidth": 100,
186
                    "reliability": 3,
187
                },
188
            )
189 1
            for path in paths:
190 1
                hops_set = set(path["hops"])
191
192
                # delay = 50 checks
193 1
                if "delay" in path["metrics"]:
194 1
                    nodes = set([
195
                        "S1:1", "S2:1", "S3:1", "S5:1", "S4:2", "User1:2",
196
                        "S5:5", "S8:2", "S5:6", "User1:3", "S6:3", "S9:1",
197
                        "S6:4", "S9:2", "S6:5", "S10:1", "S8:5", "S9:4",
198
                        "User1:4", "User4:3"
199
                    ])
200 1
                    assert not nodes & hops_set
201
202
                # bandwidth = 100 checks
203 1
                if "bandwidth" in path["metrics"]:
204 1
                    nodes = set(["S3:1", "S5:1", "User1:4", "User4:3"])
205 1
                    assert not nodes & hops_set
206
207
                # reliability = 3 checks
208 1
                if "reliability" in path["metrics"]:
209 1
                    nodes = set(["S4:1", "S5:2", "S5:3", "S6:1"])
210 1
                    assert not nodes & hops_set
211
212
                # ownership = "B" checks
213 1
                assert "ownership" in path["metrics"]
214 1
                nodes = set([
215
                    "S4:1", "S5:2", "User1:2", "S5:4",
216
                    "S6:2", "S6:5", "S10:1", "S8:6", "S10:2",
217
                    "S10:3", "User2:1"
218
                ])
219 1
                assert not nodes & hops_set
220
221 1
    def test_ownership_type_error(self):
222
        """Tests that TypeError."""
223 1
        self.initializer()
224
225 1
        with pytest.raises(TypeError):
226 1
            self.graph.constrained_k_shortest_paths(
227
                "User1", "User2", mandatory_metrics={"ownership": 1}
228
            )
229