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