tests/unit/benchmark/scenarios/availability/test_monitor_command.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
  13 # yardstick.benchmark.scenarios.availability.monitor.monitor_command
  14
  15 from __future__ import absolute_import
  16 import mock
  17 import unittest
  18
  19 from yardstick.benchmark.scenarios.availability.monitor import monitor_command
  20
  21
  22 @mock.patch('subprocess.check_output')
  23 class ExecuteShellTestCase(unittest.TestCase):
  24
  25     def test__fun_execute_shell_command_successful(self, mock_subprocess_check_output):
  26         cmd = "env"
  27         mock_subprocess_check_output.return_value = (0, 'unittest')
  28         exitcode, _ = monitor_command._execute_shell_command(cmd)
  29         self.assertEqual(exitcode, 0)
  30
  31     @mock.patch('yardstick.benchmark.scenarios.availability.monitor.monitor_command.LOG')
  32     def test__fun_execute_shell_command_fail_cmd_exception(self, mock_log,
  33                                                            mock_subprocess_check_output):
  34         cmd = "env"
  35         mock_subprocess_check_output.side_effect = RuntimeError
  36         exitcode, _ = monitor_command._execute_shell_command(cmd)
  37         self.assertEqual(exitcode, -1)
  38         mock_log.error.assert_called_once()
  39
  40
  41 @mock.patch('subprocess.check_output')
  42 class MonitorOpenstackCmdTestCase(unittest.TestCase):
  43
  44     def setUp(self):
  45         host = {
  46             "ip": "10.20.0.5",
  47             "user": "root",
  48             "key_filename": "/root/.ssh/id_rsa"
  49         }
  50         self.context = {"node1": host}
  51         self.config = {
  52             'monitor_type': 'openstack-api',
  53             'command_name': 'nova image-list',
  54             'monitor_time': 1,
  55             'sla': {'max_outage_time': 5}
  56         }
  57
  58     def test__monitor_command_monitor_func_successful(self, mock_subprocess_check_output):
  59
  60         instance = monitor_command.MonitorOpenstackCmd(self.config, None, {"nova-api": 10})
  61         instance.setup()
  62         mock_subprocess_check_output.return_value = (0, 'unittest')
  63         ret = instance.monitor_func()
  64         self.assertTrue(ret)
  65         instance._result = {"outage_time": 0}
  66         instance.verify_SLA()
  67
  68     @mock.patch('yardstick.benchmark.scenarios.availability.monitor.monitor_command.LOG')
  69     def test__monitor_command_monitor_func_failure(self, mock_log, mock_subprocess_check_output):
  70         mock_subprocess_check_output.return_value = (1, 'unittest')
  71         instance = monitor_command.MonitorOpenstackCmd(self.config, None, {"nova-api": 10})
  72         instance.setup()
  73         mock_subprocess_check_output.side_effect = RuntimeError
  74         ret = instance.monitor_func()
  75         self.assertFalse(ret)
  76         mock_log.error.assert_called_once()
  77         instance._result = {"outage_time": 10}
  78         instance.verify_SLA()
  79
  80     @mock.patch(
  81         'yardstick.benchmark.scenarios.availability.monitor.monitor_command'
  82         '.ssh')
  83     def test__monitor_command_ssh_monitor_successful(self, mock_ssh, mock_subprocess_check_output):
  84
  85         mock_subprocess_check_output.return_value = (0, 'unittest')
  86         self.config["host"] = "node1"
  87         instance = monitor_command.MonitorOpenstackCmd(
  88             self.config, self.context, {"nova-api": 10})
  89         instance.setup()
  90         mock_ssh.SSH.from_node().execute.return_value = (0, "0", '')
  91         ret = instance.monitor_func()
  92         self.assertTrue(ret)