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.iperf3.Iperf
19 from yardstick.benchmark.scenarios.networking import iperf3
22 @mock.patch('yardstick.benchmark.scenarios.networking.iperf3.ssh')
23 class IperfTestCase(unittest.TestCase):
24 output_name_tcp = 'iperf3_sample_output.json'
25 output_name_udp = 'iperf3_sample_output_udp.json'
32 'key_filename': 'mykey.key'
37 'key_filename': 'mykey.key',
38 'ipaddr': '172.16.0.138',
42 def test_iperf_successful_setup(self, mock_ssh):
44 p = iperf3.Iperf({}, self.ctx)
45 mock_ssh.SSH().execute.return_value = (0, '', '')
48 self.assertIsNotNone(p.target)
49 self.assertIsNotNone(p.host)
50 mock_ssh.SSH().execute.assert_called_with("iperf3 -s -D")
52 def test_iperf_unsuccessful_setup(self, mock_ssh):
54 p = iperf3.Iperf({}, self.ctx)
55 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
56 self.assertRaises(RuntimeError, p.setup)
58 def test_iperf_successful_teardown(self, mock_ssh):
60 p = iperf3.Iperf({}, self.ctx)
61 mock_ssh.SSH().execute.return_value = (0, '', '')
62 p.host = mock_ssh.SSH()
63 p.target = mock_ssh.SSH()
66 self.assertTrue(mock_ssh.SSH().close.called)
67 mock_ssh.SSH().execute.assert_called_with("pkill iperf3")
69 def test_iperf_successful_no_sla(self, mock_ssh):
72 args = {'options': options}
75 p = iperf3.Iperf(args, self.ctx)
76 mock_ssh.SSH().execute.return_value = (0, '', '')
77 p.host = mock_ssh.SSH()
79 sample_output = self._read_sample_output(self.output_name_tcp)
80 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
81 expected_result = json.loads(sample_output)
83 self.assertEqual(result, expected_result)
85 def test_iperf_successful_sla(self, mock_ssh):
90 'sla': {'bytes_per_second': 15000000}
94 p = iperf3.Iperf(args, self.ctx)
95 mock_ssh.SSH().execute.return_value = (0, '', '')
96 p.host = mock_ssh.SSH()
98 sample_output = self._read_sample_output(self.output_name_tcp)
99 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
100 expected_result = json.loads(sample_output)
102 self.assertEqual(result, expected_result)
104 def test_iperf_unsuccessful_sla(self, mock_ssh):
109 'sla': {'bytes_per_second': 25000000}
113 p = iperf3.Iperf(args, self.ctx)
114 mock_ssh.SSH().execute.return_value = (0, '', '')
115 p.host = mock_ssh.SSH()
117 sample_output = self._read_sample_output(self.output_name_tcp)
118 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
119 self.assertRaises(AssertionError, p.run, result)
121 def test_iperf_successful_sla_jitter(self, mock_ssh):
123 options = {"udp":"udp","bandwidth":"20m"}
126 'sla': {'jitter': 10}
130 p = iperf3.Iperf(args, self.ctx)
131 mock_ssh.SSH().execute.return_value = (0, '', '')
132 p.host = mock_ssh.SSH()
134 sample_output = self._read_sample_output(self.output_name_udp)
135 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
136 expected_result = json.loads(sample_output)
138 self.assertEqual(result, expected_result)
140 def test_iperf_unsuccessful_sla_jitter(self, mock_ssh):
142 options = {"udp":"udp","bandwidth":"20m"}
145 'sla': {'jitter': 0.0001}
149 p = iperf3.Iperf(args, self.ctx)
150 mock_ssh.SSH().execute.return_value = (0, '', '')
151 p.host = mock_ssh.SSH()
153 sample_output = self._read_sample_output(self.output_name_udp)
154 mock_ssh.SSH().execute.return_value = (0, sample_output, '')
155 self.assertRaises(AssertionError, p.run, result)
157 def test_iperf_unsuccessful_script_error(self, mock_ssh):
160 args = {'options': options}
163 p = iperf3.Iperf(args, self.ctx)
164 mock_ssh.SSH().execute.return_value = (0, '', '')
165 p.host = mock_ssh.SSH()
167 mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
168 self.assertRaises(RuntimeError, p.run, result)
170 def _read_sample_output(self,filename):
171 curr_path = os.path.dirname(os.path.abspath(__file__))
172 output = os.path.join(curr_path, filename)
173 with open(output) as f:
174 sample_output = f.read()
181 if __name__ == '__main__':