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