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