Coverage for manila/tests/share/drivers/windows/test_service_instance.py: 100%
177 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright (c) 2015 Cloudbase Solutions SRL
2# All Rights Reserved.
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
16import os
17from unittest import mock
19import ddt
20from oslo_concurrency import processutils
21from oslo_config import cfg
23from manila import exception
24from manila.share import configuration
25from manila.share.drivers import service_instance as generic_service_instance
26from manila.share.drivers.windows import service_instance
27from manila.share.drivers.windows import windows_utils
28from manila import test
30CONF = cfg.CONF
31CONF.import_opt('driver_handles_share_servers',
32 'manila.share.driver')
33CONF.register_opts(generic_service_instance.common_opts)
35serv_mgr_cls = service_instance.WindowsServiceInstanceManager
36generic_serv_mgr_cls = generic_service_instance.ServiceInstanceManager
39@ddt.ddt
40class WindowsServiceInstanceManagerTestCase(test.TestCase):
41 _FAKE_SERVER = {'ip': mock.sentinel.ip,
42 'instance_id': mock.sentinel.instance_id}
44 @mock.patch.object(windows_utils, 'WindowsUtils')
45 @mock.patch.object(serv_mgr_cls, '_check_auth_mode')
46 def setUp(self, mock_check_auth, mock_utils_cls):
47 self.flags(service_instance_user=mock.sentinel.username)
48 self._remote_execute = mock.Mock()
50 fake_conf = configuration.Configuration(None)
51 self._mgr = serv_mgr_cls(remote_execute=self._remote_execute,
52 driver_config=fake_conf)
53 self._windows_utils = mock_utils_cls.return_value
54 super(WindowsServiceInstanceManagerTestCase, self).setUp()
56 @ddt.data({},
57 {'use_cert_auth': False},
58 {'use_cert_auth': False, 'valid_pass_complexity': False},
59 {'certs_exist': False})
60 @mock.patch('os.path.exists')
61 @mock.patch.object(serv_mgr_cls, '_check_password_complexity')
62 @ddt.unpack
63 def test_check_auth_mode(self, mock_check_complexity, mock_path_exists,
64 use_cert_auth=True, certs_exist=True,
65 valid_pass_complexity=True):
66 self.flags(service_instance_password=mock.sentinel.password)
67 self._mgr._cert_pem_path = mock.sentinel.cert_path
68 self._mgr._cert_key_pem_path = mock.sentinel.key_path
69 mock_path_exists.return_value = certs_exist
70 mock_check_complexity.return_value = valid_pass_complexity
72 self._mgr._use_cert_auth = use_cert_auth
74 invalid_auth = ((use_cert_auth and not certs_exist)
75 or not valid_pass_complexity)
77 if invalid_auth:
78 self.assertRaises(exception.ServiceInstanceException,
79 self._mgr._check_auth_mode)
80 else:
81 self._mgr._check_auth_mode()
83 if not use_cert_auth:
84 mock_check_complexity.assert_called_once_with(
85 str(mock.sentinel.password))
87 @ddt.data(False, True)
88 def test_get_auth_info(self, use_cert_auth):
89 self._mgr._use_cert_auth = use_cert_auth
90 self._mgr._cert_pem_path = mock.sentinel.cert_path
91 self._mgr._cert_key_pem_path = mock.sentinel.key_path
93 auth_info = self._mgr._get_auth_info()
95 expected_auth_info = {'use_cert_auth': use_cert_auth}
96 if use_cert_auth:
97 expected_auth_info.update(cert_pem_path=mock.sentinel.cert_path,
98 cert_key_pem_path=mock.sentinel.key_path)
100 self.assertEqual(expected_auth_info, auth_info)
102 @mock.patch.object(serv_mgr_cls, '_get_auth_info')
103 @mock.patch.object(generic_serv_mgr_cls, 'get_common_server')
104 def test_common_server(self, mock_generic_get_server, mock_get_auth):
105 mock_server_details = {'backend_details': {}}
106 mock_auth_info = {'fake_auth_info': mock.sentinel.auth_info}
108 mock_generic_get_server.return_value = mock_server_details
109 mock_get_auth.return_value = mock_auth_info
110 expected_server_details = dict(backend_details=mock_auth_info)
112 server_details = self._mgr.get_common_server()
114 mock_generic_get_server.assert_called_once_with()
115 self.assertEqual(expected_server_details, server_details)
117 @mock.patch.object(serv_mgr_cls, '_get_auth_info')
118 @mock.patch.object(generic_serv_mgr_cls, '_get_new_instance_details')
119 def test_get_new_instance_details(self, mock_generic_get_details,
120 mock_get_auth):
121 mock_server_details = {'fake_server_details':
122 mock.sentinel.server_details}
123 mock_generic_get_details.return_value = mock_server_details
124 mock_auth_info = {'fake_auth_info': mock.sentinel.auth_info}
125 mock_get_auth.return_value = mock_auth_info
127 expected_server_details = dict(mock_server_details, **mock_auth_info)
128 instance_details = self._mgr._get_new_instance_details(
129 server=mock.sentinel.server)
131 mock_generic_get_details.assert_called_once_with(mock.sentinel.server)
132 self.assertEqual(expected_server_details, instance_details)
134 @ddt.data(('abAB01', True),
135 ('abcdef', False),
136 ('aA0', False))
137 @ddt.unpack
138 def test_check_password_complexity(self, password, expected_result):
139 valid_complexity = self._mgr._check_password_complexity(
140 password)
141 self.assertEqual(expected_result, valid_complexity)
143 @ddt.data(None, Exception)
144 def test_server_connection(self, side_effect):
145 self._remote_execute.side_effect = side_effect
147 expected_result = side_effect is None
148 is_available = self._mgr._test_server_connection(self._FAKE_SERVER)
150 self.assertEqual(expected_result, is_available)
151 self._remote_execute.assert_called_once_with(self._FAKE_SERVER,
152 "whoami",
153 retry=False)
155 @ddt.data(False, True)
156 def test_get_service_instance_create_kwargs(self, use_cert_auth):
157 self._mgr._use_cert_auth = use_cert_auth
158 self.flags(service_instance_password=mock.sentinel.admin_pass)
160 if use_cert_auth:
161 mock_cert_data = 'mock_cert_data'
162 self.mock_object(service_instance, 'open',
163 mock.mock_open(
164 read_data=mock_cert_data))
165 expected_kwargs = dict(user_data=mock_cert_data)
166 else:
167 expected_kwargs = dict(
168 meta=dict(admin_pass=str(mock.sentinel.admin_pass)))
170 create_kwargs = self._mgr._get_service_instance_create_kwargs()
172 self.assertEqual(expected_kwargs, create_kwargs)
174 @mock.patch.object(generic_serv_mgr_cls, 'set_up_service_instance')
175 @mock.patch.object(serv_mgr_cls, 'get_valid_security_service')
176 @mock.patch.object(serv_mgr_cls, '_setup_security_service')
177 def test_set_up_service_instance(self, mock_setup_security_service,
178 mock_get_valid_security_service,
179 mock_generic_setup_serv_inst):
180 mock_service_instance = {'instance_details': None}
181 mock_network_info = {'security_services':
182 mock.sentinel.security_services}
184 mock_generic_setup_serv_inst.return_value = mock_service_instance
185 mock_get_valid_security_service.return_value = (
186 mock.sentinel.security_service)
188 instance_details = self._mgr.set_up_service_instance(
189 mock.sentinel.context, mock_network_info)
191 mock_generic_setup_serv_inst.assert_called_once_with(
192 mock.sentinel.context, mock_network_info)
193 mock_get_valid_security_service.assert_called_once_with(
194 mock.sentinel.security_services)
196 mock_setup_security_service.assert_called_once_with(
197 mock_service_instance, mock.sentinel.security_service)
199 expected_instance_details = dict(mock_service_instance,
200 joined_domain=True)
201 self.assertEqual(expected_instance_details,
202 instance_details)
204 @mock.patch.object(serv_mgr_cls, '_run_cloudbase_init_plugin_after_reboot')
205 @mock.patch.object(serv_mgr_cls, '_join_domain')
206 def test_setup_security_service(self, mock_join_domain,
207 mock_run_cbsinit_plugin):
208 utils = self._windows_utils
209 mock_security_service = {'domain': mock.sentinel.domain,
210 'user': mock.sentinel.admin_username,
211 'password': mock.sentinel.admin_password,
212 'dns_ip': mock.sentinel.dns_ip}
213 utils.get_interface_index_by_ip.return_value = (
214 mock.sentinel.interface_index)
216 self._mgr._setup_security_service(self._FAKE_SERVER,
217 mock_security_service)
219 utils.set_dns_client_search_list.assert_called_once_with(
220 self._FAKE_SERVER,
221 [mock_security_service['domain']])
222 utils.get_interface_index_by_ip.assert_called_once_with(
223 self._FAKE_SERVER,
224 self._FAKE_SERVER['ip'])
225 utils.set_dns_client_server_addresses.assert_called_once_with(
226 self._FAKE_SERVER,
227 mock.sentinel.interface_index,
228 [mock_security_service['dns_ip']])
229 mock_run_cbsinit_plugin.assert_called_once_with(
230 self._FAKE_SERVER,
231 plugin_name=self._mgr._CBS_INIT_WINRM_PLUGIN)
232 mock_join_domain.assert_called_once_with(
233 self._FAKE_SERVER,
234 mock.sentinel.domain,
235 mock.sentinel.admin_username,
236 mock.sentinel.admin_password)
238 @ddt.data({'join_domain_side_eff': Exception},
239 {'server_available': False,
240 'expected_exception': exception.ServiceInstanceException},
241 {'join_domain_side_eff': processutils.ProcessExecutionError,
242 'expected_exception': processutils.ProcessExecutionError},
243 {'domain_mismatch': True,
244 'expected_exception': exception.ServiceInstanceException})
245 @mock.patch.object(generic_serv_mgr_cls, 'reboot_server')
246 @mock.patch.object(generic_serv_mgr_cls, 'wait_for_instance_to_be_active')
247 @mock.patch.object(generic_serv_mgr_cls, '_check_server_availability')
248 @ddt.unpack
249 def test_join_domain(self, mock_check_avail,
250 mock_wait_instance_active,
251 mock_reboot_server,
252 expected_exception=None,
253 server_available=True,
254 domain_mismatch=False,
255 join_domain_side_eff=None):
256 self._windows_utils.join_domain.side_effect = join_domain_side_eff
257 mock_check_avail.return_value = server_available
258 self._windows_utils.get_current_domain.return_value = (
259 None if domain_mismatch else mock.sentinel.domain)
260 domain_params = (mock.sentinel.domain,
261 mock.sentinel.admin_username,
262 mock.sentinel.admin_password)
264 if expected_exception:
265 self.assertRaises(expected_exception,
266 self._mgr._join_domain,
267 self._FAKE_SERVER,
268 *domain_params)
269 else:
270 self._mgr._join_domain(self._FAKE_SERVER,
271 *domain_params)
273 if join_domain_side_eff != processutils.ProcessExecutionError:
274 mock_reboot_server.assert_called_once_with(
275 self._FAKE_SERVER, soft_reboot=True)
276 mock_wait_instance_active.assert_called_once_with(
277 self._FAKE_SERVER['instance_id'],
278 timeout=self._mgr.max_time_to_build_instance)
279 mock_check_avail.assert_called_once_with(self._FAKE_SERVER)
280 if server_available:
281 self._windows_utils.get_current_domain.assert_called_once_with(
282 self._FAKE_SERVER)
284 self._windows_utils.join_domain.assert_called_once_with(
285 self._FAKE_SERVER,
286 *domain_params)
288 @ddt.data([],
289 [{'type': 'active_directory'}],
290 [{'type': 'active_directory'}] * 2,
291 [{'type': mock.sentinel.invalid_type}])
292 def test_get_valid_security_service(self, security_services):
293 valid_security_service = self._mgr.get_valid_security_service(
294 security_services)
296 if (security_services and len(security_services) == 1 and
297 security_services[0]['type'] == 'active_directory'):
298 expected_valid_sec_service = security_services[0]
299 else:
300 expected_valid_sec_service = None
302 self.assertEqual(expected_valid_sec_service,
303 valid_security_service)
305 @mock.patch.object(serv_mgr_cls, '_get_cbs_init_reg_section')
306 def test_run_cloudbase_init_plugin_after_reboot(self,
307 mock_get_cbs_init_reg):
308 self._FAKE_SERVER = {'instance_id': mock.sentinel.instance_id}
309 mock_get_cbs_init_reg.return_value = mock.sentinel.cbs_init_reg_sect
310 expected_plugin_key_path = "%(cbs_init)s\\%(instance_id)s\\Plugins" % {
311 'cbs_init': mock.sentinel.cbs_init_reg_sect,
312 'instance_id': self._FAKE_SERVER['instance_id']}
314 self._mgr._run_cloudbase_init_plugin_after_reboot(
315 server=self._FAKE_SERVER,
316 plugin_name=mock.sentinel.plugin_name)
318 mock_get_cbs_init_reg.assert_called_once_with(self._FAKE_SERVER)
319 self._windows_utils.set_win_reg_value.assert_called_once_with(
320 self._FAKE_SERVER,
321 path=expected_plugin_key_path,
322 key=mock.sentinel.plugin_name,
323 value=self._mgr._CBS_INIT_RUN_PLUGIN_AFTER_REBOOT)
325 @ddt.data(
326 {},
327 {'exec_errors': [
328 processutils.ProcessExecutionError(stderr='Cannot find path'),
329 processutils.ProcessExecutionError(stderr='Cannot find path')],
330 'expected_exception': exception.ServiceInstanceException},
331 {'exec_errors': [processutils.ProcessExecutionError(stderr='')],
332 'expected_exception': processutils.ProcessExecutionError},
333 {'exec_errors': [
334 processutils.ProcessExecutionError(stderr='Cannot find path'),
335 None]}
336 )
337 @ddt.unpack
338 def test_get_cbs_init_reg_section(self, exec_errors=None,
339 expected_exception=None):
340 self._windows_utils.normalize_path.return_value = (
341 mock.sentinel.normalized_section_path)
342 self._windows_utils.get_win_reg_value.side_effect = exec_errors
344 if expected_exception:
345 self.assertRaises(expected_exception,
346 self._mgr._get_cbs_init_reg_section,
347 mock.sentinel.server)
348 else:
349 cbs_init_section = self._mgr._get_cbs_init_reg_section(
350 mock.sentinel.server)
351 self.assertEqual(mock.sentinel.normalized_section_path,
352 cbs_init_section)
354 base_path = 'hklm:\\SOFTWARE'
355 cbs_section = 'Cloudbase Solutions\\Cloudbase-Init'
356 tested_upper_sections = ['']
358 if exec_errors and 'Cannot find path' in exec_errors[0].stderr:
359 tested_upper_sections.append('Wow6432Node')
361 tested_sections = [os.path.join(base_path,
362 upper_section,
363 cbs_section)
364 for upper_section in tested_upper_sections]
365 self._windows_utils.normalize_path.assert_has_calls(
366 [mock.call(tested_section)
367 for tested_section in tested_sections])