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):
29 'key_filename': 'mykey.key'
34 'key_filename': 'mykey.key',
35 'ipaddr': '172.16.0.138'
39 def test_pktgen_successful_setup(self, mock_ssh):
42 'options': {'packetsize': 60},
44 p = pktgen.Pktgen(args, self.ctx)
47 mock_ssh.SSH().execute.return_value = (0, '', '')
48 self.assertIsNotNone(p.server)
49 self.assertIsNotNone(p.client)
50 self.assertEqual(p.setup_done, True)
52 def test_pktgen_successful_iptables_setup(self, mock_ssh):
55 'options': {'packetsize': 60, 'number_of_ports': 10},
57 p = pktgen.Pktgen(args, self.ctx)
58 p.server = mock_ssh.SSH()
59 p.number_of_ports = args['options']['number_of_ports']
61 mock_ssh.SSH().execute.return_value = (0, '', '')
65 mock_ssh.SSH().execute.assert_called_with(
67 "sudo iptables -A INPUT -p udp --dport 1000:%s -j DROP"
70 def test_pktgen_unsuccessful_iptables_setup(self, mock_ssh):
73 'options': {'packetsize': 60, 'number_of_ports': 10},
76 p = pktgen.Pktgen(args, self.ctx)
77 p.server = mock_ssh.SSH()
78 p.number_of_ports = args['options']['number_of_ports']
80 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
81 self.assertRaises(RuntimeError, p._iptables_setup)
83 def test_pktgen_successful_iptables_get_result(self, mock_ssh):
86 'options': {'packetsize': 60, 'number_of_ports': 10},
89 p = pktgen.Pktgen(args, self.ctx)
90 p.server = mock_ssh.SSH()
91 p.number_of_ports = args['options']['number_of_ports']
93 mock_ssh.SSH().execute.return_value = (0, '150000', '')
94 p._iptables_get_result()
96 mock_ssh.SSH().execute.assert_called_with(
97 "sudo iptables -L INPUT -vnx |"
98 "awk '/dpts:1000:%s/ {{printf \"%%s\", $1}}'"
101 def test_pktgen_unsuccessful_iptables_get_result(self, mock_ssh):
104 'options': {'packetsize': 60, 'number_of_ports': 10},
107 p = pktgen.Pktgen(args, self.ctx)
109 p.server = mock_ssh.SSH()
110 p.number_of_ports = args['options']['number_of_ports']
112 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
113 self.assertRaises(RuntimeError, p._iptables_get_result)
115 def test_pktgen_successful_no_sla(self, mock_ssh):
118 'options': {'packetsize': 60, 'number_of_ports': 10},
122 p = pktgen.Pktgen(args, self.ctx)
124 p.server = mock_ssh.SSH()
125 p.client = mock_ssh.SSH()
127 mock_iptables_result = mock.Mock()
128 mock_iptables_result.return_value = 149300
129 p._iptables_get_result = mock_iptables_result
131 sample_output = '{"packets_per_second": 9753, "errors": 0, \
132 "packets_sent": 149776, "flows": 110}'
133 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
136 expected_result = json.loads(sample_output)
137 expected_result["packets_received"] = 149300
138 self.assertEqual(result, expected_result)
140 def test_pktgen_successful_sla(self, mock_ssh):
143 'options': {'packetsize': 60, 'number_of_ports': 10},
144 'sla': {'max_ppm': 10000}
148 p = pktgen.Pktgen(args, self.ctx)
150 p.server = mock_ssh.SSH()
151 p.client = mock_ssh.SSH()
153 mock_iptables_result = mock.Mock()
154 mock_iptables_result.return_value = 149300
155 p._iptables_get_result = mock_iptables_result
157 sample_output = '{"packets_per_second": 9753, "errors": 0, \
158 "packets_sent": 149776, "flows": 110}'
159 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
162 expected_result = json.loads(sample_output)
163 expected_result["packets_received"] = 149300
164 self.assertEqual(result, expected_result)
166 def test_pktgen_unsuccessful_sla(self, mock_ssh):
169 'options': {'packetsize': 60, 'number_of_ports': 10},
170 'sla': {'max_ppm': 1000}
174 p = pktgen.Pktgen(args, self.ctx)
176 p.server = mock_ssh.SSH()
177 p.client = mock_ssh.SSH()
179 mock_iptables_result = mock.Mock()
180 mock_iptables_result.return_value = 149300
181 p._iptables_get_result = mock_iptables_result
183 sample_output = '{"packets_per_second": 9753, "errors": 0, \
184 "packets_sent": 149776, "flows": 110}'
185 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
186 self.assertRaises(AssertionError, p.run, result)
188 def test_pktgen_unsuccessful_script_error(self, mock_ssh):
191 'options': {'packetsize': 60, 'number_of_ports': 10},
192 'sla': {'max_ppm': 1000}
196 p = pktgen.Pktgen(args, self.ctx)
198 p.server = mock_ssh.SSH()
199 p.client = mock_ssh.SSH()
201 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
202 self.assertRaises(RuntimeError, p.run, result)
208 if __name__ == '__main__':