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
167
168
169
|
#!/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 T-Rex traffic generator"""
def __init__(self, *args, **kwargs):
super(CTRexNoNat_Test, self).__init__(*args, **kwargs)
self.unsupported_modes = ['loopback'] # NAT requires device
pass
def setUp(self):
super(CTRexNoNat_Test, self).setUp() # launch super test class setUp process
pass
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 = 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.005 (0.5%) and not as expected to be less than 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 T-Rex traffic generator"""
def __init__(self, *args, **kwargs):
super(CTRexNat_Test, self).__init__(*args, **kwargs)
self.unsupported_modes = ['loopback'] # NAT requires device
pass
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 = 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)
## 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):
CTRexGeneral_Test.tearDown(self)
self.router.clear_nat_translations()
if __name__ == "__main__":
pass
|