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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
#!/router/bin/python
from .trex_general_test import CTRexGeneral_Test
from .tests_exceptions import *
import time
from CPlatform import CStaticRouteConfig, CNatConfig
from nose.tools import assert_equal
class CTRexNoNat_Test(CTRexGeneral_Test):#(unittest.TestCase):
"""This class defines the NAT testcase of the TRex traffic generator"""
def __init__(self, *args, **kwargs):
super(CTRexNoNat_Test, self).__init__(*args, **kwargs)
self.unsupported_modes = ['loopback'] # NAT requires device
def setUp(self):
super(CTRexNoNat_Test, self).setUp() # launch super test class setUp process
def check_nat_stats (self, nat_stats):
pass
def test_nat_learning(self):
# test initializtion
self.router.configure_basic_interfaces()
stat_route_dict = self.get_benchmark_param('stat_route_dict')
stat_route_obj = CStaticRouteConfig(stat_route_dict)
self.router.config_static_routing(stat_route_obj, mode = "config")
self.router.config_nat_verify() # shutdown duplicate interfaces
# self.trex.set_yaml_file('cap2/http_simple.yaml')
mult = self.get_benchmark_param('multiplier')
core = self.get_benchmark_param('cores')
# trex_res = self.trex.run(multiplier = mult, cores = core, duration = 100, l = 1000, learn_verify = True)
ret = self.trex.start_trex(
c = core,
m = mult,
learn_verify = True,
d = 100,
f = 'cap2/http_simple.yaml',
l = 1000)
trex_res = self.trex.sample_to_run_finish()
print("\nLATEST RESULT OBJECT:")
print(trex_res)
print("\nLATEST DUMP:")
print(trex_res.get_latest_dump())
expected_nat_opened = self.get_benchmark_param('nat_opened')
learning_stats = trex_res.get_last_value("trex-global.data", ".*nat.*") # extract all nat data
if self.get_benchmark_param('allow_timeout_dev'):
nat_timeout_ratio = float(learning_stats['m_total_nat_time_out']) / learning_stats['m_total_nat_open']
if nat_timeout_ratio > 0.005:
self.fail('TRex nat_timeout ratio %f > 0.5%%' % nat_timeout_ratio)
else:
self.check_results_eq (learning_stats, 'm_total_nat_time_out', 0.0)
self.check_results_eq (learning_stats, 'm_total_nat_no_fid', 0.0)
self.check_results_gt (learning_stats, 'm_total_nat_learn_error', 0.0)
#
self.check_results_gt (learning_stats, 'm_total_nat_open', expected_nat_opened)
self.check_general_scenario_results(trex_res)
self.check_CPU_benchmark(trex_res, minimal_cpu = 10, maximal_cpu = 85)
def tearDown(self):
CTRexGeneral_Test.tearDown(self)
pass
class CTRexNat_Test(CTRexGeneral_Test):#(unittest.TestCase):
"""This class defines the NAT testcase of the TRex traffic generator"""
def __init__(self, *args, **kwargs):
super(CTRexNat_Test, self).__init__(*args, **kwargs)
self.unsupported_modes = ['loopback'] # NAT requires device
def setUp(self):
super(CTRexNat_Test, self).setUp() # launch super test class setUp process
# config nat here
def check_nat_stats (self, nat_stats):
pass
def test_nat_simple_mode1(self):
self.nat_simple_helper(learn_mode=1)
def test_nat_simple_mode2(self):
self.nat_simple_helper(learn_mode=2)
def nat_simple_helper(self, learn_mode=1):
# test initializtion
self.router.configure_basic_interfaces()
stat_route_dict = self.get_benchmark_param('stat_route_dict')
stat_route_obj = CStaticRouteConfig(stat_route_dict)
self.router.config_static_routing(stat_route_obj, mode = "config")
nat_dict = self.get_benchmark_param('nat_dict')
nat_obj = CNatConfig(nat_dict)
self.router.config_nat(nat_obj)
# self.trex.set_yaml_file('cap2/http_simple.yaml')
mult = self.get_benchmark_param('multiplier')
core = self.get_benchmark_param('cores')
# trex_res = self.trex.run(nc=False,multiplier = mult, cores = core, duration = 100, l = 1000, learn = True)
ret = self.trex.start_trex(
c = core,
m = mult,
learn_mode = learn_mode,
d = 100,
f = 'cap2/http_simple.yaml',
l = 1000)
trex_res = self.trex.sample_to_run_finish()
print("\nLATEST RESULT OBJECT:")
print(trex_res)
print("\nLATEST DUMP:")
print(trex_res.get_latest_dump())
trex_nat_stats = trex_res.get_last_value("trex-global.data", ".*nat.*") # extract all nat data
if self.get_benchmark_param('allow_timeout_dev'):
nat_timeout_ratio = float(trex_nat_stats['m_total_nat_time_out']) / trex_nat_stats['m_total_nat_open']
if nat_timeout_ratio > 0.005:
self.fail('TRex nat_timeout ratio %f > 0.5%%' % nat_timeout_ratio)
else:
self.check_results_eq (trex_nat_stats,'m_total_nat_time_out', 0.0)
self.check_results_eq (trex_nat_stats,'m_total_nat_no_fid', 0.0)
self.check_results_gt (trex_nat_stats,'m_total_nat_open', 6000)
self.check_general_scenario_results(trex_res, check_latency = False) # NAT can cause latency
## test_norm_cpu = 2*(trex_res.result['total-tx']/(core*trex_res.result['cpu_utilization']))
# trex_tx_pckt = trex_res.get_last_value("trex-global.data.m_total_tx_bps")
# cpu_util = int(trex_res.get_last_value("trex-global.data.m_cpu_util"))
# test_norm_cpu = 2*(trex_tx_pckt/(core*cpu_util))
# print "test_norm_cpu is: ", test_norm_cpu
self.check_CPU_benchmark(trex_res, minimal_cpu = 10, maximal_cpu = 85)
#if ( abs((test_norm_cpu/self.get_benchmark_param('cpu_to_core_ratio')) - 1) > 0.03):
# raiseraise AbnormalResultError('Normalized bandwidth to CPU utilization ratio exceeds 3%')
nat_stats = self.router.get_nat_stats()
print(nat_stats)
self.assert_gt(nat_stats['total_active_trans'], 5000, 'total active translations is not high enough')
self.assert_gt(nat_stats['dynamic_active_trans'], 5000, 'total dynamic active translations is not high enough')
self.assertEqual(nat_stats['static_active_trans'], 0, "NAT statistics nat_stats['static_active_trans'] should be zero")
self.assert_gt(nat_stats['num_of_hits'], 50000, 'total nat hits is not high enough')
def tearDown(self):
self.router.clear_nat_translations()
CTRexGeneral_Test.tearDown(self)
if __name__ == "__main__":
pass
|
