yardstick/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 from __future__ import absolute_import
  15
  16 import os
  17 import unittest
  18
  19 import mock
  20 from oslo_serialization import jsonutils
  21
  22 from yardstick.benchmark.scenarios.networking import netperf
  23
  24
  25 @mock.patch('yardstick.benchmark.scenarios.networking.netperf.ssh')
  26 class NetperfTestCase(unittest.TestCase):
  27
  28     def setUp(self):
  29         self.ctx = {
  30             'host': {
  31                 'ip': '172.16.0.137',
  32                 'user': 'cirros',
  33                 'key_filename': 'mykey.key'
  34             },
  35             'target': {
  36                 'ip': '172.16.0.138',
  37                 'user': 'cirros',
  38                 'key_filename': 'mykey.key',
  39                 'ipaddr': '172.16.0.138'
  40             }
  41         }
  42
  43     def test_netperf_successful_setup(self, mock_ssh):
  44
  45         p = netperf.Netperf({}, self.ctx)
  46         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
  47
  48         p.setup()
  49         self.assertIsNotNone(p.server)
  50         self.assertIsNotNone(p.client)
  51         self.assertTrue(p.setup_done)
  52
  53     def test_netperf_successful_no_sla(self, mock_ssh):
  54
  55         options = {}
  56         args = {'options': options}
  57         result = {}
  58
  59         p = netperf.Netperf(args, self.ctx)
  60         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
  61         p.host = mock_ssh.SSH.from_node()
  62
  63         sample_output = self._read_sample_output()
  64         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
  65         expected_result = jsonutils.loads(sample_output)
  66         p.run(result)
  67         self.assertEqual(result, expected_result)
  68
  69     def test_netperf_successful_sla(self, mock_ssh):
  70
  71         options = {}
  72         args = {
  73             'options': options,
  74             'sla': {'mean_latency': 100}
  75         }
  76         result = {}
  77
  78         p = netperf.Netperf(args, self.ctx)
  79         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
  80         p.host = mock_ssh.SSH.from_node()
  81
  82         sample_output = self._read_sample_output()
  83         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
  84         expected_result = jsonutils.loads(sample_output)
  85         p.run(result)
  86         self.assertEqual(result, expected_result)
  87
  88     def test_netperf_unsuccessful_sla(self, mock_ssh):
  89
  90         options = {}
  91         args = {
  92             'options': options,
  93             'sla': {'mean_latency': 5}
  94         }
  95         result = {}
  96
  97         p = netperf.Netperf(args, self.ctx)
  98         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
  99         p.host = mock_ssh.SSH.from_node()
 100
 101         sample_output = self._read_sample_output()
 102         mock_ssh.SSH.from_node().execute.return_value = (0, sample_output, '')
 103         self.assertRaises(AssertionError, p.run, result)
 104
 105     def test_netperf_unsuccessful_script_error(self, mock_ssh):
 106
 107         options = {}
 108         args = {'options': options}
 109         result = {}
 110
 111         p = netperf.Netperf(args, self.ctx)
 112         mock_ssh.SSH.from_node().execute.return_value = (0, '', '')
 113         p.host = mock_ssh.SSH.from_node()
 114
 115         mock_ssh.SSH.from_node().execute.return_value = (1, '', 'FOOBAR')
 116         self.assertRaises(RuntimeError, p.run, result)
 117
 118     def _read_sample_output(self):
 119         curr_path = os.path.dirname(os.path.abspath(__file__))
 120         output = os.path.join(curr_path, 'netperf_sample_output.json')
 121         with open(output) as f:
 122             sample_output = f.read()
 123         return sample_output
 124
 125
 126 def main():
 127     unittest.main()
 128
 129 if __name__ == '__main__':
 130     main()