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
  25     def setUp(self):
  26         self.ctx = {
  27             'host': '172.16.0.137',
  28             'target': '172.16.0.138',
  29             'user': 'cirros',
  30             'key_filename': "mykey.key"
  31         }
  32
  33     def test_iperf_successful_setup(self, mock_ssh):
  34
  35         p = iperf3.Iperf(self.ctx)
  36         mock_ssh.SSH().execute.return_value = (0, '', '')
  37
  38         p.setup()
  39         self.assertIsNotNone(p.target)
  40         self.assertIsNotNone(p.host)
  41         mock_ssh.SSH().execute.assert_called_with("iperf3 -s -D")
  42
  43     def test_iperf_unsuccessful_setup(self, mock_ssh):
  44
  45         p = iperf3.Iperf(self.ctx)
  46         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
  47         self.assertRaises(RuntimeError, p.setup)
  48
  49     def test_iperf_successful_teardown(self, mock_ssh):
  50
  51         p = iperf3.Iperf(self.ctx)
  52         mock_ssh.SSH().execute.return_value = (0, '', '')
  53         p.host = mock_ssh.SSH()
  54         p.target = mock_ssh.SSH()
  55
  56         p.teardown()
  57         self.assertTrue(mock_ssh.SSH().close.called)
  58         mock_ssh.SSH().execute.assert_called_with("pkill iperf3")
  59
  60     def test_iperf_successful_no_sla(self, mock_ssh):
  61
  62         p = iperf3.Iperf(self.ctx)
  63         mock_ssh.SSH().execute.return_value = (0, '', '')
  64         p.host = mock_ssh.SSH()
  65
  66         options = {}
  67         args = {'options': options}
  68
  69         sample_output = self._read_sample_output()
  70         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  71         expected_result = json.loads(sample_output)
  72         result = p.run(args)
  73         self.assertEqual(result, expected_result)
  74
  75     def test_iperf_successful_sla(self, mock_ssh):
  76
  77         p = iperf3.Iperf(self.ctx)
  78         mock_ssh.SSH().execute.return_value = (0, '', '')
  79         p.host = mock_ssh.SSH()
  80
  81         options = {}
  82         args = {
  83             'options': options,
  84             'sla': {'bytes_per_second': 15000000}
  85         }
  86
  87         sample_output = self._read_sample_output()
  88         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  89         expected_result = json.loads(sample_output)
  90         result = p.run(args)
  91         self.assertEqual(result, expected_result)
  92
  93     def test_iperf_unsuccessful_sla(self, mock_ssh):
  94
  95         p = iperf3.Iperf(self.ctx)
  96         mock_ssh.SSH().execute.return_value = (0, '', '')
  97         p.host = mock_ssh.SSH()
  98
  99         options = {}
 100         args = {
 101             'options': options,
 102             'sla': {'bytes_per_second': 25000000}
 103         }
 104
 105         sample_output = self._read_sample_output()
 106         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 107         self.assertRaises(AssertionError, p.run, args)
 108
 109     def test_iperf_unsuccessful_script_error(self, mock_ssh):
 110
 111         p = iperf3.Iperf(self.ctx)
 112         mock_ssh.SSH().execute.return_value = (0, '', '')
 113         p.host = mock_ssh.SSH()
 114
 115         options = {}
 116         args = {'options': options}
 117
 118         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 119         self.assertRaises(RuntimeError, p.run, args)
 120
 121     def _read_sample_output(self):
 122         curr_path = os.path.dirname(os.path.abspath(__file__))
 123         output = os.path.join(curr_path, 'iperf3_sample_output.json')
 124         with open(output) as f:
 125             sample_output = f.read()
 126         return sample_output
 127
 128
 129 def main():
 130     unittest.main()
 131
 132 if __name__ == '__main__':
 133     main()