summaryrefslogtreecommitdiffstats
path: root/scripts/automation/regression/unit_tests/trex_nat_test.py
blob: e50831e48864a257a32ba7dd9d77c7d640e3b88b (plain)
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