999 lines
44 KiB
Python
999 lines
44 KiB
Python
# Copyright (c) 2017 Citrix Systems, Inc
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
try:
|
|
import json
|
|
except ImportError:
|
|
import simplejson as json
|
|
import mock
|
|
from mock import call
|
|
from os_xenapi.tests.plugins import plugin_test
|
|
import time
|
|
|
|
|
|
class FakeXenAPIException(Exception):
|
|
pass
|
|
|
|
|
|
class XenHostRunCmdTestCase(plugin_test.PluginTestBase):
|
|
def setUp(self):
|
|
super(XenHostRunCmdTestCase, self).setUp()
|
|
self.host = self.load_plugin("xenhost.py")
|
|
self.pluginlib = self.load_plugin("dom0_pluginlib.py")
|
|
|
|
def test_run_command(self):
|
|
self.mock_patch_object(self.host.utils,
|
|
'run_command',
|
|
'fake_run_cmd_return')
|
|
cmd_result = self.host._run_command('fake_command')
|
|
|
|
self.assertEqual(cmd_result, 'fake_run_cmd_return')
|
|
self.host.utils.run_command.assert_called_with(
|
|
'fake_command', cmd_input=None)
|
|
|
|
def test_run_command_exception(self):
|
|
side_effect = [self.host.utils.SubprocessException(
|
|
'fake_cmdline', 0,
|
|
'fake_out', 'fake_err')]
|
|
self.mock_patch_object(self.host.utils,
|
|
'run_command',
|
|
'fake_run_cmd_return')
|
|
self.host.utils.run_command.side_effect = side_effect
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host._run_command,
|
|
'fake_command')
|
|
self.host.utils.run_command.assert_called_with(
|
|
'fake_command', cmd_input=None)
|
|
|
|
|
|
class VMOperationTestCase(plugin_test.PluginTestBase):
|
|
def setUp(self):
|
|
super(VMOperationTestCase, self).setUp()
|
|
self.host = self.load_plugin("xenhost.py")
|
|
self.pluginlib = self.load_plugin("dom0_pluginlib.py")
|
|
self.mock_patch_object(self.host,
|
|
'_run_command',
|
|
'fake_run_cmd_return')
|
|
|
|
@mock.patch.object(time, 'sleep')
|
|
def test_resume_compute(self, mock_sleep):
|
|
self.mock_patch_object(self.session.xenapi.VM,
|
|
'start')
|
|
self.host._resume_compute(self.session,
|
|
'fake_compute_ref',
|
|
'fake_compute_uuid')
|
|
|
|
self.session.xenapi.VM.start.assert_called_with(
|
|
'fake_compute_ref', False, True)
|
|
mock_sleep.time.assert_not_called()
|
|
|
|
@mock.patch.object(time, 'sleep')
|
|
def test_resume_compute_exception_compute_VM_restart(self, mock_sleep):
|
|
side_effect_xenapi_failure = FakeXenAPIException
|
|
self.mock_patch_object(self.session.xenapi.VM,
|
|
'start')
|
|
self.host.XenAPI.Failure = FakeXenAPIException
|
|
self.session.xenapi.VM.start.side_effect = \
|
|
side_effect_xenapi_failure
|
|
|
|
self.host._resume_compute(self.session,
|
|
'fake_compute_ref',
|
|
'fake_compute_uuid')
|
|
self.session.xenapi.VM.start.assert_called_with(
|
|
'fake_compute_ref', False, True)
|
|
self.host._run_command.assert_called_with(
|
|
["xe", "vm-start", "uuid=%s" % 'fake_compute_uuid']
|
|
)
|
|
mock_sleep.assert_not_called()
|
|
|
|
@mock.patch.object(time, 'sleep')
|
|
def test_resume_compute_exception_wait_slave_available(self, mock_sleep):
|
|
side_effect_xenapi_failure = FakeXenAPIException
|
|
side_effect_plugin_error = [self.pluginlib.PluginError(
|
|
"Wait for the slave to become available"),
|
|
None]
|
|
self.mock_patch_object(self.session.xenapi.VM,
|
|
'start')
|
|
self.session.xenapi.VM.start.side_effect = \
|
|
side_effect_xenapi_failure
|
|
self.host._run_command.side_effect = side_effect_plugin_error
|
|
self.host.XenAPI.Failure = FakeXenAPIException
|
|
expected = [call(["xe", "vm-start", "uuid=%s" % 'fake_compute_uuid']),
|
|
call(["xe", "vm-start", "uuid=%s" % 'fake_compute_uuid'])]
|
|
|
|
self.host._resume_compute(self.session,
|
|
'fake_compute_ref',
|
|
'fake_compute_uuid')
|
|
self.session.xenapi.VM.start.assert_called_with(
|
|
'fake_compute_ref', False, True)
|
|
self.assertEqual(expected, self.host._run_command.call_args_list)
|
|
mock_sleep.assert_called_once()
|
|
|
|
@mock.patch.object(time, 'sleep')
|
|
def test_resume_compute_exception_unrecoverable(self, mock_sleep):
|
|
fake_compute_ref = -1
|
|
side_effect_xenapi_failure = FakeXenAPIException
|
|
side_effect_plugin_error = (
|
|
[self.pluginlib.PluginError] *
|
|
self.host.DEFAULT_TRIES)
|
|
self.mock_patch_object(self.session.xenapi.VM,
|
|
'start')
|
|
self.session.xenapi.VM.start.side_effect = \
|
|
side_effect_xenapi_failure
|
|
self.host.XenAPI.Failure = FakeXenAPIException
|
|
self.host._run_command.side_effect = side_effect_plugin_error
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host._resume_compute,
|
|
self.session,
|
|
fake_compute_ref,
|
|
'fake_compute_uuid')
|
|
self.session.xenapi.VM.start.assert_called_with(
|
|
-1, False, True)
|
|
self.host._run_command.assert_called_with(
|
|
["xe", "vm-start", "uuid=%s" % 'fake_compute_uuid']
|
|
)
|
|
mock_sleep.assert_called()
|
|
|
|
def test_set_host_enabled_no_enabled_key_in_arg_dict(self):
|
|
temp_dict = {}
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host.set_host_enabled,
|
|
self.host, temp_dict)
|
|
|
|
def test_set_host_enabled_unexpected_enabled_key(self):
|
|
temp_dict = {}
|
|
temp_dict.update({'enabled': 'unexpected_status'})
|
|
temp_dict.update({'host_uuid': 'fake_host_uuid'})
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host.set_host_enabled,
|
|
self.host, temp_dict)
|
|
|
|
def test_set_host_enabled_host_enable_disable_cmd_return_not_empty(self):
|
|
temp_dict = {}
|
|
temp_dict.update({'enabled': 'true'})
|
|
temp_dict.update({'host_uuid': 'fake_host_uuid'})
|
|
fake_run_command_return = 'not empty'
|
|
self.host._run_command.return_value = fake_run_command_return
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host.set_host_enabled,
|
|
self.host, temp_dict)
|
|
self.host._run_command.assert_called_once
|
|
|
|
def test_set_host_enabled_request_host_enabled(self):
|
|
temp_dict = {}
|
|
side_effects = ['', 'any_value']
|
|
temp_dict.update({'enabled': 'true'})
|
|
temp_dict.update({'host_uuid': 'fake_host_uuid'})
|
|
expected = [call(['xe', 'host-enable', 'uuid=fake_host_uuid']),
|
|
call(['xe', 'host-param-get', 'uuid=fake_host_uuid',
|
|
'param-name=enabled'])]
|
|
self.host._run_command.side_effect = side_effects
|
|
self.host.set_host_enabled(self.host, temp_dict)
|
|
self.assertEqual(self.host._run_command.call_args_list, expected)
|
|
|
|
def test_set_host_enabled_request_cmd_host_disable(self):
|
|
temp_dict = {}
|
|
side_effects = ['', 'any_value']
|
|
temp_dict.update({'enabled': 'false'})
|
|
temp_dict.update({'host_uuid': 'fake_host_uuid'})
|
|
expected = [call(["xe", "host-disable", "uuid=%s" %
|
|
temp_dict['host_uuid']],),
|
|
call(["xe", "host-param-get", "uuid=%s"
|
|
% temp_dict['host_uuid'],
|
|
"param-name=enabled"],)]
|
|
self.host._run_command.side_effect = side_effects
|
|
self.host.set_host_enabled(self.host, temp_dict)
|
|
self.assertEqual(self.host._run_command.call_args_list, expected)
|
|
|
|
def test_set_host_enabled_confirm_host_enabled(self):
|
|
temp_dict = {}
|
|
side_effects = ['', 'true']
|
|
temp_dict.update({'enabled': 'true'})
|
|
temp_dict.update({'host_uuid': 'fake_host_uuid'})
|
|
self.host._run_command.side_effect = side_effects
|
|
result_status = self.host.set_host_enabled(self.host, temp_dict)
|
|
self.assertEqual(result_status, '{"status": "enabled"}')
|
|
|
|
def test_set_host_enabled_confirm_host_disabled(self):
|
|
temp_dict = {}
|
|
side_effects = ['', 'any_value']
|
|
temp_dict.update({'enabled': 'false'})
|
|
temp_dict.update({'host_uuid': 'fake_host_uuid'})
|
|
self.host._run_command.side_effect = side_effects
|
|
result_status = self.host.set_host_enabled(self.host, temp_dict)
|
|
self.assertEqual(result_status, '{"status": "disabled"}')
|
|
|
|
|
|
class HostOptTestCase(plugin_test.PluginTestBase):
|
|
def setUp(self):
|
|
super(HostOptTestCase, self).setUp()
|
|
self.host = self.load_plugin("xenhost.py")
|
|
self.pluginlib = self.load_plugin("dom0_pluginlib.py")
|
|
self.mock_patch_object(self.host,
|
|
'_run_command',
|
|
'fake_run_cmd_return')
|
|
|
|
# may be removed because the operation would be deprecate
|
|
def test_power_action_disable_cmd_result_not_empty(self):
|
|
temp_arg_dict = {'host_uuid': 'fake_host_uuid'}
|
|
self.host._run_command.return_value = 'not_empty'
|
|
expected_cmd_arg = ["xe", "host-disable", "uuid=%s" %
|
|
'fake_host_uuid']
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host._power_action,
|
|
'fake_action', temp_arg_dict)
|
|
self.host._run_command.assert_called_with(expected_cmd_arg)
|
|
|
|
# may be removed because the operation would be deprecate
|
|
def test_power_action_shutdown_cmd_result_not_empty(self):
|
|
side_effects = [None, 'not_empty']
|
|
temp_arg_dict = {'host_uuid': 'fake_host_uuid'}
|
|
self.host._run_command.side_effect = side_effects
|
|
expected_cmd_arg_list = [call(["xe", "host-disable", "uuid=%s"
|
|
% 'fake_host_uuid']),
|
|
call(["xe", "vm-shutdown", "--multiple",
|
|
"resident-on=%s" % 'fake_host_uuid'])]
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host._power_action,
|
|
'fake_action', temp_arg_dict)
|
|
self.assertEqual(self.host._run_command.call_args_list,
|
|
expected_cmd_arg_list)
|
|
|
|
# may be removed because the operation would be deprecate
|
|
def test_power_action_input_cmd_result_not_empty(self):
|
|
side_effects = [None, None, 'not_empty']
|
|
temp_arg_dict = {'host_uuid': 'fake_host_uuid'}
|
|
self.host._run_command.side_effect = side_effects
|
|
cmds = {"reboot": "host-reboot",
|
|
"startup": "host-power-on",
|
|
"shutdown": "host-shutdown"}
|
|
fake_action = 'reboot' # 'statup' and 'shutdown' should be same
|
|
expected_cmd_arg_list = [call(["xe", "host-disable", "uuid=%s"
|
|
% 'fake_host_uuid']),
|
|
call(["xe", "vm-shutdown", "--multiple",
|
|
"resident-on=%s" % 'fake_host_uuid']),
|
|
call(["xe", cmds[fake_action], "uuid=%s" %
|
|
'fake_host_uuid'])]
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host._power_action,
|
|
fake_action, temp_arg_dict)
|
|
self.assertEqual(self.host._run_command.call_args_list,
|
|
expected_cmd_arg_list)
|
|
|
|
def test_power_action(self):
|
|
temp_arg_dict = {'host_uuid': 'fake_host_uuid'}
|
|
self.host._run_command.return_value = None
|
|
cmds = {"reboot": "host-reboot",
|
|
"startup": "host-power-on",
|
|
"shutdown": "host-shutdown"}
|
|
fake_action = 'reboot' # 'statup' and 'shutdown' should be same
|
|
expected_cmd_arg_list = [call(["xe", "host-disable", "uuid=%s"
|
|
% 'fake_host_uuid']),
|
|
call(["xe", "vm-shutdown", "--multiple",
|
|
"resident-on=%s" % 'fake_host_uuid']),
|
|
call(["xe", cmds[fake_action], "uuid=%s" %
|
|
'fake_host_uuid'])]
|
|
expected_result = {"power_action": fake_action}
|
|
|
|
action_result = self.host._power_action(fake_action, temp_arg_dict)
|
|
self.assertEqual(self.host._run_command.call_args_list,
|
|
expected_cmd_arg_list)
|
|
self.assertEqual(action_result, expected_result)
|
|
|
|
def test_host_reboot(self):
|
|
fake_action = 'reboot'
|
|
self.mock_patch_object(self.host,
|
|
'_power_action',
|
|
'fake_action_result')
|
|
self.host.host_reboot(self.host, 'fake_arg_dict')
|
|
|
|
self.host._power_action.assert_called_with(fake_action,
|
|
'fake_arg_dict')
|
|
|
|
def test_host_shutdown(self):
|
|
fake_action = 'shutdown'
|
|
self.mock_patch_object(self.host,
|
|
'_power_action',
|
|
'fake_action_result')
|
|
self.host.host_shutdown(self.host, 'fake_arg_dict')
|
|
|
|
self.host._power_action.assert_called_with(fake_action,
|
|
'fake_arg_dict')
|
|
|
|
def test_host_start(self):
|
|
fake_action = 'startup'
|
|
self.mock_patch_object(self.host,
|
|
'_power_action',
|
|
'fake_action_result')
|
|
self.host.host_start(self.host, 'fake_arg_dict')
|
|
|
|
self.host._power_action.assert_called_with(fake_action,
|
|
'fake_arg_dict')
|
|
|
|
def test_host_join(self):
|
|
temp_arg_dict = {'url': 'fake_url',
|
|
'user': 'fake_user',
|
|
'password': 'fake_password',
|
|
'master_addr': 'fake_master_addr',
|
|
'master_user': 'fake_master_user',
|
|
'master_pass': 'fake_master_pass',
|
|
'compute_uuid': 'fake_compute_uuid'}
|
|
self.mock_patch_object(self.host, '_resume_compute')
|
|
self.host.XenAPI = mock.Mock()
|
|
self.host.XenAPI.Session = mock.Mock()
|
|
|
|
self.host.host_join(self.host, temp_arg_dict)
|
|
self.host.XenAPI.Session().login_with_password.assert_called_once()
|
|
self.host.XenAPI.Session().xenapi.pool.join.assert_called_with(
|
|
'fake_master_addr',
|
|
'fake_master_user',
|
|
'fake_master_pass')
|
|
self.host._resume_compute.assert_called_with(
|
|
self.host.XenAPI.Session(),
|
|
self.host.XenAPI.Session().xenapi.VM.get_by_uuid(
|
|
'fake_compute_uuid'),
|
|
'fake_compute_uuid')
|
|
|
|
def test_host_join_force_join(self):
|
|
temp_arg_dict = {'force': 'true',
|
|
'master_addr': 'fake_master_addr',
|
|
'master_user': 'fake_master_user',
|
|
'master_pass': 'fake_master_pass',
|
|
'compute_uuid': 'fake_compute_uuid'}
|
|
self.mock_patch_object(self.host, '_resume_compute')
|
|
self.host.XenAPI = mock.Mock()
|
|
self.host.XenAPI.Session = mock.Mock()
|
|
|
|
self.host.host_join(self.host, temp_arg_dict)
|
|
self.host.XenAPI.Session().login_with_password.assert_called_once()
|
|
self.host.XenAPI.Session().xenapi.pool.join_force.assert_called_with(
|
|
'fake_master_addr',
|
|
'fake_master_user',
|
|
'fake_master_pass')
|
|
self.host._resume_compute.assert_called_with(
|
|
self.host.XenAPI.Session(),
|
|
self.host.XenAPI.Session().xenapi.VM.get_by_uuid(
|
|
'fake_compute_uuid'),
|
|
'fake_compute_uuid')
|
|
|
|
def test_host_data(self):
|
|
temp_arg_dict = {'host_uuid': 'fake_host_uuid'}
|
|
fake_dict_after_cleanup = {'new_key': 'new_value'}
|
|
fake_config_setting = {'config': 'fake_config_setting'}
|
|
self.host._run_command.return_value = 'fake_resp'
|
|
self.mock_patch_object(self.host,
|
|
'parse_response',
|
|
'fake_parsed_data')
|
|
self.mock_patch_object(self.host,
|
|
'cleanup',
|
|
fake_dict_after_cleanup)
|
|
self.mock_patch_object(self.host,
|
|
'_get_config_dict',
|
|
fake_config_setting)
|
|
expected_ret_dict = fake_dict_after_cleanup
|
|
expected_ret_dict.update(fake_config_setting)
|
|
|
|
return_host_data = self.host.host_data(self.host,
|
|
temp_arg_dict)
|
|
self.host._run_command.assert_called_with(
|
|
["xe", "host-param-list", "uuid=%s" % temp_arg_dict['host_uuid']]
|
|
)
|
|
self.host.parse_response.assert_called_with('fake_resp')
|
|
self.host.cleanup('fake_parsed_data')
|
|
self.host._get_config_dict.assert_called_once()
|
|
self.assertEqual(expected_ret_dict, json.loads(return_host_data))
|
|
|
|
def test_parse_response(self):
|
|
fake_resp = 'fake_name ( fake_flag): fake_value'
|
|
expected_parsed_resp = {'fake_name': 'fake_value'}
|
|
result_data = self.host.parse_response(fake_resp)
|
|
self.assertEqual(result_data, expected_parsed_resp)
|
|
|
|
def test_parse_response_one_invalid_line(self):
|
|
fake_resp = "(exeception line)\n \
|
|
fake_name ( fake_flag): fake_value"
|
|
expected_parsed_resp = {'fake_name': 'fake_value'}
|
|
|
|
result_data = self.host.parse_response(fake_resp)
|
|
self.assertEqual(result_data, expected_parsed_resp)
|
|
|
|
def test_host_uptime(self):
|
|
self.host._run_command.return_value = 'fake_uptime'
|
|
uptime_return = self.host.host_uptime(self.host, 'fake_arg_dict')
|
|
|
|
self.assertEqual(uptime_return, '{"uptime": "fake_uptime"}')
|
|
|
|
|
|
class ConfigOptTestCase(plugin_test.PluginTestBase):
|
|
def setUp(self):
|
|
super(ConfigOptTestCase, self).setUp()
|
|
self.host = self.load_plugin("xenhost.py")
|
|
self.pluginlib = self.load_plugin("dom0_pluginlib.py")
|
|
self.mock_patch_object(self.host,
|
|
'_run_command',
|
|
'fake_run_cmd_return')
|
|
|
|
def test_get_config_no_config_key(self):
|
|
temp_dict = {'params': '{"key": "fake_key"}'}
|
|
fake_conf_dict = {}
|
|
self.mock_patch_object(self.host,
|
|
'_get_config_dict',
|
|
fake_conf_dict)
|
|
|
|
config_return = self.host.get_config(self.host, temp_dict)
|
|
self.assertEqual(json.loads(config_return), "None")
|
|
self.host._get_config_dict.assert_called_once()
|
|
|
|
def test_get_config_json(self):
|
|
temp_dict = {'params': '{"key": "fake_key"}'}
|
|
fake_conf_dict = {'fake_key': 'fake_conf_key'}
|
|
self.mock_patch_object(self.host,
|
|
'_get_config_dict',
|
|
fake_conf_dict)
|
|
config_return = self.host.get_config(self.host, temp_dict)
|
|
self.assertEqual(json.loads(config_return), 'fake_conf_key')
|
|
self.host._get_config_dict.assert_called_once()
|
|
|
|
def test_get_config_dict(self):
|
|
temp_dict = {'params': {"key": "fake_key"}}
|
|
fake_conf_dict = {'fake_key': 'fake_conf_key'}
|
|
self.mock_patch_object(self.host,
|
|
'_get_config_dict',
|
|
fake_conf_dict)
|
|
config_return = self.host.get_config(self.host, temp_dict)
|
|
self.assertEqual(json.loads(config_return), 'fake_conf_key')
|
|
self.host._get_config_dict.assert_called_once()
|
|
|
|
def test_set_config_remove_none_key(self):
|
|
temp_arg_dict = {'params': {"key": "fake_key", "value": None}}
|
|
temp_conf = {'fake_key': 'fake_value'}
|
|
self.mock_patch_object(self.host,
|
|
'_get_config_dict',
|
|
temp_conf)
|
|
self.mock_patch_object(self.host,
|
|
'_write_config_dict')
|
|
|
|
self.host.set_config(self.host, temp_arg_dict)
|
|
self.assertTrue("fake_key" not in temp_conf)
|
|
self.host._get_config_dict.assert_called_once()
|
|
self.host._write_config_dict.assert_called_with(temp_conf)
|
|
|
|
def test_set_config_overwrite_key_value(self):
|
|
temp_arg_dict = {'params': {"key": "fake_key", "value": "new_value"}}
|
|
temp_conf = {'fake_key': 'fake_value'}
|
|
self.mock_patch_object(self.host,
|
|
'_get_config_dict',
|
|
temp_conf)
|
|
self.mock_patch_object(self.host,
|
|
'_write_config_dict')
|
|
|
|
self.host.set_config(self.host, temp_arg_dict)
|
|
self.assertTrue('fake_key' in temp_conf)
|
|
self.host._get_config_dict.assert_called_once()
|
|
temp_conf.update({'fake_key': 'new_value'})
|
|
self.host._write_config_dict.assert_called_with(temp_conf)
|
|
|
|
|
|
class NetworkTestCase(plugin_test.PluginTestBase):
|
|
def setUp(self):
|
|
super(NetworkTestCase, self).setUp()
|
|
self.host = self.load_plugin("xenhost.py")
|
|
self.pluginlib = self.load_plugin("dom0_pluginlib.py")
|
|
self.mock_patch_object(self.host,
|
|
'_run_command',
|
|
'fake_run_cmd_return')
|
|
|
|
def test_ovs_add_patch_port(self):
|
|
brige_name = 'fake_brige_name'
|
|
port_name = 'fake_port_name'
|
|
peer_port_name = 'fake_peer_port_name'
|
|
side_effects = [brige_name, port_name, peer_port_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ovs-vsctl', '--', '--if-exists', 'del-port',
|
|
port_name, '--', 'add-port', brige_name,
|
|
'fake_port_name', '--', 'set', 'interface',
|
|
'fake_port_name', 'type=patch',
|
|
'options:peer=%s' % peer_port_name]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge_name'),
|
|
call('fake_args', 'port_name'),
|
|
call('fake_args', 'peer_port_name')]
|
|
|
|
self.host._ovs_add_patch_port('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_ovs_del_port(self):
|
|
bridge_name = 'fake_brige_name'
|
|
port_name = 'fake_port_name'
|
|
side_effects = [bridge_name, port_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ovs-vsctl', '--', '--if-exists', 'del-port',
|
|
bridge_name, port_name]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge_name'),
|
|
call('fake_args', 'port_name')]
|
|
|
|
self.host._ovs_del_port('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_ovs_del_br(self):
|
|
bridge_name = 'fake_brige_name'
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
bridge_name)
|
|
expected_cmd_args = ['ovs-vsctl', '--', '--if-exists',
|
|
'del-br', bridge_name]
|
|
|
|
self.host._ovs_del_br('fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'bridge_name')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
|
|
def test_ovs_set_if_external_id(self):
|
|
interface = 'fake_interface'
|
|
extneral_id = 'fake_extneral_id'
|
|
value = 'fake_value'
|
|
side_effects = [interface, extneral_id, value]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ovs-vsctl', 'set', 'Interface', interface,
|
|
'external-ids:%s=%s' % (extneral_id, value)]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'interface'),
|
|
call('fake_args', 'extneral_id'),
|
|
call('fake_args', 'value')]
|
|
|
|
self.host._ovs_set_if_external_id('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_ovs_add_port(self):
|
|
bridge_name = 'fake_brige_name'
|
|
port_name = 'fake_port_name'
|
|
side_effects = [bridge_name, port_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ovs-vsctl', '--', '--if-exists', 'del-port',
|
|
port_name, '--', 'add-port', bridge_name,
|
|
port_name]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge_name'),
|
|
call('fake_args', 'port_name')]
|
|
|
|
self.host._ovs_add_port('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_ovs_create_port(self):
|
|
bridge_name = 'fake_brige_name'
|
|
port_name = 'fake_port_name'
|
|
iface_id = 'fake_iface_id'
|
|
mac = 'fake_mac'
|
|
status = 'fake_status'
|
|
side_effects = [bridge_name, port_name, iface_id, mac, status]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ovs-vsctl', '--', '--if-exists', 'del-port',
|
|
port_name, '--', 'add-port', bridge_name,
|
|
port_name, '--', 'set', 'Interface', port_name,
|
|
'external_ids:iface-id=%s' % iface_id,
|
|
'external_ids:iface-status=%s' % status,
|
|
'external_ids:attached-mac=%s' % mac,
|
|
'external_ids:xs-vif-uuid=%s' % iface_id]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge'),
|
|
call('fake_args', 'port'),
|
|
call('fake_args', 'iface-id'),
|
|
call('fake_args', 'mac'),
|
|
call('fake_args', 'status')]
|
|
|
|
self.host._ovs_create_port('fake_args')
|
|
self.pluginlib.exists.assert_called()
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_ip_link_get_dev(self):
|
|
device_name = 'fake_device_name'
|
|
expected_cmd_args = ['ip', 'link', 'show', device_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
device_name)
|
|
|
|
self.host._ip_link_get_dev('fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'device_name')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
|
|
def test_ip_link_del_dev(self):
|
|
device_name = 'fake_device_name'
|
|
expected_cmd_args = ['ip', 'link', 'delete', device_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
'fake_device_name')
|
|
|
|
self.host._ip_link_del_dev('fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'device_name')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
|
|
def test_ip_link_add_veth_pair(self):
|
|
dev1_name = 'fake_brige_name'
|
|
dev2_name = 'fake_port_name'
|
|
side_effects = [dev1_name, dev2_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ip', 'link', 'add', dev1_name, 'type',
|
|
'veth', 'peer', 'name', dev2_name]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'dev1_name'),
|
|
call('fake_args', 'dev2_name')]
|
|
|
|
self.host._ip_link_add_veth_pair('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_ip_link_set_dev(self):
|
|
device_name = 'fake_device_name'
|
|
option = 'fake_option'
|
|
side_effects = [device_name, option]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ip', 'link', 'set', device_name, option]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'device_name'),
|
|
call('fake_args', 'option')]
|
|
|
|
self.host._ip_link_set_dev('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_ip_link_set_promisc(self):
|
|
device_name = 'fake_device_name'
|
|
option = 'fake_option'
|
|
side_effects = [device_name, option]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['ip', 'link', 'set', device_name, 'promisc',
|
|
option]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'device_name'),
|
|
call('fake_args', 'option')]
|
|
|
|
self.host._ip_link_set_promisc('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_brctl_add_br(self):
|
|
bridge_name = 'fake_bridge_name'
|
|
cmd_args = 'fake_option'
|
|
side_effects = [bridge_name, cmd_args]
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
bridge_name)
|
|
expected_cmd_args = ['brctl', 'addbr', bridge_name]
|
|
|
|
self.host._brctl_add_br('fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'bridge_name')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
|
|
def test_brctl_del_br(self):
|
|
bridge_name = 'fake_bridge_name'
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
bridge_name)
|
|
expected_cmd_args = ['brctl', 'delbr', bridge_name]
|
|
|
|
self.host._brctl_del_br('fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'bridge_name')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
|
|
def test_brctl_set_fd(self):
|
|
bridge_name = 'fake_device_name'
|
|
fd = 'fake_fd'
|
|
side_effects = [bridge_name, fd]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['brctl', 'setfd', bridge_name, fd]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge_name'),
|
|
call('fake_args', 'fd')]
|
|
|
|
self.host._brctl_set_fd('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_brctl_set_stp(self):
|
|
bridge_name = 'fake_device_name'
|
|
option = 'fake_option'
|
|
side_effects = [bridge_name, option]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['brctl', 'stp', bridge_name, option]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge_name'),
|
|
call('fake_args', 'option')]
|
|
|
|
self.host._brctl_set_stp('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_brctl_add_if(self):
|
|
bridge_name = 'fake_device_name'
|
|
if_name = 'fake_if_name'
|
|
side_effects = [bridge_name, if_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['brctl', 'addif', bridge_name, if_name]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge_name'),
|
|
call('fake_args', 'interface_name')]
|
|
|
|
self.host._brctl_add_if('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
def test_brctl_del_if(self):
|
|
bridge_name = 'fake_device_name'
|
|
if_name = 'fake_if_name'
|
|
side_effects = [bridge_name, if_name]
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
expected_cmd_args = ['brctl', 'delif', bridge_name, if_name]
|
|
expected_pluginlib_arg_list = [call('fake_args', 'bridge_name'),
|
|
call('fake_args', 'interface_name')]
|
|
|
|
self.host._brctl_del_if('fake_args')
|
|
self.host._run_command.assert_called_with(expected_cmd_args)
|
|
self.assertEqual(self.pluginlib.exists.call_args_list,
|
|
expected_pluginlib_arg_list)
|
|
|
|
@mock.patch.object(json, 'loads')
|
|
def test_iptables_config(self, mock_json_loads):
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
'fake_cmd_args')
|
|
self.mock_patch_object(self.pluginlib,
|
|
'optional',
|
|
'fake_cmd_pro_input')
|
|
self.host._run_command.return_value = 'fake_run_cmd_resule'
|
|
mock_json_loads.return_value = ['iptables-save']
|
|
expected = json.dumps(dict(out='fake_run_cmd_resule', err=''))
|
|
|
|
ret_str = self.host.iptables_config('fake_session', 'fake_args')
|
|
self.pluginlib.exists.assert_called_once()
|
|
self.pluginlib.optional.assert_called_once()
|
|
mock_json_loads.assert_called_with('fake_cmd_args')
|
|
self.assertEqual(ret_str, expected)
|
|
self.host._run_command.assert_called_with(
|
|
map(str, ['iptables-save']), 'fake_cmd_pro_input')
|
|
|
|
@mock.patch.object(json, 'loads')
|
|
def test_iptables_config_plugin_error(self, mock_json_loads):
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.mock_patch_object(self.pluginlib,
|
|
'optional')
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host.iptables_config,
|
|
'fake_session', 'fake_args')
|
|
self.pluginlib.exists.assert_called_once()
|
|
self.pluginlib.optional.assert_called_once()
|
|
mock_json_loads.assert_called_with(None)
|
|
|
|
def test_network_config_invalid_cmd(self):
|
|
fake_invalid_cmd = 0
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
fake_invalid_cmd)
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host.network_config,
|
|
'fake_session', 'fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'cmd')
|
|
|
|
def test_network_config_unexpected_cmd(self):
|
|
fake_unexpected_cmd = 'fake_unknow_cmd'
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists',
|
|
fake_unexpected_cmd)
|
|
|
|
self.assertRaises(self.pluginlib.PluginError,
|
|
self.host.network_config,
|
|
'fake_session', 'fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'cmd')
|
|
|
|
def test_network_config(self):
|
|
fake_valid_cmd = 'ovs_add_patch_port'
|
|
side_effects = [fake_valid_cmd, 'fake_cmd_args']
|
|
self.mock_patch_object(self.pluginlib,
|
|
'exists')
|
|
self.pluginlib.exists.side_effect = side_effects
|
|
mock_func = self.mock_patch_object(self.host,
|
|
'_ovs_add_patch_port')
|
|
self.host.ALLOWED_NETWORK_CMDS['ovs_add_patch_port'] = mock_func
|
|
|
|
self.host.network_config('fake_session', 'fake_args')
|
|
self.pluginlib.exists.assert_called_with('fake_args', 'args')
|
|
self.host._ovs_add_patch_port.assert_called_with('fake_cmd_args')
|
|
|
|
|
|
class XenHostTestCase(plugin_test.PluginTestBase):
|
|
def setUp(self):
|
|
super(XenHostTestCase, self).setUp()
|
|
self.host = self.load_plugin("xenhost.py")
|
|
self.pluginlib = self.load_plugin("dom0_pluginlib.py")
|
|
self.mock_patch_object(self.host,
|
|
'_run_command',
|
|
'fake_run_cmd_return')
|
|
|
|
def test_clean_up(self):
|
|
fake_arg_dict = {
|
|
'enabled': 'enabled',
|
|
'memory-total': '0',
|
|
'memory-overhead': '1',
|
|
'memory-free': '2',
|
|
'memory-free-computed': '3',
|
|
'uuid': 'fake_uuid',
|
|
'name-label': 'fake_name-label',
|
|
'name-description': 'fake_name-description',
|
|
'hostname': 'fake_hostname',
|
|
'address': 'fake_address',
|
|
'other-config': 'config:fake_other-config_1; \
|
|
config:fake_other-config_2',
|
|
'capabilities': 'fake_cap_1; fake_cap_2',
|
|
'cpu_info': 'cpu_count:1; family:101; unknow:1'
|
|
}
|
|
expected_out = {
|
|
'enabled': 'enabled',
|
|
'host_memory': {'total': 0,
|
|
'overhead': 1,
|
|
'free': 2,
|
|
'free-computed': 3},
|
|
'host_uuid': 'fake_uuid',
|
|
'host_name-label': 'fake_name-label',
|
|
'host_name-description': 'fake_name-description',
|
|
'host_hostname': 'fake_hostname',
|
|
'host_ip_address': 'fake_address',
|
|
'host_other-config': {'config': 'fake_other-config_1',
|
|
'config': 'fake_other-config_2'},
|
|
'host_capabilities': ['fake_cap_1', 'fake_cap_2'],
|
|
'host_cpu_info': {'cpu_count': 1, 'family': 101,
|
|
'unknow': '1'}
|
|
}
|
|
|
|
out = self.host.cleanup(fake_arg_dict)
|
|
self.assertEqual(out, expected_out)
|
|
|
|
def test_clean_up_exception_invalid_memory_value(self):
|
|
fake_arg_dict = {
|
|
'enabled': 'enabled',
|
|
'memory-total': 'invalid',
|
|
'memory-overhead': 'invalid',
|
|
'memory-free': 'invalid',
|
|
'memory-free-computed': 'invalid',
|
|
'uuid': 'fake_uuid',
|
|
'name-label': 'fake_name-label',
|
|
'name-description': 'fake_name-description',
|
|
'hostname': 'fake_hostname',
|
|
'address': 'fake_address',
|
|
'other-config': 'config:fake_other-config_1; \
|
|
config:fake_other-config_2',
|
|
'capabilities': 'fake_cap_1; fake_cap_2',
|
|
'cpu_info': 'cpu_count:1; family:101; unknow:1'
|
|
}
|
|
expected_out = {
|
|
'enabled': 'enabled',
|
|
'host_memory': {'total': None,
|
|
'overhead': None,
|
|
'free': None,
|
|
'free-computed': None},
|
|
'host_uuid': 'fake_uuid',
|
|
'host_name-label': 'fake_name-label',
|
|
'host_name-description': 'fake_name-description',
|
|
'host_hostname': 'fake_hostname',
|
|
'host_ip_address': 'fake_address',
|
|
'host_other-config': {'config': 'fake_other-config_1',
|
|
'config': 'fake_other-config_2'},
|
|
'host_capabilities': ['fake_cap_1', 'fake_cap_2'],
|
|
'host_cpu_info': {'cpu_count': 1, 'family': 101,
|
|
'unknow': '1'}
|
|
}
|
|
|
|
out = self.host.cleanup(fake_arg_dict)
|
|
self.assertEqual(out, expected_out)
|
|
|
|
def test_query_gc_running(self):
|
|
fake_cmd_result = "Currently running: True"
|
|
self.host._run_command.return_value = fake_cmd_result
|
|
query_gc_result = self.host.query_gc('fake_session',
|
|
'fake_sr_uuid',
|
|
'fake_vdi_uuid')
|
|
self.assertTrue(query_gc_result)
|
|
self.host._run_command.assert_called_with(
|
|
["/opt/xensource/sm/cleanup.py",
|
|
"-q", "-u", 'fake_sr_uuid'])
|
|
|
|
def test_query_gc_not_running(self):
|
|
fake_cmd_result = "Currently running: False"
|
|
self.host._run_command.return_value = fake_cmd_result
|
|
query_gc_result = self.host.query_gc('fake_session',
|
|
'fake_sr_uuid',
|
|
'fake_vdi_uuid')
|
|
self.assertFalse(query_gc_result)
|
|
self.host._run_command.assert_called_with(
|
|
["/opt/xensource/sm/cleanup.py",
|
|
"-q", "-u", 'fake_sr_uuid'])
|
|
|
|
def test_get_pci_device_details(self):
|
|
self.host.get_pci_device_details('fake_session')
|
|
self.host._run_command.assert_called_with(
|
|
["lspci", "-vmmnk"])
|
|
|
|
def test_get_pci_type_no_domain(self):
|
|
fake_pci_device = '00:00.0'
|
|
self.host._run_command.return_value = ['fake_pci_type', ]
|
|
|
|
self.host.get_pci_type('fake_session', fake_pci_device)
|
|
self.host._run_command.assert_called_with(
|
|
["ls", "/sys/bus/pci/devices/" + '0000:' +
|
|
fake_pci_device + "/"])
|
|
|
|
def test_get_pci_type_physfn(self):
|
|
fake_pci_device = '0000:00:00.0'
|
|
self.host._run_command.return_value = ['physfn', ]
|
|
|
|
output = self.host.get_pci_type('fake_session', fake_pci_device)
|
|
self.host._run_command.assert_called_with(
|
|
["ls", "/sys/bus/pci/devices/" + fake_pci_device + "/"])
|
|
self.assertEqual(output, 'type-VF')
|
|
|
|
def test_get_pci_type_virtfn(self):
|
|
fake_pci_device = '0000:00:00.0'
|
|
self.host._run_command.return_value = ['virtfn', ]
|
|
|
|
output = self.host.get_pci_type('fake_session', fake_pci_device)
|
|
self.host._run_command.assert_called_with(
|
|
["ls", "/sys/bus/pci/devices/" + fake_pci_device + "/"])
|
|
self.assertEqual(output, 'type-PF')
|
|
|
|
def test_get_pci_type_PCI(self):
|
|
fake_pci_device = '0000:00:00.0'
|
|
self.host._run_command.return_value = ['other', ]
|
|
|
|
output = self.host.get_pci_type('fake_session', fake_pci_device)
|
|
self.host._run_command.assert_called_with(
|
|
["ls", "/sys/bus/pci/devices/" + fake_pci_device + "/"])
|
|
self.assertEqual(output, 'type-PCI')
|