3 ##############################################################################
4 # Copyright (c) 2015 Ericsson AB and others.
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 ##############################################################################
12 # Unittest for yardstick.benchmark.scenarios.networking.pktgen.Pktgen
18 from yardstick.benchmark.scenarios.networking import pktgen
21 @mock.patch('yardstick.benchmark.scenarios.networking.pktgen.ssh')
22 class PktgenTestCase(unittest.TestCase):
26 'host': '172.16.0.137',
27 'target': '172.16.0.138',
29 'key_filename': "mykey.key"
32 def test_pktgen_successful_setup(self, mock_ssh):
34 p = pktgen.Pktgen(self.ctx)
36 'options': {'packetsize': 60},
37 'ipaddr': '172.16.0.139'
41 mock_ssh.SSH().execute.return_value = (0, '', '')
42 self.assertIsNotNone(p.server)
43 self.assertIsNotNone(p.client)
44 self.assertEqual(p.setup_done, True)
46 def test_pktgen_successful_iptables_setup(self, mock_ssh):
48 p = pktgen.Pktgen(self.ctx)
50 'options': {'packetsize': 60, 'number_of_ports': 10},
51 'ipaddr': '172.16.0.139'
53 p.server = mock_ssh.SSH()
54 p.number_of_ports = args['options']['number_of_ports']
56 mock_ssh.SSH().execute.return_value = (0, '', '')
60 mock_ssh.SSH().execute.assert_called_with(
62 "sudo iptables -A INPUT -p udp --dport 1000:%s -j DROP"
65 def test_pktgen_unsuccessful_iptables_setup(self, mock_ssh):
67 p = pktgen.Pktgen(self.ctx)
69 'options': {'packetsize': 60, 'number_of_ports': 10},
70 'ipaddr': '172.16.0.139'
72 p.server = mock_ssh.SSH()
73 p.number_of_ports = args['options']['number_of_ports']
75 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
76 self.assertRaises(RuntimeError, p._iptables_setup)
78 def test_pktgen_successful_iptables_get_result(self, mock_ssh):
80 p = pktgen.Pktgen(self.ctx)
82 'options': {'packetsize': 60, 'number_of_ports': 10},
83 'ipaddr': '172.16.0.139'
85 p.server = mock_ssh.SSH()
86 p.number_of_ports = args['options']['number_of_ports']
88 mock_ssh.SSH().execute.return_value = (0, '150000', '')
89 p._iptables_get_result()
91 mock_ssh.SSH().execute.assert_called_with(
92 "sudo iptables -L INPUT -vnx |"
93 "awk '/dpts:1000:%s/ {{printf \"%%s\", $1}}'"
96 def test_pktgen_unsuccessful_iptables_get_result(self, mock_ssh):
98 p = pktgen.Pktgen(self.ctx)
100 'options': {'packetsize': 60, 'number_of_ports': 10},
101 'ipaddr': '172.16.0.139'
103 p.server = mock_ssh.SSH()
104 p.number_of_ports = args['options']['number_of_ports']
106 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
107 self.assertRaises(RuntimeError, p._iptables_get_result)
109 def test_pktgen_successful_no_sla(self, mock_ssh):
111 p = pktgen.Pktgen(self.ctx)
113 'options': {'packetsize': 60, 'number_of_ports': 10},
114 'ipaddr': '172.16.0.139'
116 p.server = mock_ssh.SSH()
117 p.client = mock_ssh.SSH()
119 mock_iptables_result = mock.Mock()
120 mock_iptables_result.return_value = 149300
121 p._iptables_get_result = mock_iptables_result
123 sample_output = '{"packets_per_second": 9753, "errors": 0, \
124 "packets_sent": 149776, "flows": 110}'
125 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
128 expected_result = json.loads(sample_output)
129 expected_result["packets_received"] = 149300
130 self.assertEqual(result, expected_result)
132 def test_pktgen_successful_sla(self, mock_ssh):
134 p = pktgen.Pktgen(self.ctx)
136 'options': {'packetsize': 60, 'number_of_ports': 10},
137 'ipaddr': '172.16.0.139',
138 'sla': {'max_ppm': 10000}
140 p.server = mock_ssh.SSH()
141 p.client = mock_ssh.SSH()
143 mock_iptables_result = mock.Mock()
144 mock_iptables_result.return_value = 149300
145 p._iptables_get_result = mock_iptables_result
147 sample_output = '{"packets_per_second": 9753, "errors": 0, \
148 "packets_sent": 149776, "flows": 110}'
149 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
152 expected_result = json.loads(sample_output)
153 expected_result["packets_received"] = 149300
154 self.assertEqual(result, expected_result)
156 def test_pktgen_unsuccessful_sla(self, mock_ssh):
158 p = pktgen.Pktgen(self.ctx)
160 'options': {'packetsize': 60, 'number_of_ports': 10},
161 'ipaddr': '172.16.0.139',
162 'sla': {'max_ppm': 1000}
164 p.server = mock_ssh.SSH()
165 p.client = mock_ssh.SSH()
167 mock_iptables_result = mock.Mock()
168 mock_iptables_result.return_value = 149300
169 p._iptables_get_result = mock_iptables_result
171 sample_output = '{"packets_per_second": 9753, "errors": 0, \
172 "packets_sent": 149776, "flows": 110}'
173 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
174 self.assertRaises(AssertionError, p.run, args)
176 def test_pktgen_unsuccessful_script_error(self, mock_ssh):
178 p = pktgen.Pktgen(self.ctx)
180 'options': {'packetsize': 60, 'number_of_ports': 10},
181 'ipaddr': '172.16.0.139',
182 'sla': {'max_ppm': 1000}
184 p.server = mock_ssh.SSH()
185 p.client = mock_ssh.SSH()
187 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
188 self.assertRaises(RuntimeError, p.run, args)
194 if __name__ == '__main__':