-
Notifications
You must be signed in to change notification settings - Fork 0
/
code_root_ctrl_MCTestScen_v5_CSPF.py
129 lines (107 loc) · 5.46 KB
/
code_root_ctrl_MCTestScen_v5_CSPF.py
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
#!/usr/bin/python
import logging
import unittest
from code_root_ctrl_MCTestScen_v5 import RootCtrlTest
class RootCtrlTestCSPF(RootCtrlTest):
""" Multi-Controller Test Scenario V5 test that uses the CSPFRecomp
TE-optimisation method with a candidate (path) revsort attribute of
True (heavy hitters considered first)
Test Scenarios:
Docs/MCTestScenarios/scen-v5-CSPFRecomp-CandidateRevsort.png
"""
revsort = True
def __init__(self, *args, **kwargs):
super(RootCtrlTestCSPF, self).__init__(*args, **kwargs)
def test_case_01(self, show_print=True):
""" Scenario 1: h1 -> h8 60M, h2 -> h8 50M
Outcome:
c1 Should swap h1-h8 to s2-s6
"""
if show_print:
print("\nTesting scenario 1 (h1->h8 60M | h2-h8 50M) CSPF")
# Configure the initial state of the test by running through previous scenarios
self.test_case_00(show_print=False)
# Ingress and egress change modifications received from local controllers for local
# TE optimisation
c1_change = {"cid": "c1", "hkey": ("h1", "h8"), "new_paths": [
{"action": "add", "in": -1, "out": ("s02", 5), "out_addr": "0."},
{"action": "add", "in": -1, "out": ("s02", 4), "out_addr": "0."}]
}
c2_change = {"cid": "c2", "hkey": ("h1", "h8"), "new_paths": [
{"action": "add", "in": ("s06", 3), "out": ("s05", 4)},
{"action": "add", "in": ("s04", 3), "out": ("s07", 2)}]
}
# Initial expected path instructions for state at start of scenario (default state)
exp_paths = {}
exp_inst = {"c1": {}}
# Perform egress change on C1
self.ctrl._path_info_changed(c1_change)
exp_inst["c1"][("h1", "h8")] = c1_change["new_paths"]
self._check_computed_paths(exp_paths, exp_inst)
# Perform the ingress and egress change at C2
self.ctrl._path_info_changed(c2_change)
exp_paths[("h1", "h8")] = {}
exp_paths[("h1", "h8")]["prim"] = ["h1", "c1", "s02", "s06", "c2", "s05", "s08", "c3", "h8"]
exp_paths[("h1", "h8")]["sec"] = ["h1", "c1", "s02", "s04", "c2", "s07", "s09", "c3", "h8"]
exp_inst["c2"] = {}
exp_inst["c2"][("h1", "h8")] = c2_change["new_paths"]
self._check_computed_paths(exp_paths, exp_inst)
def test_case_03(self, show_print=True):
""" Scenario 3: h1 -> h8 110M, h2 -> h8 100M
Outcome:
C2 should swap h1-h8 to s7-s9
"""
if show_print:
print("\nTesting scenario 3 (h1->h8 110M | h2-h8 100M) CSPF")
# Configure the initial state of the test by running through previous scenarios
self.test_case_02(show_print=False)
# Local controller ingress and egress notifications sent to root controller
c2_change = {"cid": "c2", "hkey": ("h1", "h8"), "new_paths": [
{"action": "add", "in": ("s06", 3), "out": ("s07", 2)},
{"action": "add", "in": ("s04", 3), "out": ("s05", 4)}]
}
c3_change = {"cid": "c3", "hkey": ("h1", "h8"), "new_paths": [
{"action": "add", "in": ("s09", 2), "out": -1, "out_eth": "a::"},
{"action": "add", "in": ("s08", 3), "out": -1, "out_eth": "a::"}]
}
# Expected path and instructions
exp_paths = {
("h1", "h8"): {
"prim": ["h1", "c1", "s02", "s06", "c2", "s07", "s08", "c3", "h8"],
"sec": ["h1", "c1", "s02", "s04", "c2", "s05", "s09", "c3", "h8"]},
("h2", "h8"): {
"prim": ["h2", "c1", "s02", "s06", "c2", "s05", "s08", "c3", "h8"],
"sec": ["h2", "c1", "s02", "s04", "c2", "s07", "s09", "c3", "h8"]}
}
exp_inst = {
"c1": {
("h1", "h8"): [
{"action": "add", "in": -1, "out": ("s02", 5), "out_addr": "0."},
{"action": "add", "in": -1, "out": ("s02", 4), "out_addr": "0."}],
("h2", "h8"): [
{"action": "add", "in": -1, "out": ("s02", 5), "out_addr": "0."},
{"action": "add", "in": -1, "out": ("s02", 4), "out_addr": "0."}],
}, "c2": {
("h2", "h8"): [
{"action": "add", "in": ("s06", 3), "out": ("s05", 4)},
{"action": "add", "in": ("s04", 3), "out": ("s07", 2)}],
}, "c3": {
("h1", "h8"): [
{"action": "add", "in": ("s08", 3), "out": -1, "out_eth": "a::"},
{"action": "add", "in": ("s09", 2), "out": -1, "out_eth": "a::"}],
("h2", "h8"): [
{"action": "add", "in": ("s08", 3), "out": -1, "out_eth": "a::"},
{"action": "add", "in": ("s09", 2), "out": -1, "out_eth": "a::"}],
},
}
# Perform the ingress and egress change at C2
self.ctrl._path_info_changed(c2_change)
exp_inst["c2"][("h1", "h8")] = c2_change["new_paths"]
self._check_computed_paths(exp_paths, exp_inst)
# Perform the ingress change at C3
self.ctrl._path_info_changed(c3_change)
exp_paths[("h1", "h8")] = {}
exp_paths[("h1", "h8")]["prim"] = ["h1", "c1", "s02", "s06", "c2", "s07", "s09", "c3", "h8"]
exp_paths[("h1", "h8")]["sec"] = ["h1", "c1", "s02", "s04", "c2", "s05", "s08", "c3", "h8"]
exp_inst["c3"][("h1", "h8")] = c3_change["new_paths"]
self._check_computed_paths(exp_paths, exp_inst)