tests/unit/benchmark/scenarios/networking/test_netperf.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2015 Huawei Technologies Co.,Ltd 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.netperf.Netperf
  13
  14 import mock
  15 import unittest
  16 import os
  17 import json
  18
  19 from yardstick.benchmark.scenarios.networking import netperf
  20
  21
  22 @mock.patch('yardstick.benchmark.scenarios.networking.netperf.ssh')
  23 class NetperfTestCase(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_netperf_successful_setup(self, mock_ssh):
  34
  35         p = netperf.Netperf(self.ctx)
  36         mock_ssh.SSH().execute.return_value = (0, '', '')
  37
  38         p.setup()
  39         self.assertIsNotNone(p.server)
  40         self.assertIsNotNone(p.client)
  41         self.assertEqual(p.setup_done, True)
  42
  43     def test_netperf_successful_no_sla(self, mock_ssh):
  44
  45         p = netperf.Netperf(self.ctx)
  46         mock_ssh.SSH().execute.return_value = (0, '', '')
  47         p.host = mock_ssh.SSH()
  48
  49         options = {}
  50         args = {'options': options}
  51         result = {}
  52
  53         sample_output = self._read_sample_output()
  54         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  55         expected_result = json.loads(sample_output)
  56         p.run(args, result)
  57         self.assertEqual(result, expected_result)
  58
  59     def test_netperf_successful_sla(self, mock_ssh):
  60
  61         p = netperf.Netperf(self.ctx)
  62         mock_ssh.SSH().execute.return_value = (0, '', '')
  63         p.host = mock_ssh.SSH()
  64
  65         options = {}
  66         args = {
  67             'options': options,
  68             'sla': {'mean_latency': 100}
  69         }
  70         result = {}
  71
  72         sample_output = self._read_sample_output()
  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_netperf_unsuccessful_sla(self, mock_ssh):
  79
  80         p = netperf.Netperf(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': {'mean_latency': 5}
  88         }
  89         result = {}
  90
  91         sample_output = self._read_sample_output()
  92         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  93         self.assertRaises(AssertionError, p.run, args, result)
  94
  95     def test_netperf_unsuccessful_script_error(self, mock_ssh):
  96
  97         p = netperf.Netperf(self.ctx)
  98         mock_ssh.SSH().execute.return_value = (0, '', '')
  99         p.host = mock_ssh.SSH()
 100
 101         options = {}
 102         args = {'options': options}
 103         result = {}
 104
 105         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 106         self.assertRaises(RuntimeError, p.run, args, result)
 107
 108     def _read_sample_output(self):
 109         curr_path = os.path.dirname(os.path.abspath(__file__))
 110         output = os.path.join(curr_path, 'netperf_sample_output.json')
 111         with open(output) as f:
 112             sample_output = f.read()
 113         return sample_output
 114
 115
 116 def main():
 117     unittest.main()
 118
 119 if __name__ == '__main__':
 120     main()