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
  71         sample_output = self._read_sample_output(self.output_name_tcp)
  72         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  73         expected_result = json.loads(sample_output)
  74         result = p.run(args)
  75         self.assertEqual(result, expected_result)
  76
  77     def test_iperf_successful_sla(self, mock_ssh):
  78
  79         p = iperf3.Iperf(self.ctx)
  80         mock_ssh.SSH().execute.return_value = (0, '', '')
  81         p.host = mock_ssh.SSH()
  82
  83         options = {}
  84         args = {
  85             'options': options,
  86             'sla': {'bytes_per_second': 15000000}
  87         }
  88
  89         sample_output = self._read_sample_output(self.output_name_tcp)
  90         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  91         expected_result = json.loads(sample_output)
  92         result = p.run(args)
  93         self.assertEqual(result, expected_result)
  94
  95     def test_iperf_unsuccessful_sla(self, mock_ssh):
  96
  97         p = iperf3.Iperf(self.ctx)
  98         mock_ssh.SSH().execute.return_value = (0, '', '')
  99         p.host = mock_ssh.SSH()
 100
 101         options = {}
 102         args = {
 103             'options': options,
 104             'sla': {'bytes_per_second': 25000000}
 105         }
 106
 107         sample_output = self._read_sample_output(self.output_name_tcp)
 108         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 109         self.assertRaises(AssertionError, p.run, args)
 110
 111     def test_iperf_successful_sla_jitter(self, mock_ssh):
 112
 113         p = iperf3.Iperf(self.ctx)
 114         mock_ssh.SSH().execute.return_value = (0, '', '')
 115         p.host = mock_ssh.SSH()
 116
 117         options = {"udp":"udp","bandwidth":"20m"}
 118         args = {
 119             'options': options,
 120             'sla': {'jitter': 10}
 121         }
 122
 123         sample_output = self._read_sample_output(self.output_name_udp)
 124         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 125         expected_result = json.loads(sample_output)
 126         result = p.run(args)
 127         self.assertEqual(result, expected_result)
 128
 129     def test_iperf_unsuccessful_sla_jitter(self, mock_ssh):
 130
 131         p = iperf3.Iperf(self.ctx)
 132         mock_ssh.SSH().execute.return_value = (0, '', '')
 133         p.host = mock_ssh.SSH()
 134
 135         options = {"udp":"udp","bandwidth":"20m"}
 136         args = {
 137             'options': options,
 138             'sla': {'jitter': 0.0001}
 139         }
 140
 141         sample_output = self._read_sample_output(self.output_name_udp)
 142         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 143         self.assertRaises(AssertionError, p.run, args)
 144
 145     def test_iperf_unsuccessful_script_error(self, mock_ssh):
 146
 147         p = iperf3.Iperf(self.ctx)
 148         mock_ssh.SSH().execute.return_value = (0, '', '')
 149         p.host = mock_ssh.SSH()
 150
 151         options = {}
 152         args = {'options': options}
 153
 154         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 155         self.assertRaises(RuntimeError, p.run, args)
 156
 157     def _read_sample_output(self,filename):
 158         curr_path = os.path.dirname(os.path.abspath(__file__))
 159         output = os.path.join(curr_path, filename)
 160         with open(output) as f:
 161             sample_output = f.read()
 162         return sample_output
 163
 164
 165 def main():
 166     unittest.main()
 167
 168 if __name__ == '__main__':
 169     main()