yardstick/tests/unit/benchmark/scenarios/networking/test_iperf3.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2015 Ericsson AB and others.
   5 #
   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 ##############################################################################
  11
  12 # Unittest for yardstick.benchmark.scenarios.networking.iperf3.Iperf
  13
  14 from __future__ import absolute_import
  15
  16 import os
  17 import unittest
  18
  19 import mock
  20 from oslo_serialization import jsonutils
  21
  22 from yardstick.common import utils
  23 from yardstick.benchmark.scenarios.networking import iperf3
  24
  25
  26 @mock.patch('yardstick.benchmark.scenarios.networking.iperf3.ssh')
  27 class IperfTestCase(unittest.TestCase):
  28     output_name_tcp = 'iperf3_sample_output.json'
  29     output_name_udp = 'iperf3_sample_output_udp.json'
  30
  31     def setUp(self):
  32         self.ctx = {
  33             'host': {
  34                 'ip': '172.16.0.137',
  35                 'user': 'root',
  36                 'key_filename': 'mykey.key'
  37             },
  38             'target': {
  39                 'ip': '172.16.0.138',
  40                 'user': 'root',
  41                 'key_filename': 'mykey.key',
  42                 'ipaddr': '172.16.0.138',
  43             }
  44         }
  45
  46     def test_iperf_successful_setup(self, mock_ssh):
  47
  48         p = iperf3.Iperf({}, self.ctx)
  49         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
  50
  51         p.setup()
  52         self.assertIsNotNone(p.target)
  53         self.assertIsNotNone(p.host)
  54         mock_ssh.SSH.from_node().execute.assert_called_with("iperf3 -s -D")
  55
  56     def test_iperf_unsuccessful_setup(self, mock_ssh):
  57
  58         p = iperf3.Iperf({}, self.ctx)
  59         mock_ssh.SSH.from_node().execute.return_value = (1, '', 'FOOBAR')
  60         self.assertRaises(RuntimeError, p.setup)
  61
  62     def test_iperf_successful_teardown(self, mock_ssh):
  63
  64         p = iperf3.Iperf({}, self.ctx)
  65         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
  66         p.host = mock_ssh.SSH.from_node()
  67         p.target = mock_ssh.SSH.from_node()
  68
  69         p.teardown()
  70         self.assertTrue(mock_ssh.SSH.from_node().close.called)
  71         mock_ssh.SSH.from_node().execute.assert_called_with("pkill iperf3")
  72
  73     def test_iperf_successful_no_sla(self, mock_ssh):
  74
  75         options = {}
  76         args = {'options': options}
  77         result = {}
  78
  79         p = iperf3.Iperf(args, self.ctx)
  80         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
  81         p.host = mock_ssh.SSH.from_node()
  82
  83         sample_output = self._read_sample_output(self.output_name_tcp)
  84         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
  85         expected_result = utils.flatten_dict_key(jsonutils.loads(sample_output))
  86         p.run(result)
  87         self.assertEqual(result, expected_result)
  88
  89     def test_iperf_successful_sla(self, mock_ssh):
  90
  91         options = {}
  92         args = {
  93             'options': options,
  94             'sla': {'bytes_per_second': 15000000}
  95         }
  96         result = {}
  97
  98         p = iperf3.Iperf(args, self.ctx)
  99         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
 100         p.host = mock_ssh.SSH.from_node()
 101
 102         sample_output = self._read_sample_output(self.output_name_tcp)
 103         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
 104         expected_result = utils.flatten_dict_key(jsonutils.loads(sample_output))
 105         p.run(result)
 106         self.assertEqual(result, expected_result)
 107
 108     def test_iperf_unsuccessful_sla(self, mock_ssh):
 109
 110         options = {}
 111         args = {
 112             'options': options,
 113             'sla': {'bytes_per_second': 25000000}
 114         }
 115         result = {}
 116
 117         p = iperf3.Iperf(args, self.ctx)
 118         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
 119         p.host = mock_ssh.SSH.from_node()
 120
 121         sample_output = self._read_sample_output(self.output_name_tcp)
 122         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
 123         self.assertRaises(AssertionError, p.run, result)
 124
 125     def test_iperf_successful_sla_jitter(self, mock_ssh):
 126         options = {"protocol": "udp", "bandwidth": "20m"}
 127         args = {
 128             'options': options,
 129             'sla': {'jitter': 10}
 130         }
 131         result = {}
 132
 133         p = iperf3.Iperf(args, self.ctx)
 134         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
 135         p.host = mock_ssh.SSH.from_node()
 136
 137         sample_output = self._read_sample_output(self.output_name_udp)
 138         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
 139         expected_result = utils.flatten_dict_key(jsonutils.loads(sample_output))
 140         p.run(result)
 141         self.assertEqual(result, expected_result)
 142
 143     def test_iperf_unsuccessful_sla_jitter(self, mock_ssh):
 144         options = {"protocol": "udp", "bandwidth": "20m"}
 145         args = {
 146             'options': options,
 147             'sla': {'jitter': 0.0001}
 148         }
 149         result = {}
 150
 151         p = iperf3.Iperf(args, self.ctx)
 152         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
 153         p.host = mock_ssh.SSH.from_node()
 154
 155         sample_output = self._read_sample_output(self.output_name_udp)
 156         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
 157         self.assertRaises(AssertionError, p.run, result)
 158
 159     def test_iperf_successful_tcp_protocal(self, mock_ssh):
 160         options = {"protocol": "tcp", "nodelay": "yes"}
 161         args = {
 162             'options': options,
 163             'sla': {'bytes_per_second': 15000000}
 164         }
 165         result = {}
 166
 167         p = iperf3.Iperf(args, self.ctx)
 168         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
 169         p.host = mock_ssh.SSH.from_node()
 170
 171         sample_output = self._read_sample_output(self.output_name_tcp)
 172         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
 173         expected_result = utils.flatten_dict_key(jsonutils.loads(sample_output))
 174         p.run(result)
 175         self.assertEqual(result, expected_result)
 176
 177     def test_iperf_unsuccessful_script_error(self, mock_ssh):
 178
 179         options = {}
 180         args = {'options': options}
 181         result = {}
 182
 183         p = iperf3.Iperf(args, self.ctx)
 184         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
 185         p.host = mock_ssh.SSH.from_node()
 186
 187         mock_ssh.SSH.from_node().execute.return_value = (1, '', 'FOOBAR')
 188         self.assertRaises(RuntimeError, p.run, result)
 189
 190     def _read_sample_output(self, filename):
 191         curr_path = os.path.dirname(os.path.abspath(__file__))
 192         output = os.path.join(curr_path, filename)
 193         with open(output) as f:
 194             sample_output = f.read()
 195         return sample_output
 196
 197
 198 def main():
 199     unittest.main()
 200
 201 if __name__ == '__main__':
 202     main()