yardstick/tests/unit/benchmark/scenarios/availability/test_monitor_command.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import mock
  11 import unittest
  12
  13 from yardstick.benchmark.scenarios.availability.monitor import monitor_command
  14
  15
  16 class ExecuteShellTestCase(unittest.TestCase):
  17
  18     def setUp(self):
  19         self._mock_subprocess = mock.patch.object(monitor_command, 'subprocess')
  20         self.mock_subprocess = self._mock_subprocess.start()
  21         self.addCleanup(self._stop_mock)
  22
  23     def _stop_mock(self):
  24         self._mock_subprocess.stop()
  25
  26     def test__fun_execute_shell_command_successful(self):
  27         cmd = "env"
  28         self.mock_subprocess.check_output.return_value = (0, 'unittest')
  29         exitcode, _t = monitor_command._execute_shell_command(cmd)
  30         self.assertEqual(exitcode, 0)
  31
  32     @mock.patch.object(monitor_command, 'LOG')
  33     def test__fun_execute_shell_command_fail_cmd_exception(self, mock_log):
  34         cmd = "env"
  35         self.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 class MonitorOpenstackCmdTestCase(unittest.TestCase):
  42
  43     def setUp(self):
  44         host = {
  45             "ip": "10.20.0.5",
  46             "user": "root",
  47             "key_filename": "/root/.ssh/id_rsa"
  48         }
  49         self.context = {"node1": host}
  50         self.config = {
  51             'monitor_type': 'openstack-api',
  52             'command_name': 'nova image-list',
  53             'monitor_time': 1,
  54             'sla': {'max_outage_time': 5}
  55         }
  56         self._mock_subprocess = mock.patch.object(monitor_command, 'subprocess')
  57         self.mock_subprocess = self._mock_subprocess.start()
  58         self.addCleanup(self._stop_mock)
  59
  60     def _stop_mock(self):
  61         self._mock_subprocess.stop()
  62
  63     def test__monitor_command_monitor_func_successful(self):
  64
  65         instance = monitor_command.MonitorOpenstackCmd(self.config, None, {"nova-api": 10})
  66         instance.setup()
  67         self.mock_subprocess.check_output.return_value = (0, 'unittest')
  68         ret = instance.monitor_func()
  69         self.assertTrue(ret)
  70         instance._result = {"outage_time": 0}
  71         instance.verify_SLA()
  72
  73     @mock.patch.object(monitor_command, 'LOG')
  74     def test__monitor_command_monitor_func_failure(self, mock_log):
  75         self.mock_subprocess.check_output.return_value = (1, 'unittest')
  76         instance = monitor_command.MonitorOpenstackCmd(self.config, None, {"nova-api": 10})
  77         instance.setup()
  78         self.mock_subprocess.check_output.side_effect = RuntimeError
  79         ret = instance.monitor_func()
  80         self.assertFalse(ret)
  81         mock_log.error.assert_called_once()
  82         instance._result = {"outage_time": 10}
  83         instance.verify_SLA()
  84
  85     @mock.patch.object(monitor_command, 'ssh')
  86     def test__monitor_command_ssh_monitor_successful(self, mock_ssh):
  87
  88         self.mock_subprocess.check_output.return_value = (0, 'unittest')
  89         self.config["host"] = "node1"
  90         instance = monitor_command.MonitorOpenstackCmd(
  91             self.config, self.context, {"nova-api": 10})
  92         instance.setup()
  93         mock_ssh.SSH.from_node().execute.return_value = (0, "0", '')
  94         ret = instance.monitor_func()
  95         self.assertTrue(ret)