789 lines
36 KiB
Python
789 lines
36 KiB
Python
# Copyright 2017 Cloudbase Solutions Srl
|
|
#
|
|
# 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 importlib
|
|
import os
|
|
import unittest
|
|
try:
|
|
import unittest.mock as mock
|
|
except ImportError:
|
|
import mock
|
|
|
|
from cloudbaseinit import conf as cloudbaseinit_conf
|
|
from cloudbaseinit import exception
|
|
from cloudbaseinit.tests import testutils
|
|
from cloudbaseinit.utils import encoding
|
|
|
|
CONF = cloudbaseinit_conf.CONF
|
|
MODPATH = "cloudbaseinit.metadata.services.azureservice.AzureService"
|
|
|
|
|
|
class AzureServiceTest(unittest.TestCase):
|
|
|
|
@mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
|
|
def setUp(self, mock_osutils):
|
|
self._mock_osutils = mock_osutils
|
|
self._mock_untangle = mock.MagicMock()
|
|
self._mock_ctypes = mock.MagicMock()
|
|
self._mock_wintypes = mock.MagicMock()
|
|
self._moves_mock = mock.MagicMock()
|
|
|
|
self._module_patcher = mock.patch.dict(
|
|
'sys.modules',
|
|
{'untangle': self._mock_untangle,
|
|
'ctypes': self._mock_ctypes,
|
|
'ctypes.wintypes': self._mock_wintypes,
|
|
'six.moves': self._moves_mock
|
|
})
|
|
self._module_patcher.start()
|
|
self._azureservice_module = importlib.import_module(
|
|
'cloudbaseinit.metadata.services.azureservice')
|
|
|
|
self._azureservice = self._azureservice_module.AzureService()
|
|
self._logsnatcher = testutils.LogSnatcher(
|
|
'cloudbaseinit.metadata.services.azureservice')
|
|
|
|
def tearDown(self):
|
|
self._module_patcher.stop()
|
|
|
|
@mock.patch('time.sleep')
|
|
@mock.patch('socket.inet_ntoa')
|
|
@mock.patch('cloudbaseinit.utils.dhcp.get_dhcp_options')
|
|
def _test_get_wire_server_endpoint_address(self, mock_dhcp,
|
|
mock_inet_ntoa,
|
|
mock_time_sleep,
|
|
dhcp_option=None):
|
|
mock_dhcp.return_value = dhcp_option
|
|
if not dhcp_option:
|
|
self.assertRaises(exception.MetadaNotFoundException,
|
|
(self._azureservice.
|
|
_get_wire_server_endpoint_address))
|
|
else:
|
|
mock_inet_ntoa.return_value = mock.sentinel.endpoint
|
|
res = self._azureservice._get_wire_server_endpoint_address()
|
|
self.assertEqual(res, mock.sentinel.endpoint)
|
|
|
|
def test_get_wire_server_endpoint_address_no_endpoint(self):
|
|
self._test_get_wire_server_endpoint_address()
|
|
|
|
def test_get_wire_server_endpoint_address(self):
|
|
dhcp_option = {
|
|
self._azureservice_module.WIRESERVER_DHCP_OPTION:
|
|
'mock.sentinel.endpoint'}
|
|
self._test_get_wire_server_endpoint_address(dhcp_option=dhcp_option)
|
|
|
|
@mock.patch('cloudbaseinit.metadata.services.base.'
|
|
'BaseHTTPMetadataService._http_request')
|
|
def _test_wire_server_request(self,
|
|
mock_http_request, mock_base_url=None,
|
|
path=None, data_xml=None, headers=None,
|
|
parse_xml=True):
|
|
self._azureservice._base_url = mock_base_url
|
|
if not mock_base_url:
|
|
self.assertRaises(exception.CloudbaseInitException,
|
|
self._azureservice._wire_server_request, path)
|
|
return
|
|
if headers and data_xml:
|
|
expected_headers = self._azureservice._headers.copy()
|
|
expected_headers["Content-Type"] = "text/xml; charset=utf-8"
|
|
expected_headers.update(headers)
|
|
self._azureservice._wire_server_request(path, data_xml, headers,
|
|
parse_xml)
|
|
mock_http_request.assert_called_once_with(path, data_xml,
|
|
headers=expected_headers)
|
|
return
|
|
mock_http_request.return_value = str(mock.sentinel.data)
|
|
res = self._azureservice._wire_server_request(path, data_xml,
|
|
headers, parse_xml)
|
|
self.assertEqual(mock_http_request.call_count, 1)
|
|
|
|
if parse_xml:
|
|
self.assertEqual(self._mock_untangle.parse.call_count, 1)
|
|
self.assertEqual(res, self._mock_untangle.parse.return_value)
|
|
else:
|
|
self.assertEqual(res, str(mock.sentinel.data))
|
|
|
|
def test_wire_server_request_url_not_set(self):
|
|
self._test_wire_server_request()
|
|
|
|
def test_wire_server_request_url_set_no_parse(self):
|
|
mock_base_url = "fake-url"
|
|
self._test_wire_server_request(mock_base_url=mock_base_url,
|
|
parse_xml=False)
|
|
|
|
def test_wire_server_request_url_set_with_headers(self):
|
|
mock_base_url = "fake-url"
|
|
self._test_wire_server_request(mock_base_url=mock_base_url,
|
|
parse_xml=False,
|
|
headers={"fake-header": "fake-value"},
|
|
data_xml="fake-data")
|
|
|
|
def test_wire_server_request_parse_xml(self):
|
|
mock_base_url = "fake-url"
|
|
self._test_wire_server_request(mock_base_url=mock_base_url)
|
|
|
|
def test_encode_xml(self):
|
|
fake_root_xml = self._azureservice_module.ElementTree.Element(
|
|
"faketag")
|
|
expected_encoded_xml = ("<?xml version='1.0' encoding='utf-8'?>"
|
|
"\n<faketag />").encode()
|
|
self.assertEqual(self._azureservice._encode_xml(fake_root_xml),
|
|
expected_encoded_xml)
|
|
|
|
@mock.patch(MODPATH + "._get_role_instance_id")
|
|
@mock.patch(MODPATH + "._get_container_id")
|
|
@mock.patch(MODPATH + "._get_incarnation")
|
|
def test__get_health_report_xml(self, mock_get_incarnation,
|
|
mock_get_container_id,
|
|
mock_get_role_instance_id):
|
|
mock_state = 'FakeState'
|
|
mock_substatus = 'FakeStatus'
|
|
mock_description = 'FakeDescription'
|
|
mock_get_incarnation.return_value = "fake"
|
|
mock_get_container_id.return_value = "fakeid"
|
|
mock_get_role_instance_id.return_value = "fakeroleid"
|
|
res = self._azureservice._get_health_report_xml(mock_state,
|
|
mock_substatus,
|
|
mock_description)
|
|
|
|
expected_result = "<?xml version='1.0' encoding='utf-8'?>\n<Health>" \
|
|
"<GoalStateIncarnation>{}</GoalStateIncarnation>" \
|
|
"<Container><ContainerId>{}</ContainerId>" \
|
|
"<RoleInstanceList><Role><InstanceId>{}</InstanceId>" \
|
|
"<Health><State>{}</State><Details><SubStatus>{}</SubStatus>" \
|
|
"<Description>{}</Description></Details></Health>" \
|
|
"</Role></RoleInstanceList></Container></Health>"
|
|
self.assertEqual(encoding.get_as_string(res),
|
|
expected_result.format(
|
|
mock_get_incarnation.return_value,
|
|
mock_get_container_id.return_value,
|
|
mock_get_role_instance_id.return_value,
|
|
mock_state,
|
|
mock_substatus,
|
|
mock_description))
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
def _test_get_goal_state(self, mock_wire_server_request,
|
|
goal_state=True, invalid_state=False):
|
|
mock_goalstate = mock.Mock()
|
|
mock_goalstate.GoalState = mock.Mock()
|
|
mock_goalstate.GoalState.Machine = mock.Mock()
|
|
mock_wire_server_request.return_value = mock_goalstate
|
|
if goal_state:
|
|
self._azureservice._goal_state = mock_goalstate
|
|
else:
|
|
self._azureservice._goal_state = False
|
|
if invalid_state:
|
|
mock_goalstate.GoalState.Machine.ExpectedState = \
|
|
not self._azureservice_module.GOAL_STATE_STARTED
|
|
self.assertRaises(exception.CloudbaseInitException,
|
|
self._azureservice._get_goal_state)
|
|
else:
|
|
if not goal_state:
|
|
mock_goalstate.GoalState.Machine.ExpectedState = \
|
|
self._azureservice_module.GOAL_STATE_STARTED
|
|
else:
|
|
self._azureservice._goal_state.Machine.ExpectedState = \
|
|
self._azureservice_module.GOAL_STATE_STARTED
|
|
res = self._azureservice._get_goal_state()
|
|
self.assertEqual(res, mock_goalstate.GoalState)
|
|
|
|
if not goal_state:
|
|
mock_wire_server_request.assert_called_once_with(
|
|
"machine?comp=goalstate")
|
|
|
|
def test_get_goal_state_exception(self):
|
|
self._test_get_goal_state(invalid_state=True)
|
|
|
|
def test_get_goal_state(self):
|
|
self._test_get_goal_state(goal_state=False)
|
|
|
|
@mock.patch(MODPATH + "._get_goal_state")
|
|
def test__get_incarnation(self, mock_get_goal_state):
|
|
mock_goal_state = mock.Mock()
|
|
mock_get_goal_state.return_value = mock_goal_state
|
|
mock_goal_state.Incarnation.cdata = mock.sentinel.cdata
|
|
|
|
res = self._azureservice._get_incarnation()
|
|
mock_get_goal_state.assert_called_once_with()
|
|
self.assertEqual(res, mock.sentinel.cdata)
|
|
|
|
@mock.patch(MODPATH + "._get_goal_state")
|
|
def test__get_container_id(self, mock_get_goal_state):
|
|
mock_goal_state = mock.Mock()
|
|
mock_get_goal_state.return_value = mock_goal_state
|
|
mock_goal_state.Container.ContainerId.cdata = mock.sentinel.cdata
|
|
|
|
res = self._azureservice._get_container_id()
|
|
mock_get_goal_state.assert_called_once_with()
|
|
self.assertEqual(res, mock.sentinel.cdata)
|
|
|
|
@mock.patch(MODPATH + "._get_goal_state")
|
|
def test__get_role_instance_config(self, mock_get_goal_state):
|
|
mock_goal_state = mock.Mock()
|
|
mock_role = mock.Mock()
|
|
mock_get_goal_state.return_value = mock_goal_state
|
|
mock_goal_state.Container.RoleInstanceList.RoleInstance = mock_role
|
|
mock_role.Configuration = mock.sentinel.config_role
|
|
|
|
res = self._azureservice._get_role_instance_config()
|
|
mock_get_goal_state.assert_called_once_with()
|
|
self.assertEqual(res, mock.sentinel.config_role)
|
|
|
|
@mock.patch(MODPATH + "._get_goal_state")
|
|
def test__get_role_instance_id(self, mock_get_goal_state):
|
|
mock_goal_state = mock.Mock()
|
|
mock_role = mock.Mock()
|
|
mock_get_goal_state.return_value = mock_goal_state
|
|
mock_goal_state.Container.RoleInstanceList.RoleInstance = mock_role
|
|
mock_role.InstanceId.cdata = mock.sentinel.config_role
|
|
|
|
res = self._azureservice._get_role_instance_id()
|
|
mock_get_goal_state.assert_called_once_with()
|
|
self.assertEqual(res, mock.sentinel.config_role)
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
@mock.patch(MODPATH + "._get_health_report_xml")
|
|
def test__post_health_status(self, mock_get_health_report_xml,
|
|
mock_wire_server_request):
|
|
mock_get_health_report_xml.return_value = mock.sentinel.report_xml
|
|
mock_state = mock.sentinel.state
|
|
expected_logging = ["Health data: %s" % mock.sentinel.report_xml]
|
|
with self._logsnatcher:
|
|
self._azureservice._post_health_status(state=mock_state)
|
|
self.assertEqual(self._logsnatcher.output, expected_logging)
|
|
mock_get_health_report_xml.assert_called_once_with(mock_state,
|
|
None, None)
|
|
mock_wire_server_request.assert_called_once_with(
|
|
"machine?comp=health", mock.sentinel.report_xml, parse_xml=False)
|
|
|
|
@mock.patch(MODPATH + "._post_health_status")
|
|
def test_provisioning_started(self, mock_post_health_status):
|
|
self._azureservice.provisioning_started()
|
|
mock_post_health_status.assert_called_once_with(
|
|
self._azureservice_module.HEALTH_STATE_NOT_READY,
|
|
self._azureservice_module.HEALTH_SUBSTATE_PROVISIONING,
|
|
"Cloudbase-Init is preparing your computer for first use...")
|
|
|
|
@mock.patch(MODPATH + "._post_health_status")
|
|
def test_provisioning_completed(self, mock_post_health_status):
|
|
self._azureservice.provisioning_completed()
|
|
mock_post_health_status.assert_called_once_with(
|
|
self._azureservice_module.HEALTH_STATE_READY)
|
|
|
|
@mock.patch(MODPATH + "._post_health_status")
|
|
def test_provisioning_failed(self, mock_post_health_status):
|
|
self._azureservice.provisioning_failed()
|
|
mock_post_health_status.assert_called_once_with(
|
|
self._azureservice_module.HEALTH_STATE_NOT_READY,
|
|
self._azureservice_module.HEALTH_SUBSTATE_PROVISIONING_FAILED,
|
|
"Provisioning failed")
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
@mock.patch(MODPATH + "._get_role_properties_xml")
|
|
def test__post_role_properties(self, mock_get_role_properties_xml,
|
|
mock_wire_server_request):
|
|
mock_properties = mock.sentinel.properties
|
|
mock_get_role_properties_xml.return_value = mock_properties
|
|
expected_logging = ["Role properties data: %s" % mock_properties]
|
|
with self._logsnatcher:
|
|
self._azureservice._post_role_properties(mock_properties)
|
|
self.assertEqual(self._logsnatcher.output, expected_logging)
|
|
mock_get_role_properties_xml.assert_called_once_with(mock_properties)
|
|
mock_wire_server_request.assert_called_once_with(
|
|
"machine?comp=roleProperties", mock_properties, parse_xml=False)
|
|
|
|
def test_can_post_rdp_cert_thumbprint(self):
|
|
self.assertTrue(self._azureservice.can_post_rdp_cert_thumbprint)
|
|
|
|
@mock.patch(MODPATH + "._post_role_properties")
|
|
def test_post_rdp_cert_thumbprint(self, mock_post_role_properties):
|
|
mock_thumbprint = mock.sentinel.thumbprint
|
|
self._azureservice.post_rdp_cert_thumbprint(mock_thumbprint)
|
|
expected_props = {
|
|
self._azureservice_module.ROLE_PROPERTY_CERT_THUMB:
|
|
mock_thumbprint}
|
|
mock_post_role_properties.assert_called_once_with(expected_props)
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
@mock.patch(MODPATH + "._get_role_instance_config")
|
|
def test__get_hosting_environment(self, mock_get_role_instance_config,
|
|
mock_wire_server_request):
|
|
mock_config = mock.Mock()
|
|
mock_get_role_instance_config.return_value = mock_config
|
|
mock_config.HostingEnvironmentConfig.cdata = mock.sentinel.data
|
|
|
|
self._azureservice._get_hosting_environment()
|
|
mock_get_role_instance_config.assert_called_once_with()
|
|
mock_wire_server_request.assert_called_once_with(mock.sentinel.data)
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
@mock.patch(MODPATH + "._get_role_instance_config")
|
|
def test__get_shared_config(self, mock_get_role_instance_config,
|
|
mock_wire_server_request):
|
|
mock_config = mock.Mock()
|
|
mock_get_role_instance_config.return_value = mock_config
|
|
mock_config.SharedConfig.cdata = mock.sentinel.data
|
|
|
|
self._azureservice._get_shared_config()
|
|
mock_get_role_instance_config.assert_called_once_with()
|
|
mock_wire_server_request.assert_called_once_with(mock.sentinel.data)
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
@mock.patch(MODPATH + "._get_role_instance_config")
|
|
def test__get_extensions_config(self, mock_get_role_instance_config,
|
|
mock_wire_server_request):
|
|
mock_config = mock.Mock()
|
|
mock_get_role_instance_config.return_value = mock_config
|
|
mock_config.ExtensionsConfig.cdata = mock.sentinel.data
|
|
|
|
self._azureservice._get_extensions_config()
|
|
mock_get_role_instance_config.assert_called_once_with()
|
|
mock_wire_server_request.assert_called_once_with(mock.sentinel.data)
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
@mock.patch(MODPATH + "._get_role_instance_config")
|
|
def test__get_full_config(self, mock_get_role_instance_config,
|
|
mock_wire_server_request):
|
|
mock_config = mock.Mock()
|
|
mock_get_role_instance_config.return_value = mock_config
|
|
mock_config.FullConfig.cdata = mock.sentinel.data
|
|
|
|
self._azureservice._get_full_config()
|
|
mock_get_role_instance_config.assert_called_once_with()
|
|
mock_wire_server_request.assert_called_once_with(mock.sentinel.data)
|
|
|
|
def test__create_transport_cert(self):
|
|
mock_cert_mgr = mock.Mock()
|
|
expected_certs = (mock.sentinel.thumbprint, mock.sentinel.cert)
|
|
|
|
mock_cert_mgr.create_self_signed_cert.return_value = (
|
|
mock.sentinel.thumbprint, mock.sentinel.cert)
|
|
with testutils.ConfPatcher(key='transport_cert_store_name',
|
|
value='fake_name', group="azure"):
|
|
with self._azureservice._create_transport_cert(mock_cert_mgr) as r:
|
|
self.assertEqual(r, expected_certs)
|
|
(mock_cert_mgr.create_self_signed_cert.
|
|
assert_called_once_with("CN=Cloudbase-Init AzureService Transport",
|
|
machine_keyset=True,
|
|
store_name="fake_name"))
|
|
(mock_cert_mgr.delete_certificate_from_store.
|
|
assert_called_once_with(mock.sentinel.thumbprint,
|
|
machine_keyset=True,
|
|
store_name="fake_name"))
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
def test__get_encoded_cert(self, mock_wire_server_request):
|
|
mock_cert_config = mock.Mock()
|
|
mock_transport_cert = mock.Mock()
|
|
mock_cert_url = mock.sentinel.cert_url
|
|
|
|
mock_transport_cert.replace.return_value = mock.sentinel.transport_cert
|
|
mock_wire_server_request.return_value = mock_cert_config
|
|
mock_cert_config.CertificateFile.Data.cdata = mock.sentinel.cert_data
|
|
mock_cert_config.CertificateFile.Format.cdata = mock.sentinel.cert_fmt
|
|
|
|
expected_headers = {
|
|
"x-ms-guest-agent-public-x509-cert": mock.sentinel.transport_cert}
|
|
expected_result = (mock.sentinel.cert_data, mock.sentinel.cert_fmt)
|
|
res = self._azureservice._get_encoded_cert(mock_cert_url,
|
|
mock_transport_cert)
|
|
(mock_wire_server_request.
|
|
assert_called_once_with(mock_cert_url, headers=expected_headers))
|
|
self.assertEqual(res, expected_result)
|
|
|
|
@mock.patch(MODPATH + "._get_versions")
|
|
def _test__check_version_header(self, mock_get_versions, version):
|
|
mock_version = mock.Mock()
|
|
mock_get_versions.return_value = mock_version
|
|
mock_version.Versions.Supported.Version = [version]
|
|
if self._azureservice_module.WIRE_SERVER_VERSION is not version:
|
|
self.assertRaises(exception.MetadaNotFoundException,
|
|
self._azureservice._check_version_header)
|
|
else:
|
|
self._azureservice._check_version_header()
|
|
self.assertEqual(self._azureservice._headers["x-ms-version"],
|
|
version)
|
|
|
|
def test_check_version_header_unsupported_version(self):
|
|
version = "fake-version"
|
|
self._test__check_version_header(version=version)
|
|
|
|
def test_check_version_header_supported(self):
|
|
version = self._azureservice_module.WIRE_SERVER_VERSION
|
|
self._test__check_version_header(version=version)
|
|
|
|
@mock.patch(MODPATH + "._wire_server_request")
|
|
def test__get_versions(self, mock_server_request):
|
|
mock_server_request.return_value = mock.sentinel.version
|
|
res = self._azureservice._get_versions()
|
|
mock_server_request.assert_called_once_with("?comp=Versions")
|
|
self.assertEqual(res, mock.sentinel.version)
|
|
|
|
@mock.patch(MODPATH + "._get_role_instance_id")
|
|
def test_get_instance_id(self, mock_get_role_instance_id):
|
|
mock_get_role_instance_id.return_value = mock.sentinel.id
|
|
self.assertEqual(self._azureservice.get_instance_id(),
|
|
mock.sentinel.id)
|
|
|
|
@mock.patch("os.path.exists")
|
|
@mock.patch(MODPATH + "._get_config_set_drive_path")
|
|
def _test__get_ovf_env_path(self, mock_get_drives, mock_path_exists,
|
|
path_exists=True):
|
|
mock_get_drives.return_value = 'fake path'
|
|
mock_path_exists.return_value = path_exists
|
|
self._azureservice._osutils.get_logical_drives = mock_get_drives
|
|
if not path_exists:
|
|
self.assertRaises(exception.ItemNotFoundException,
|
|
self._azureservice._get_ovf_env_path)
|
|
else:
|
|
res = self._azureservice._get_ovf_env_path()
|
|
ovf_env_path = os.path.join(
|
|
"fake path", self._azureservice_module.OVF_ENV_FILENAME)
|
|
self.assertEqual(res, ovf_env_path)
|
|
mock_path_exists.assert_called_once_with(ovf_env_path)
|
|
mock_get_drives.assert_called_once_with()
|
|
|
|
def test_get_ovf_env_path_exists(self):
|
|
self._test__get_ovf_env_path()
|
|
|
|
def test_get_ovf_env_path_not_exists(self):
|
|
self._test__get_ovf_env_path(path_exists=False)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env_path")
|
|
def test_get_ovf_env(self, mock_get_ovf_env_path):
|
|
fake_xml = '<?xml version="1.0"?><root><child name="fake"/></root>'
|
|
mock_get_ovf_env_path.return_value = fake_xml
|
|
res = self._azureservice._get_ovf_env()
|
|
self.assertIsNotNone(res)
|
|
mock_get_ovf_env_path.assert_called_once_with()
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def test_get_admin_username(self, mock_get_ovf_env):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_prov_section = mock.Mock()
|
|
mock_win_prov = mock.Mock()
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
mock_ovf_env.Environment.wa_ProvisioningSection = mock_prov_section
|
|
mock_prov_section.WindowsProvisioningConfigurationSet = mock_win_prov
|
|
mock_win_prov.AdminUsername.cdata = mock.sentinel.cdata
|
|
res = self._azureservice.get_admin_username()
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
self.assertEqual(res, mock.sentinel.cdata)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def test_get_admin_password(self, mock_get_ovf_env):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_prov_section = mock.Mock()
|
|
mock_win_prov = mock.Mock()
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
mock_ovf_env.Environment.wa_ProvisioningSection = mock_prov_section
|
|
mock_prov_section.WindowsProvisioningConfigurationSet = mock_win_prov
|
|
mock_win_prov.AdminPassword.cdata = mock.sentinel.cdata
|
|
res = self._azureservice.get_admin_password()
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
self.assertEqual(res, mock.sentinel.cdata)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def test_get_host_name(self, mock_get_ovf_env):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_prov_section = mock.Mock()
|
|
mock_win_prov = mock.Mock()
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
mock_ovf_env.Environment.wa_ProvisioningSection = mock_prov_section
|
|
mock_prov_section.WindowsProvisioningConfigurationSet = mock_win_prov
|
|
mock_win_prov.ComputerName.cdata = mock.sentinel.cdata
|
|
res = self._azureservice.get_host_name()
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
self.assertEqual(res, mock.sentinel.cdata)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def test_get_enable_automatic_updates(self, mock_get_ovf_env,
|
|
enable_updates=True):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_prov_section = mock.Mock()
|
|
mock_win_prov = mock.Mock()
|
|
if not enable_updates:
|
|
mock_win_prov = mock.MagicMock(spec="")
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
mock_ovf_env.Environment.wa_ProvisioningSection = mock_prov_section
|
|
mock_prov_section.WindowsProvisioningConfigurationSet = mock_win_prov
|
|
res = self._azureservice.get_enable_automatic_updates()
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
self.assertFalse(res)
|
|
|
|
def test_get_enable_automatic_updates_no_updates(self):
|
|
self.test_get_enable_automatic_updates(enable_updates=False)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def test_get_winrm_listeners_configuration(self, mock_get_ovf_env):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_prov_section = mock.Mock()
|
|
mock_win_prov = mock.Mock()
|
|
mock_listener = mock.Mock()
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
mock_ovf_env.Environment.wa_ProvisioningSection = mock_prov_section
|
|
mock_prov_section.WindowsProvisioningConfigurationSet = mock_win_prov
|
|
mock_win_prov.WinRM.Listeners.Listener = [mock_listener]
|
|
mock_listener.Protocol.cdata = mock.sentinel.fake_protocol
|
|
(mock_listener.CertificateThumbprint.
|
|
cdata) = mock.sentinel.fake_thumbprint
|
|
|
|
expected_result = [
|
|
{
|
|
'certificate_thumbprint': mock.sentinel.fake_thumbprint,
|
|
'protocol': mock.sentinel.fake_protocol,
|
|
}]
|
|
res = self._azureservice.get_winrm_listeners_configuration()
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
self.assertEqual(res, expected_result)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def test_get_vm_agent_package_provisioning_data(self, mock_get_ovf_env):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_package_name = mock.sentinel.package_name
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
(mock_ovf_env.Environment.wa_PlatformSettingsSection.
|
|
PlatformSettings.GuestAgentPackageName.cdata) = mock_package_name
|
|
res = self._azureservice.get_vm_agent_package_provisioning_data()
|
|
expected_provisioning_data = {
|
|
'provision': False, 'package_name': mock_package_name}
|
|
self.assertEqual(res, expected_provisioning_data)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def test_get_kms_host(self, mock_get_ovf_env):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
self.assertTrue(self._azureservice.get_kms_host())
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
def _test_get_use_avma_licensing(self, mock_get_ovf_env, use_avma):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_ovf_env.Environment = mock.Mock()
|
|
plat_sett_section = mock.Mock()
|
|
if not use_avma:
|
|
plat_sett_section.PlatformSettings = mock.MagicMock(spec="")
|
|
mock_ovf_env.Environment.wa_PlatformSettingsSection = plat_sett_section
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
|
|
self.assertFalse(self._azureservice.get_use_avma_licensing())
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
|
|
def test_get_use_avma_licensing(self):
|
|
self._test_get_use_avma_licensing(use_avma=True)
|
|
|
|
def test_get_use_avma_licensing_no_use_avma(self):
|
|
self._test_get_use_avma_licensing(use_avma=False)
|
|
|
|
@mock.patch(MODPATH + "._get_ovf_env")
|
|
@mock.patch(MODPATH + "._check_version_header")
|
|
@mock.patch(MODPATH + "._get_wire_server_endpoint_address")
|
|
def _test_load(self, mock_get_endpoint_address,
|
|
mock_check_version_header, mock_get_ovf_env,
|
|
endpoint_side_effect=None, load_side_effect=None):
|
|
if endpoint_side_effect:
|
|
mock_get_endpoint_address.side_effect = endpoint_side_effect
|
|
expected_logging = ["Azure WireServer endpoint not found"]
|
|
with self._logsnatcher:
|
|
res = self._azureservice.load()
|
|
self.assertFalse(res)
|
|
self.assertEqual(self._logsnatcher.output, expected_logging)
|
|
mock_get_endpoint_address.assert_called_once_with()
|
|
return
|
|
|
|
mock_endpoint = mock.sentinel.endpoint
|
|
mock_get_endpoint_address.return_value = mock_endpoint
|
|
if load_side_effect:
|
|
mock_check_version_header.side_effect = load_side_effect
|
|
res = self._azureservice.load()
|
|
self.assertFalse(res)
|
|
return
|
|
else:
|
|
res = self._azureservice.load()
|
|
self.assertTrue(res)
|
|
self.assertIn(str(mock_endpoint), self._azureservice._base_url)
|
|
mock_check_version_header.assert_called_once_with()
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
return
|
|
|
|
def test_load_no_endpoint(self):
|
|
self._test_load(endpoint_side_effect=Exception)
|
|
|
|
def test_load_exception(self):
|
|
exc = Exception("Fake exception")
|
|
self._test_load(load_side_effect=exc)
|
|
|
|
def test_load(self):
|
|
self._test_load()
|
|
|
|
@mock.patch('os.path.exists')
|
|
def _test_get_config_set_drive_path(self, mock_path_exists,
|
|
path_exists=True):
|
|
self._azureservice._set_config_drive_path = None
|
|
mock_osutils = mock.Mock()
|
|
mock_osutils.get_logical_drives.return_value = ['fake path'] * 3
|
|
self._azureservice._osutils = mock_osutils
|
|
mock_path_exists.side_effect = [False] * 2 + [path_exists]
|
|
if path_exists:
|
|
result = self._azureservice._get_config_set_drive_path()
|
|
self.assertEqual(result, 'fake path')
|
|
else:
|
|
self.assertRaises(exception.ItemNotFoundException,
|
|
self._azureservice._get_config_set_drive_path)
|
|
self.assertEqual(mock_path_exists.call_count, 3)
|
|
|
|
def test_get_config_set_drive_path(self):
|
|
self._test_get_config_set_drive_path()
|
|
|
|
def test_get_config_set_drive_path_not_exists(self):
|
|
self._test_get_config_set_drive_path(path_exists=False)
|
|
|
|
@mock.patch(MODPATH + '._get_ovf_env')
|
|
def test_check_ovf_env_custom_data(self, mock_get_ovf_env,
|
|
custom_data=True):
|
|
mock_ovf_env = mock.Mock()
|
|
mock_prov_section = mock.Mock()
|
|
mock_win_prov = mock.Mock()
|
|
mock_win_prov.PlatformSettings = mock.Mock()
|
|
if not custom_data:
|
|
mock_win_prov.PlatformSettings = mock.MagicMock(spec="")
|
|
mock_get_ovf_env.return_value = mock_ovf_env
|
|
mock_ovf_env.Environment.wa_ProvisioningSection = mock_prov_section
|
|
mock_prov_section.WindowsProvisioningConfigurationSet = mock_win_prov
|
|
res = self._azureservice._check_ovf_env_custom_data()
|
|
mock_get_ovf_env.assert_called_once_with()
|
|
self.assertTrue(res)
|
|
|
|
@mock.patch(MODPATH + '._check_ovf_env_custom_data')
|
|
def test_get_user_data_ItemNotFound(self, mock_check_custom_data):
|
|
mock_check_custom_data.return_value = True
|
|
self._azureservice._config_set_drive_path = "fake path"
|
|
self.assertRaises(exception.ItemNotFoundException,
|
|
self._azureservice.get_user_data)
|
|
|
|
@mock.patch(MODPATH + '._check_ovf_env_custom_data')
|
|
def test_get_user_data_NoMetadataException(self, mock_check_custom_data):
|
|
mock_check_custom_data.return_value = False
|
|
self._azureservice._config_set_drive_path = "fake path"
|
|
self.assertRaises(
|
|
self._azureservice_module.base.NotExistingMetadataException,
|
|
self._azureservice.get_user_data)
|
|
|
|
@mock.patch(MODPATH + '._get_role_instance_config')
|
|
def test_get_server_certs_no_certs(self, mock_get_instance_config):
|
|
mock_get_instance_config.return_value = mock.MagicMock(spec="")
|
|
res = self._azureservice.get_server_certs()
|
|
self.assertEqual(res, [])
|
|
|
|
@mock.patch(MODPATH + '._get_hosting_environment')
|
|
@mock.patch(MODPATH + '._get_encoded_cert')
|
|
@mock.patch(MODPATH + '._get_role_instance_config')
|
|
@mock.patch(MODPATH + '._create_transport_cert')
|
|
@mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager')
|
|
def test_get_server_certs(self, mock_cert_manager, mock_create_cert,
|
|
mock_get_config, mock_get_encoded_cert,
|
|
mock_get_hosting_env):
|
|
cert_model = {
|
|
"storeName": mock.sentinel.storeName,
|
|
"configurationLevel": mock.sentinel.configurationLevel,
|
|
"certificateId": mock.sentinel.certificateId,
|
|
"name": mock.sentinel.name
|
|
}
|
|
mock_cert_mgr = mock.Mock()
|
|
mock_cert_mgr.decode_pkcs7_base64_blob.return_value = \
|
|
mock.sentinel.pfx_data
|
|
mock_cert_manager.return_value = mock_cert_mgr
|
|
mock_create_cert.return_value.__enter__.return_value = \
|
|
(mock.sentinel.thumbprint, mock.sentinel.cert)
|
|
mock_get_encoded_cert.return_value = \
|
|
(mock.sentinel.cert_data, mock.sentinel.cert_format)
|
|
mock_host_env = mock.Mock()
|
|
mock_host_env_config = mock_host_env.HostingEnvironmentConfig
|
|
mock_host_env_config.StoredCertificates.StoredCertificate = \
|
|
[cert_model]
|
|
mock_get_hosting_env.return_value = mock_host_env
|
|
|
|
res = self._azureservice.get_server_certs()
|
|
expected_result = [{
|
|
"store_name": mock.sentinel.storeName,
|
|
"store_location": mock.sentinel.configurationLevel,
|
|
"certificate_id": mock.sentinel.certificateId,
|
|
"name": mock.sentinel.name,
|
|
"pfx_data": mock.sentinel.pfx_data,
|
|
}]
|
|
self.assertEqual(res, expected_result)
|
|
self.assertEqual(mock_create_cert.call_count, 1)
|
|
self.assertEqual(mock_get_encoded_cert.call_count, 1)
|
|
self.assertEqual(mock_cert_mgr.decode_pkcs7_base64_blob.call_count, 1)
|
|
mock_cert_manager.assert_called_once_with()
|
|
mock_get_config.assert_called_once_with()
|
|
mock_get_hosting_env.assert_called_once_with()
|
|
|
|
@mock.patch(MODPATH + '.get_user_data')
|
|
def test_get_decoded_user_data(self, mock_get_user_data):
|
|
mock_get_user_data.return_value = mock.sentinel.user_data
|
|
res = self._azureservice.get_decoded_user_data()
|
|
self.assertEqual(res, mock.sentinel.user_data)
|
|
|
|
@mock.patch(MODPATH + '.get_content')
|
|
def test_get_ephemeral_disk_data_loss_warning(self, mock_get_content):
|
|
mock_get_content.return_value = mock.sentinel.content
|
|
res = self._azureservice.get_ephemeral_disk_data_loss_warning()
|
|
self.assertEqual(res, mock.sentinel.content)
|
|
mock_get_content.assert_called_once_with(
|
|
self._azureservice_module.DATALOSS_WARNING_PATH)
|
|
|
|
@mock.patch(MODPATH + "._get_role_instance_id")
|
|
@mock.patch(MODPATH + "._get_container_id")
|
|
def _test_get_role_properties_xml(self, mock_get_container_id,
|
|
mock_get_role_instance_id,
|
|
properties):
|
|
mock_get_container_id.return_value = "fake container id"
|
|
mock_get_role_instance_id.return_value = "fake instance id"
|
|
|
|
res = self._azureservice._get_role_properties_xml(properties)
|
|
expected_properties = ""
|
|
property_template = '<Property name=\"{property_name}"'\
|
|
' value="{value}" />'
|
|
result_template = ("<?xml version=\'1.0\' encoding=\'utf-8\'?>\n"
|
|
"<RoleProperties><Container><ContainerId>"
|
|
"{container_id}</ContainerId><RoleInstances>"
|
|
"<RoleInstance><Id>{instance_id}</Id>"
|
|
"{properties}</RoleInstance>"
|
|
"</RoleInstances></Container></RoleProperties>")
|
|
if properties:
|
|
expected_properties = "<Properties>"
|
|
for name, value in properties.items():
|
|
expected_properties += property_template.format(
|
|
property_name=name,
|
|
value=value)
|
|
expected_properties += "</Properties>"
|
|
else:
|
|
expected_properties = "<Properties />"
|
|
expected_result = result_template.format(
|
|
container_id=mock_get_container_id.return_value,
|
|
instance_id=mock_get_role_instance_id.return_value,
|
|
properties=expected_properties)
|
|
self.assertEqual(encoding.get_as_string(res), expected_result)
|
|
|
|
def test_get_role_properties_xml_no_properties(self):
|
|
self._test_get_role_properties_xml(properties={})
|
|
|
|
def test_get_role_properties_xml(self):
|
|
properties = {
|
|
"fake property 1": "fake value 1",
|
|
"fake property 2": "fake value 2"
|
|
}
|
|
self._test_get_role_properties_xml(properties=properties)
|