849a063a02
https://wiki.openstack.org/wiki/LegalIssuesFAQ said about copyrights, this patch is to change the new added files' copyrights from 'OpenStack Foundation' to 'Citrix Systems' Change-Id: Iec9dab76bf9906d865c54c3c9a2df0eb1fb73761
201 lines
8.3 KiB
Python
201 lines
8.3 KiB
Python
# Copyright (c) 2016 Citrix Systems
|
|
# 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.
|
|
|
|
import mock
|
|
import os
|
|
from os_xenapi.tests.plugins import plugin_test
|
|
import time
|
|
|
|
|
|
class FakeUnplugException(Exception):
|
|
def __init__(self, details):
|
|
self.details = details
|
|
|
|
|
|
class PluginlibDom0(plugin_test.PluginTestBase):
|
|
def setUp(self):
|
|
super(PluginlibDom0, self).setUp()
|
|
self.dom0_pluginlib = self.load_plugin("dom0_pluginlib.py")
|
|
|
|
@mock.patch.object(os.path, 'exists')
|
|
def test_configure_logging_log_dir_not_exist(self, mock_path_exist):
|
|
name = 'fake_name'
|
|
mock_Logger_setLevel = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.Logger, 'setLevel')
|
|
mock_sysh_setLevel = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.handlers.SysLogHandler, 'setLevel')
|
|
mock_Formatter = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging, 'Formatter')
|
|
mock_sysh_setFormatter = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.handlers.SysLogHandler, 'setFormatter')
|
|
mock_Logger_addHandler = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.Logger, 'addHandler')
|
|
mock_socket = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.handlers.SysLogHandler,
|
|
'_connect_unixsocket')
|
|
mock_path_exist.return_value = False
|
|
|
|
self.dom0_pluginlib.configure_logging(name)
|
|
|
|
self.assertTrue(mock_Logger_setLevel.called)
|
|
self.assertFalse(mock_sysh_setLevel.called)
|
|
self.assertFalse(mock_Formatter.called)
|
|
self.assertFalse(mock_sysh_setFormatter.called)
|
|
self.assertFalse(mock_Logger_addHandler.called)
|
|
self.assertFalse(mock_socket.called)
|
|
|
|
@mock.patch.object(os.path, 'exists')
|
|
def test_configure_logging_log(self, mock_path_exist):
|
|
name = 'fake_name'
|
|
mock_Logger_setLevel = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.Logger, 'setLevel')
|
|
mock_sysh_setLevel = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.handlers.SysLogHandler, 'setLevel')
|
|
mock_Formatter = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging, 'Formatter')
|
|
mock_sysh_setFormatter = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.handlers.SysLogHandler, 'setFormatter')
|
|
mock_Logger_addHandler = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.Logger, 'addHandler')
|
|
mock_socket = self.mock_patch_object(
|
|
self.dom0_pluginlib.logging.handlers.SysLogHandler,
|
|
'_connect_unixsocket')
|
|
mock_path_exist.return_value = True
|
|
|
|
self.dom0_pluginlib.configure_logging(name)
|
|
|
|
self.assertTrue(mock_Logger_setLevel.called)
|
|
self.assertTrue(mock_sysh_setLevel.called)
|
|
self.assertTrue(mock_Formatter.called)
|
|
self.assertTrue(mock_sysh_setFormatter.called)
|
|
self.assertTrue(mock_Logger_addHandler.called)
|
|
self.assertTrue(mock_socket.called)
|
|
|
|
def test_exists_ok(self):
|
|
fake_args = {'k1': 'v1'}
|
|
self.assertEqual('v1', self.dom0_pluginlib.exists(fake_args, 'k1'))
|
|
|
|
def test_exists_exception(self):
|
|
fake_args = {'k1': 'v1'}
|
|
self.assertRaises(self.dom0_pluginlib.ArgumentError,
|
|
self.dom0_pluginlib.exists,
|
|
fake_args,
|
|
'no_key')
|
|
|
|
def test_optional_exist(self):
|
|
fake_args = {'k1': 'v1'}
|
|
self.assertEqual('v1',
|
|
self.dom0_pluginlib.optional(fake_args, 'k1'))
|
|
|
|
def test_optional_none(self):
|
|
fake_args = {'k1': 'v1'}
|
|
self.assertIsNone(self.dom0_pluginlib.optional(fake_args,
|
|
'no_key'))
|
|
|
|
def test_get_domain_0(self):
|
|
mock_get_this_host = self.mock_patch_object(
|
|
self.session.xenapi.session,
|
|
'get_this_host',
|
|
return_val='fake_host_ref')
|
|
mock_get_vm_records = self.mock_patch_object(
|
|
self.session.xenapi.VM,
|
|
'get_all_records_where',
|
|
return_val={"fake_vm_ref": "fake_value"})
|
|
|
|
ret_value = self.dom0_pluginlib._get_domain_0(self.session)
|
|
|
|
self.assertTrue(mock_get_this_host.called)
|
|
self.assertTrue(mock_get_vm_records.called)
|
|
self.assertEqual('fake_vm_ref', ret_value)
|
|
|
|
def test_with_vdi_in_dom0(self):
|
|
self.mock_patch_object(
|
|
self.dom0_pluginlib,
|
|
'_get_domain_0',
|
|
return_val='fake_dom0_ref')
|
|
mock_vbd_create = self.mock_patch_object(
|
|
self.session.xenapi.VBD,
|
|
'create',
|
|
return_val='fake_vbd_ref')
|
|
mock_vbd_plug = self.mock_patch_object(
|
|
self.session.xenapi.VBD,
|
|
'plug')
|
|
self.mock_patch_object(
|
|
self.session.xenapi.VBD,
|
|
'get_device',
|
|
return_val='fake_device_xvda')
|
|
mock_vbd_unplug_with_retry = self.mock_patch_object(
|
|
self.dom0_pluginlib,
|
|
'_vbd_unplug_with_retry')
|
|
mock_vbd_destroy = self.mock_patch_object(
|
|
self.session.xenapi.VBD,
|
|
'destroy')
|
|
|
|
def handle_function(vbd):
|
|
# the fake vbd handle function
|
|
self.assertEqual(vbd, 'fake_device_xvda')
|
|
self.assertTrue(mock_vbd_plug.called)
|
|
self.assertFalse(mock_vbd_unplug_with_retry.called)
|
|
return 'function_called'
|
|
|
|
fake_vdi = 'fake_vdi'
|
|
return_value = self.dom0_pluginlib.with_vdi_in_dom0(
|
|
self.session, fake_vdi, False, handle_function)
|
|
|
|
self.assertEqual('function_called', return_value)
|
|
self.assertTrue(mock_vbd_plug.called)
|
|
self.assertTrue(mock_vbd_unplug_with_retry.called)
|
|
self.assertTrue(mock_vbd_destroy.called)
|
|
args, kwargs = mock_vbd_create.call_args
|
|
self.assertEqual('fake_dom0_ref', args[0]['VM'])
|
|
self.assertEqual('RW', args[0]['mode'])
|
|
|
|
def test_vbd_unplug_with_retry_success_at_first_time(self):
|
|
self.dom0_pluginlib._vbd_unplug_with_retry(self.session,
|
|
'fake_vbd_ref')
|
|
self.assertEqual(1, self.session.xenapi.VBD.unplug.call_count)
|
|
|
|
def test_vbd_unplug_with_retry_detached_already(self):
|
|
error = FakeUnplugException(['DEVICE_ALREADY_DETACHED'])
|
|
|
|
self.session.xenapi.VBD.unplug.side_effect = error
|
|
self.dom0_pluginlib.XenAPI.Failure = FakeUnplugException
|
|
self.dom0_pluginlib._vbd_unplug_with_retry(self.session,
|
|
'fake_vbd_ref')
|
|
self.assertEqual(1, self.session.xenapi.VBD.unplug.call_count)
|
|
|
|
def test_vbd_unplug_with_retry_success_at_second_time(self):
|
|
side_effects = [FakeUnplugException(['DEVICE_DETACH_REJECTED']),
|
|
None]
|
|
|
|
self.session.xenapi.VBD.unplug.side_effect = side_effects
|
|
self.dom0_pluginlib.XenAPI.Failure = FakeUnplugException
|
|
self.dom0_pluginlib._vbd_unplug_with_retry(self.session,
|
|
'fake_vbd_ref')
|
|
self.assertEqual(2, self.session.xenapi.VBD.unplug.call_count)
|
|
|
|
@mock.patch.object(time, 'sleep')
|
|
def test_vbd_unplug_with_retry_exceed_max_attempts(self, mock_sleep):
|
|
side_effects = ([FakeUnplugException(['DEVICE_DETACH_REJECTED'])]
|
|
* (self.dom0_pluginlib.MAX_VBD_UNPLUG_RETRIES + 1))
|
|
self.session.xenapi.VBD.unplug.side_effect = side_effects
|
|
self.dom0_pluginlib.XenAPI.Failure = FakeUnplugException
|
|
|
|
self.assertRaises(self.dom0_pluginlib.PluginError,
|
|
self.dom0_pluginlib._vbd_unplug_with_retry,
|
|
self.session, 'fake_vbd_ref')
|
|
self.assertEqual(self.dom0_pluginlib.MAX_VBD_UNPLUG_RETRIES,
|
|
self.session.xenapi.VBD.unplug.call_count)
|