<s> import contextlib <EOL> import functools <EOL> import importlib <EOL> import os <EOL> try : <EOL> import unittest . mock as mock <EOL> except ImportError : <EOL> import mock <EOL> from oslo_config import cfg <EOL> import six <EOL> from cloudbaseinit import exception <EOL> from cloudbaseinit . tests import fake <EOL> from cloudbaseinit . tests import testutils <EOL> CONF = cfg . CONF <EOL> class WMIError ( Exception ) : <EOL> com_error = "<STR_LIT>" <EOL> class TestWindowsUtils ( testutils . CloudbaseInitTestBase ) : <EOL> '''<STR_LIT>''' <EOL> _CONFIG_NAME = '<STR_LIT>' <EOL> _DESTINATION = '<STR_LIT>' <EOL> _GATEWAY = '<STR_LIT>' <EOL> _NETMASK = '<STR_LIT>' <EOL> _PASSWORD = '<STR_LIT>' <EOL> _SECTION = '<STR_LIT>' <EOL> _USERNAME = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> self . _pywintypes_mock = mock . MagicMock ( ) <EOL> self . _pywintypes_mock . com_error = fake . FakeComError <EOL> self . _win32com_mock = mock . MagicMock ( ) <EOL> self . _win32process_mock = mock . MagicMock ( ) <EOL> self . _win32security_mock = mock . MagicMock ( ) <EOL> self . _win32net_mock = mock . MagicMock ( ) <EOL> self . _win32netcon_mock = mock . MagicMock ( ) <EOL> self . _wmi_mock = mock . MagicMock ( ) <EOL> self . _wmi_mock . x_wmi = WMIError <EOL> self . _moves_mock = mock . MagicMock ( ) <EOL> self . _xmlrpc_client_mock = mock . MagicMock ( ) <EOL> self . _ctypes_mock = mock . MagicMock ( ) <EOL> self . _tzlocal_mock = mock . Mock ( ) <EOL> self . _win32net_mock . error = Exception <EOL> module_path = "<STR_LIT>" <EOL> _module_patcher = mock . patch . dict ( <EOL> '<STR_LIT>' , <EOL> { '<STR_LIT>' : self . _win32com_mock , <EOL> '<STR_LIT>' : self . _win32process_mock , <EOL> '<STR_LIT>' : self . _win32security_mock , <EOL> '<STR_LIT>' : self . _win32net_mock , <EOL> '<STR_LIT>' : self . _win32netcon_mock , <EOL> '<STR_LIT>' : self . _wmi_mock , <EOL> '<STR_LIT>' : self . _moves_mock , <EOL> '<STR_LIT>' : self . _xmlrpc_client_mock , <EOL> '<STR_LIT>' : self . _ctypes_mock , <EOL> '<STR_LIT>' : self . _pywintypes_mock , <EOL> '<STR_LIT>' : self . _tzlocal_mock , <EOL> '<STR_LIT>' : mock . MagicMock ( ) } ) <EOL> _module_patcher . start ( ) <EOL> self . addCleanup ( _module_patcher . stop ) <EOL> exception . ctypes . GetLastError = mock . MagicMock ( ) <EOL> exception . ctypes . FormatError = mock . MagicMock ( ) <EOL> with mock . patch ( "<STR_LIT>" ) : <EOL> self . windows_utils = importlib . import_module ( module_path ) <EOL> self . _winreg_mock = self . _moves_mock . winreg <EOL> self . _windll_mock = self . _ctypes_mock . windll <EOL> self . _wintypes_mock = self . _ctypes_mock . wintypes <EOL> self . _client_mock = self . _win32com_mock . client <EOL> self . windows_utils . WindowsError = mock . MagicMock ( ) <EOL> self . _winutils = self . windows_utils . WindowsUtils ( ) <EOL> self . _kernel32 = self . _windll_mock . kernel32 <EOL> self . _iphlpapi = self . _windll_mock . iphlpapi <EOL> self . _ntdll = self . _windll_mock . ntdll <EOL> self . snatcher = testutils . LogSnatcher ( module_path ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_reboot ( self , mock_privilege_module , ret_value , <EOL> expected_ret_value = None ) : <EOL> mock_privilege_module . acquire_privilege = mock . MagicMock ( ) <EOL> advapi32 = self . _windll_mock . advapi32 <EOL> advapi32 . InitiateSystemShutdownExW = mock . MagicMock ( <EOL> return_value = ret_value ) <EOL> if not ret_value : <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , expected_ret_value ) : <EOL> self . _winutils . reboot ( ) <EOL> else : <EOL> self . _winutils . reboot ( ) <EOL> advapi32 . InitiateSystemShutdownExW . assert_called_with ( <EOL> <NUM_LIT:0> , <EOL> "<STR_LIT>" , <EOL> <NUM_LIT:0> , True , True , <NUM_LIT:0> ) <EOL> mock_privilege_module . acquire_privilege . assert_called_once_with ( <EOL> self . _win32security_mock . SE_SHUTDOWN_NAME ) <EOL> def test_reboot ( self ) : <EOL> self . _test_reboot ( ret_value = True ) <EOL> def test_reboot_failed ( self ) : <EOL> self . _test_reboot ( ret_value = None , expected_ret_value = <NUM_LIT:100> ) <EOL> def _test_get_user_info ( self , exc = None ) : <EOL> userget_mock = self . _win32net_mock . NetUserGetInfo <EOL> level = mock . Mock ( ) <EOL> ret = mock . Mock ( ) <EOL> if exc : <EOL> userget_mock . side_effect = [ exc ] <EOL> error_class = ( <EOL> exception . ItemNotFoundException if <EOL> exc . args [ <NUM_LIT:0> ] == self . _winutils . NERR_UserNotFound else <EOL> exception . CloudbaseInitException ) <EOL> with self . assertRaises ( error_class ) : <EOL> self . _winutils . _get_user_info ( self . _USERNAME , level ) <EOL> return <EOL> userget_mock . return_value = ret <EOL> response = self . _winutils . _get_user_info ( self . _USERNAME , level ) <EOL> userget_mock . assert_called_once_with ( None , self . _USERNAME , level ) <EOL> self . assertEqual ( ret , response ) <EOL> def test_get_user_info ( self ) : <EOL> self . _test_get_user_info ( ) <EOL> def test_get_user_info_not_found ( self ) : <EOL> exc = self . _win32net_mock . error ( self . _winutils . NERR_UserNotFound , <EOL> * ( [ mock . Mock ( ) ] * <NUM_LIT:2> ) ) <EOL> self . _test_get_user_info ( exc = exc ) <EOL> def test_get_user_info_failed ( self ) : <EOL> exc = self . _win32net_mock . error ( * ( [ mock . Mock ( ) ] * <NUM_LIT:3> ) ) <EOL> self . _test_get_user_info ( exc = exc ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_set_user_password ( self , mock_get_user_info , <EOL> expire = False , fail = False ) : <EOL> user_info = mock . MagicMock ( ) <EOL> mock_get_user_info . return_value = user_info <EOL> user_info [ "<STR_LIT:password>" ] = self . _PASSWORD <EOL> if expire : <EOL> user_info [ "<STR_LIT>" ] &= ~ self . _win32netcon_mock . UF_DONT_EXPIRE_PASSWD <EOL> else : <EOL> user_info [ "<STR_LIT>" ] |= self . _win32netcon_mock . UF_DONT_EXPIRE_PASSWD <EOL> if fail : <EOL> self . _win32net_mock . NetUserSetInfo . side_effect = [ <EOL> self . _win32net_mock . error ( * ( [ mock . Mock ( ) ] * <NUM_LIT:3> ) ) ] <EOL> with self . assertRaises ( exception . CloudbaseInitException ) : <EOL> self . _winutils . set_user_password ( self . _USERNAME , <EOL> self . _PASSWORD ) <EOL> return <EOL> self . _winutils . set_user_password ( self . _USERNAME , self . _PASSWORD , <EOL> password_expires = expire ) <EOL> mock_get_user_info . assert_called_once_with ( self . _USERNAME , <NUM_LIT:1> ) <EOL> self . _win32net_mock . NetUserSetInfo . assert_called_once_with ( <EOL> None , self . _USERNAME , <NUM_LIT:1> , user_info ) <EOL> def test_set_user_password ( self ) : <EOL> self . _test_set_user_password ( ) <EOL> def test_set_user_password_expire ( self ) : <EOL> self . _test_set_user_password ( expire = True ) <EOL> def test_set_user_password_fail ( self ) : <EOL> self . _test_set_user_password ( fail = True ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_change_password_next_logon ( self , mock_get_user_info , <EOL> fail = False ) : <EOL> user_info = mock . MagicMock ( ) <EOL> mock_get_user_info . return_value = user_info <EOL> user_info [ "<STR_LIT>" ] &= ~ self . _win32netcon_mock . UF_DONT_EXPIRE_PASSWD <EOL> user_info [ "<STR_LIT>" ] = <NUM_LIT:1> <EOL> if fail : <EOL> self . _win32net_mock . NetUserSetInfo . side_effect = [ <EOL> self . _win32net_mock . error ( * ( [ mock . Mock ( ) ] * <NUM_LIT:3> ) ) ] <EOL> with self . assertRaises ( exception . CloudbaseInitException ) : <EOL> self . _winutils . change_password_next_logon ( self . _USERNAME ) <EOL> return <EOL> self . _winutils . change_password_next_logon ( self . _USERNAME ) <EOL> self . _win32net_mock . NetUserSetInfo . assert_called_once_with ( <EOL> None , self . _USERNAME , <NUM_LIT:4> , user_info ) <EOL> def test_change_password_next_logon ( self ) : <EOL> self . _test_change_password_next_logon ( ) <EOL> def test_change_password_next_logon_fail ( self ) : <EOL> self . _test_change_password_next_logon ( fail = True ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_user_exists ( self , mock_get_user_info , exists ) : <EOL> if not exists : <EOL> mock_get_user_info . side_effect = [ exception . ItemNotFoundException ] <EOL> response = self . _winutils . user_exists ( self . _USERNAME ) <EOL> self . assertEqual ( False , response ) <EOL> return <EOL> response = self . _winutils . user_exists ( self . _USERNAME ) <EOL> mock_get_user_info . assert_called_once_with ( self . _USERNAME , <NUM_LIT:1> ) <EOL> self . assertEqual ( True , response ) <EOL> def test_user_exists ( self ) : <EOL> self . _test_user_exists ( exists = True ) <EOL> def test_user_does_not_exist ( self ) : <EOL> self . _test_user_exists ( exists = False ) <EOL> def test_sanitize_shell_input ( self ) : <EOL> unsanitised = '<STR_LIT>' <EOL> response = self . _winutils . sanitize_shell_input ( unsanitised ) <EOL> sanitised = '<STR_LIT>' <EOL> self . assertEqual ( sanitised , response ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_user_sid_and_domain ( self , mock_cch_referenced , <EOL> ret_val , last_error = None ) : <EOL> cbSid = mock . Mock ( ) <EOL> sid = mock . Mock ( ) <EOL> size = <NUM_LIT> <EOL> cchReferencedDomainName = mock . Mock ( ) <EOL> domainName = mock . Mock ( ) <EOL> sidNameUse = mock . Mock ( ) <EOL> advapi32 = self . _windll_mock . advapi32 <EOL> mock_cch_referenced . return_value = cchReferencedDomainName <EOL> self . _ctypes_mock . create_string_buffer . return_value = sid <EOL> self . _ctypes_mock . sizeof . return_value = size <EOL> self . _ctypes_mock . create_unicode_buffer . return_value = domainName <EOL> advapi32 . LookupAccountNameW . return_value = ret_val <EOL> if ret_val is None : <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , <EOL> last_error ) : <EOL> self . _winutils . _get_user_sid_and_domain ( <EOL> self . _USERNAME ) <EOL> else : <EOL> response = self . _winutils . _get_user_sid_and_domain ( self . _USERNAME ) <EOL> advapi32 . LookupAccountNameW . assert_called_with ( <EOL> <NUM_LIT:0> , six . text_type ( self . _USERNAME ) , sid , <EOL> self . _ctypes_mock . byref ( cbSid ) , domainName , <EOL> self . _ctypes_mock . byref ( cchReferencedDomainName ) , <EOL> self . _ctypes_mock . byref ( sidNameUse ) ) <EOL> self . assertEqual ( ( sid , domainName . value ) , response ) <EOL> mock_cch_referenced . assert_called_once_with ( <EOL> self . _ctypes_mock . create_unicode_buffer . return_value ) <EOL> def test_get_cch_referenced_domain_name ( self ) : <EOL> self . _ctypes_mock . sizeof . side_effect = [ <NUM_LIT> , <NUM_LIT> ] <EOL> result = self . _winutils . _get_cch_referenced_domain_name ( <EOL> mock . sentinel . domain_name ) <EOL> self . _wintypes_mock . DWORD . assert_called_once_with ( <NUM_LIT> // <NUM_LIT> ) <EOL> self . assertEqual ( result , self . _wintypes_mock . DWORD . return_value ) <EOL> def test_get_user_sid_and_domain ( self ) : <EOL> fake_obj = mock . Mock ( ) <EOL> self . _test_get_user_sid_and_domain ( ret_val = fake_obj ) <EOL> def test_get_user_sid_and_domain_no_return_value ( self ) : <EOL> self . _test_get_user_sid_and_domain ( ret_val = None , last_error = <NUM_LIT:100> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_add_user_to_local_group ( self , <EOL> mock_Win32_LOCALGROUP_MEMBERS_INFO_3 , <EOL> ret_value ) : <EOL> lmi = mock_Win32_LOCALGROUP_MEMBERS_INFO_3 ( ) <EOL> group_name = '<STR_LIT>' <EOL> netapi32 = self . _windll_mock . netapi32 <EOL> netapi32 . NetLocalGroupAddMembers . return_value = ret_value <EOL> is_in_alias = ret_value != self . _winutils . ERROR_MEMBER_IN_ALIAS <EOL> if ret_value is not <NUM_LIT:0> and is_in_alias : <EOL> self . assertRaises ( <EOL> exception . CloudbaseInitException , <EOL> self . _winutils . add_user_to_local_group , <EOL> self . _USERNAME , group_name ) <EOL> else : <EOL> self . _winutils . add_user_to_local_group ( self . _USERNAME , <EOL> group_name ) <EOL> netapi32 . NetLocalGroupAddMembers . assert_called_with ( <EOL> <NUM_LIT:0> , six . text_type ( group_name ) , <NUM_LIT:3> , <EOL> self . _ctypes_mock . addressof . return_value , <NUM_LIT:1> ) <EOL> self . _ctypes_mock . addressof . assert_called_once_with ( lmi ) <EOL> self . assertEqual ( lmi . lgrmi3_domainandname , <EOL> six . text_type ( self . _USERNAME ) ) <EOL> def test_add_user_to_local_group_no_error ( self ) : <EOL> self . _test_add_user_to_local_group ( ret_value = <NUM_LIT:0> ) <EOL> def test_add_user_to_local_group_not_found ( self ) : <EOL> self . _test_add_user_to_local_group ( <EOL> ret_value = self . _winutils . NERR_GroupNotFound ) <EOL> def test_add_user_to_local_group_access_denied ( self ) : <EOL> self . _test_add_user_to_local_group ( <EOL> ret_value = self . _winutils . ERROR_ACCESS_DENIED ) <EOL> def test_add_user_to_local_group_no_member ( self ) : <EOL> self . _test_add_user_to_local_group ( <EOL> ret_value = self . _winutils . ERROR_NO_SUCH_MEMBER ) <EOL> def test_add_user_to_local_group_member_in_alias ( self ) : <EOL> self . _test_add_user_to_local_group ( <EOL> ret_value = self . _winutils . ERROR_MEMBER_IN_ALIAS ) <EOL> def test_add_user_to_local_group_invalid_member ( self ) : <EOL> self . _test_add_user_to_local_group ( <EOL> ret_value = self . _winutils . ERROR_INVALID_MEMBER ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_user_sid ( self , mock_get_user_info , fail ) : <EOL> if fail : <EOL> mock_get_user_info . side_effect = [ exception . ItemNotFoundException ] <EOL> response = self . _winutils . get_user_sid ( self . _USERNAME ) <EOL> self . assertEqual ( None , response ) <EOL> return <EOL> user_info = mock . MagicMock ( ) <EOL> mock_get_user_info . return_value = user_info <EOL> response = self . _winutils . get_user_sid ( self . _USERNAME ) <EOL> mock_get_user_info . assert_called_once_with ( self . _USERNAME , <NUM_LIT:4> ) <EOL> self . assertEqual ( response , str ( user_info [ "<STR_LIT>" ] ) [ <NUM_LIT:6> : ] ) <EOL> def test_get_user_sid ( self ) : <EOL> self . _test_get_user_sid ( fail = False ) <EOL> def test_get_user_sid_fail ( self ) : <EOL> self . _test_get_user_sid ( fail = True ) <EOL> def _test_create_user ( self , expire = False , fail = False ) : <EOL> user_info = { <EOL> "<STR_LIT:name>" : self . _USERNAME , <EOL> "<STR_LIT:password>" : self . _PASSWORD , <EOL> "<STR_LIT>" : self . _win32netcon_mock . USER_PRIV_USER , <EOL> "<STR_LIT>" : ( self . _win32netcon_mock . UF_NORMAL_ACCOUNT | <EOL> self . _win32netcon_mock . UF_SCRIPT ) <EOL> } <EOL> if not expire : <EOL> user_info [ "<STR_LIT>" ] |= self . _win32netcon_mock . UF_DONT_EXPIRE_PASSWD <EOL> if fail : <EOL> self . _win32net_mock . NetUserAdd . side_effect = [ <EOL> self . _win32net_mock . error ( * ( [ mock . Mock ( ) ] * <NUM_LIT:3> ) ) ] <EOL> with self . assertRaises ( exception . CloudbaseInitException ) : <EOL> self . _winutils . create_user ( self . _USERNAME , <EOL> self . _PASSWORD ) <EOL> return <EOL> self . _winutils . create_user ( self . _USERNAME , self . _PASSWORD , <EOL> password_expires = expire ) <EOL> self . _win32net_mock . NetUserAdd . assert_called_once_with ( <EOL> None , <NUM_LIT:1> , user_info ) <EOL> def test_create_user ( self ) : <EOL> self . _test_create_user ( ) <EOL> def test_create_user_expire ( self ) : <EOL> self . _test_create_user ( expire = True ) <EOL> def test_create_user_fail ( self ) : <EOL> self . _test_create_user ( fail = True ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_create_user_logon_session ( self , mock_Win32_PROFILEINFO , logon , <EOL> loaduser , load_profile = True , <EOL> last_error = None ) : <EOL> self . _wintypes_mock . HANDLE = mock . MagicMock ( ) <EOL> pi = self . windows_utils . Win32_PROFILEINFO ( ) <EOL> advapi32 = self . _windll_mock . advapi32 <EOL> userenv = self . _windll_mock . userenv <EOL> kernel32 = self . _windll_mock . kernel32 <EOL> advapi32 . LogonUserW . return_value = logon <EOL> if not logon : <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , last_error ) : <EOL> self . _winutils . create_user_logon_session ( <EOL> self . _USERNAME , self . _PASSWORD , domain = '<STR_LIT:.>' , <EOL> load_profile = load_profile ) <EOL> elif load_profile and not loaduser : <EOL> userenv . LoadUserProfileW . return_value = None <EOL> kernel32 . CloseHandle . return_value = None <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , last_error ) : <EOL> self . _winutils . create_user_logon_session ( <EOL> self . _USERNAME , self . _PASSWORD , domain = '<STR_LIT:.>' , <EOL> load_profile = load_profile ) <EOL> userenv . LoadUserProfileW . assert_called_with ( <EOL> self . _wintypes_mock . HANDLE . return_value , <EOL> self . _ctypes_mock . byref . return_value ) <EOL> self . _ctypes_mock . byref . assert_called_with ( pi ) <EOL> kernel32 . CloseHandle . assert_called_with ( <EOL> self . _wintypes_mock . HANDLE . return_value ) <EOL> elif not load_profile : <EOL> response = self . _winutils . create_user_logon_session ( <EOL> self . _USERNAME , self . _PASSWORD , domain = '<STR_LIT:.>' , <EOL> load_profile = load_profile ) <EOL> self . assertTrue ( response is not None ) <EOL> else : <EOL> size = <NUM_LIT> <EOL> self . _ctypes_mock . sizeof . return_value = size <EOL> mock_Win32_PROFILEINFO . return_value = loaduser <EOL> response = self . _winutils . create_user_logon_session ( <EOL> self . _USERNAME , self . _PASSWORD , domain = '<STR_LIT:.>' , <EOL> load_profile = load_profile ) <EOL> userenv . LoadUserProfileW . assert_called_with ( <EOL> self . _wintypes_mock . HANDLE . return_value , <EOL> self . _ctypes_mock . byref . return_value ) <EOL> self . assertTrue ( response is not None ) <EOL> def test_create_user_logon_session_fail_load_false ( self ) : <EOL> self . _test_create_user_logon_session ( logon = <NUM_LIT:0> , loaduser = <NUM_LIT:0> , <EOL> load_profile = True , <EOL> last_error = <NUM_LIT:100> ) <EOL> def test_create_user_logon_session_fail_load_true ( self ) : <EOL> self . _test_create_user_logon_session ( logon = <NUM_LIT:0> , loaduser = <NUM_LIT:0> , <EOL> load_profile = False , <EOL> last_error = <NUM_LIT:100> ) <EOL> def test_create_user_logon_session_load_true ( self ) : <EOL> m = mock . Mock ( ) <EOL> n = mock . Mock ( ) <EOL> self . _test_create_user_logon_session ( logon = m , loaduser = n , <EOL> load_profile = True ) <EOL> def test_create_user_logon_session_load_false ( self ) : <EOL> m = mock . Mock ( ) <EOL> n = mock . Mock ( ) <EOL> self . _test_create_user_logon_session ( logon = m , loaduser = n , <EOL> load_profile = False ) <EOL> def test_create_user_logon_session_no_load_true ( self ) : <EOL> m = mock . Mock ( ) <EOL> self . _test_create_user_logon_session ( logon = m , loaduser = None , <EOL> load_profile = True , <EOL> last_error = <NUM_LIT:100> ) <EOL> def test_create_user_logon_session_no_load_false ( self ) : <EOL> m = mock . Mock ( ) <EOL> self . _test_create_user_logon_session ( logon = m , loaduser = None , <EOL> load_profile = False , <EOL> last_error = <NUM_LIT:100> ) <EOL> def test_close_user_logon_session ( self ) : <EOL> token = mock . Mock ( ) <EOL> self . _windll_mock . kernel32 . CloseHandle = mock . MagicMock ( ) <EOL> self . _winutils . close_user_logon_session ( token ) <EOL> self . _windll_mock . kernel32 . CloseHandle . assert_called_with ( token ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_set_host_name ( self , mock_SetComputerNameExW , ret_value , <EOL> last_error = None ) : <EOL> mock_SetComputerNameExW . return_value = ret_value <EOL> if not ret_value : <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , last_error ) : <EOL> self . _winutils . set_host_name ( '<STR_LIT>' ) <EOL> else : <EOL> self . assertTrue ( self . _winutils . set_host_name ( '<STR_LIT>' ) ) <EOL> mock_SetComputerNameExW . assert_called_with ( <EOL> self . _winutils . ComputerNamePhysicalDnsHostname , <EOL> six . text_type ( '<STR_LIT>' ) ) <EOL> def test_set_host_name ( self ) : <EOL> self . _test_set_host_name ( ret_value = '<STR_LIT>' ) <EOL> def test_set_host_exception ( self ) : <EOL> self . _test_set_host_name ( ret_value = None , last_error = <NUM_LIT:100> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_user_home ( self , user_sid , mock_get_user_sid ) : <EOL> mock_get_user_sid . return_value = user_sid <EOL> with self . snatcher : <EOL> response = self . _winutils . get_user_home ( self . _USERNAME ) <EOL> if user_sid : <EOL> mock_get_user_sid . assert_called_with ( self . _USERNAME ) <EOL> self . _winreg_mock . OpenKey . assert_called_with ( <EOL> self . _winreg_mock . HKEY_LOCAL_MACHINE , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % mock_get_user_sid ( ) ) <EOL> self . assertTrue ( response is not None ) <EOL> self . _winreg_mock . QueryValueEx . assert_called_with ( <EOL> self . _winreg_mock . OpenKey . return_value . __enter__ . return_value , <EOL> '<STR_LIT>' ) <EOL> else : <EOL> self . assertEqual ( <EOL> [ "<STR_LIT>" % self . _USERNAME ] , <EOL> self . snatcher . output ) <EOL> self . assertTrue ( response is None ) <EOL> def test_get_user_home ( self ) : <EOL> user = mock . MagicMock ( ) <EOL> self . _test_get_user_home ( user ) <EOL> def test_get_user_home_fail ( self ) : <EOL> self . _test_get_user_home ( None ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_network_adapters ( self , is_xp_2003 , mock_check_os_version ) : <EOL> conn = self . _wmi_mock . WMI <EOL> mock_response = mock . MagicMock ( ) <EOL> conn . return_value . query . return_value = [ mock_response ] <EOL> mock_check_os_version . return_value = not is_xp_2003 <EOL> wql = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if not is_xp_2003 : <EOL> wql += '<STR_LIT>' <EOL> response = self . _winutils . get_network_adapters ( ) <EOL> conn . return_value . query . assert_called_with ( wql ) <EOL> self . assertEqual ( [ ( mock_response . Name , mock_response . MACAddress ) ] , <EOL> response ) <EOL> def test_get_network_adapters ( self ) : <EOL> self . _test_get_network_adapters ( False ) <EOL> def test_get_network_adapters_xp_2003 ( self ) : <EOL> self . _test_get_network_adapters ( True ) <EOL> def _test_set_static_network_config ( self , adapter = True , static_val = ( <NUM_LIT:0> , ) , <EOL> gateway_val = ( <NUM_LIT:0> , ) , dns_val = ( <NUM_LIT:0> , ) ) : <EOL> conn = self . _wmi_mock . WMI <EOL> mac_address = '<STR_LIT>' <EOL> address = '<STR_LIT>' <EOL> broadcast = '<STR_LIT>' <EOL> dns_list = [ '<STR_LIT>' ] <EOL> set_static_call = functools . partial ( <EOL> self . _winutils . set_static_network_config , <EOL> mac_address , address , self . _NETMASK , <EOL> broadcast , self . _GATEWAY , dns_list <EOL> ) <EOL> if adapter : <EOL> adapter = mock . MagicMock ( ) <EOL> else : <EOL> self . assertRaises ( <EOL> exception . CloudbaseInitException , <EOL> set_static_call <EOL> ) <EOL> return <EOL> expected_log = [ ] <EOL> for ( ret_val , ) , msg in ( ( static_val , "<STR_LIT>" ) , <EOL> ( gateway_val , "<STR_LIT>" ) , <EOL> ( dns_val , "<STR_LIT>" ) ) : <EOL> if ret_val in ( <NUM_LIT:0> , <NUM_LIT:1> ) : <EOL> expected_log . append ( msg ) <EOL> conn . return_value . query . return_value = adapter <EOL> adapter_config = adapter [ <NUM_LIT:0> ] . associators . return_value [ <NUM_LIT:0> ] <EOL> adapter_config . EnableStatic . return_value = static_val <EOL> adapter_config . SetGateways . return_value = gateway_val <EOL> adapter_config . SetDNSServerSearchOrder . return_value = dns_val <EOL> adapter . __len__ . return_value = <NUM_LIT:1> <EOL> if static_val [ <NUM_LIT:0> ] > <NUM_LIT:1> or gateway_val [ <NUM_LIT:0> ] > <NUM_LIT:1> or dns_val [ <NUM_LIT:0> ] > <NUM_LIT:1> : <EOL> self . assertRaises ( <EOL> exception . CloudbaseInitException , <EOL> set_static_call ) <EOL> else : <EOL> with self . snatcher : <EOL> response = set_static_call ( ) <EOL> if static_val [ <NUM_LIT:0> ] or gateway_val [ <NUM_LIT:0> ] or dns_val [ <NUM_LIT:0> ] : <EOL> self . assertTrue ( response ) <EOL> else : <EOL> self . assertFalse ( response ) <EOL> self . assertEqual ( expected_log , self . snatcher . output ) <EOL> select = ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( mac_address ) ) <EOL> conn . return_value . query . assert_called_once_with ( select ) <EOL> adapter [ <NUM_LIT:0> ] . associators . assert_called_with ( <EOL> wmi_result_class = '<STR_LIT>' ) <EOL> adapter_config . EnableStatic . assert_called_with ( <EOL> [ address ] , [ self . _NETMASK ] ) <EOL> adapter_config . SetGateways . assert_called_with ( <EOL> [ self . _GATEWAY ] , [ <NUM_LIT:1> ] ) <EOL> adapter_config . SetDNSServerSearchOrder . assert_called_with ( <EOL> dns_list ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def _test_set_static_network_config_v6 ( self , mock_get_adapter_addresses , <EOL> mock_check_os_version , <EOL> v6adapters = True , v6error = False ) : <EOL> friendly_name = "<STR_LIT>" <EOL> interface_index = "<STR_LIT:4>" <EOL> mac_address = '<STR_LIT>' <EOL> address6 = "<STR_LIT>" <EOL> netmask6 = "<STR_LIT>" <EOL> gateway6 = "<STR_LIT>" <EOL> conn = self . _wmi_mock . WMI <EOL> netip = conn . return_value . query . return_value [ <NUM_LIT:0> ] <EOL> if v6error : <EOL> netip . Create . side_effect = WMIError <EOL> adapter_addresses = [ ] <EOL> if v6adapters : <EOL> adapter_addresses = [ <EOL> { <EOL> "<STR_LIT>" : mac_address , <EOL> "<STR_LIT>" : friendly_name , <EOL> "<STR_LIT>" : interface_index <EOL> } <EOL> ] <EOL> mock_get_adapter_addresses . return_value = adapter_addresses <EOL> mock_check_os_version . return_value = True <EOL> set_static_call = functools . partial ( <EOL> self . _winutils . set_static_network_config_v6 , <EOL> mac_address , address6 , netmask6 , gateway6 ) <EOL> expected_log = [ ] <EOL> if not mock_check_os_version . return_value : <EOL> expected_log . append ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if not v6adapters or v6error : <EOL> self . assertRaises ( <EOL> exception . CloudbaseInitException , <EOL> set_static_call ) <EOL> else : <EOL> expected_log . append ( "<STR_LIT>" % friendly_name ) <EOL> with self . snatcher : <EOL> set_static_call ( ) <EOL> mock_get_adapter_addresses . assert_called_once_with ( ) <EOL> select = ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( friendly_name ) ) <EOL> conn . return_value . query . assert_called_once_with ( select ) <EOL> params = { <EOL> "<STR_LIT>" : interface_index , <EOL> "<STR_LIT>" : friendly_name , <EOL> "<STR_LIT>" : address6 , <EOL> "<STR_LIT>" : self . windows_utils . AF_INET6 , <EOL> "<STR_LIT>" : netmask6 , <EOL> "<STR_LIT>" : self . windows_utils . UNICAST , <EOL> "<STR_LIT>" : self . windows_utils . MANUAL , <EOL> "<STR_LIT>" : self . windows_utils . MANUAL , <EOL> "<STR_LIT>" : self . windows_utils . PREFERRED_ADDR , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : gateway6 , <EOL> "<STR_LIT>" : None , <EOL> "<STR_LIT>" : False , <EOL> } <EOL> netip . Create . assert_called_once_with ( ** params ) <EOL> self . assertEqual ( expected_log , self . snatcher . output ) <EOL> def test_set_static_network_config ( self ) : <EOL> ret_val1 = ( <NUM_LIT:1> , ) <EOL> ret_val2 = ( <NUM_LIT:1> , ) <EOL> ret_val3 = ( <NUM_LIT:0> , ) <EOL> self . _test_set_static_network_config ( static_val = ret_val1 , <EOL> gateway_val = ret_val2 , <EOL> dns_val = ret_val3 ) <EOL> def test_set_static_network_config_query_fail ( self ) : <EOL> self . _test_set_static_network_config ( adapter = False ) <EOL> def test_set_static_network_config_cannot_set_ip ( self ) : <EOL> ret_val1 = ( <NUM_LIT:2> , ) <EOL> self . _test_set_static_network_config ( static_val = ret_val1 ) <EOL> def test_set_static_network_config_cannot_set_gateway ( self ) : <EOL> ret_val1 = ( <NUM_LIT:1> , ) <EOL> ret_val2 = ( <NUM_LIT:2> , ) <EOL> self . _test_set_static_network_config ( static_val = ret_val1 , <EOL> gateway_val = ret_val2 ) <EOL> def test_set_static_network_config_cannot_set_DNS ( self ) : <EOL> ret_val1 = ( <NUM_LIT:1> , ) <EOL> ret_val2 = ( <NUM_LIT:1> , ) <EOL> ret_val3 = ( <NUM_LIT:2> , ) <EOL> self . _test_set_static_network_config ( static_val = ret_val1 , <EOL> gateway_val = ret_val2 , <EOL> dns_val = ret_val3 ) <EOL> def test_set_static_network_config_no_reboot ( self ) : <EOL> ret_val1 = ( <NUM_LIT:0> , ) <EOL> ret_val2 = ( <NUM_LIT:0> , ) <EOL> ret_val3 = ( <NUM_LIT:0> , ) <EOL> self . _test_set_static_network_config ( static_val = ret_val1 , <EOL> gateway_val = ret_val2 , <EOL> dns_val = ret_val3 ) <EOL> def test_set_static_network_config_v6 ( self ) : <EOL> self . _test_set_static_network_config_v6 ( ) <EOL> def test_set_static_network_config_v6_no_adapters ( self ) : <EOL> self . _test_set_static_network_config_v6 ( v6adapters = False ) <EOL> def test_set_static_network_config_v6_error ( self ) : <EOL> self . _test_set_static_network_config_v6 ( v6error = True ) <EOL> def _test_get_config_key_name ( self , section ) : <EOL> response = self . _winutils . _get_config_key_name ( section ) <EOL> if section : <EOL> self . assertEqual ( self . _winutils . _config_key + section + '<STR_LIT:\\>' , <EOL> response ) <EOL> else : <EOL> self . assertEqual ( self . _winutils . _config_key , response ) <EOL> def test_get_config_key_name_with_section ( self ) : <EOL> self . _test_get_config_key_name ( self . _SECTION ) <EOL> def test_get_config_key_name_no_section ( self ) : <EOL> self . _test_get_config_key_name ( None ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_set_config_value ( self , value , mock_get_config_key_name ) : <EOL> key_name = ( self . _winutils . _config_key + self . _SECTION + '<STR_LIT:\\>' + <EOL> self . _CONFIG_NAME ) <EOL> mock_get_config_key_name . return_value = key_name <EOL> self . _winutils . set_config_value ( self . _CONFIG_NAME , value , <EOL> self . _SECTION ) <EOL> key = self . _winreg_mock . CreateKey . return_value . __enter__ . return_value <EOL> self . _winreg_mock . CreateKey . assert_called_with ( <EOL> self . _winreg_mock . HKEY_LOCAL_MACHINE , key_name ) <EOL> mock_get_config_key_name . assert_called_with ( self . _SECTION ) <EOL> if type ( value ) == int : <EOL> self . _winreg_mock . SetValueEx . assert_called_with ( <EOL> key , self . _CONFIG_NAME , <NUM_LIT:0> , self . _winreg_mock . REG_DWORD , value ) <EOL> else : <EOL> self . _winreg_mock . SetValueEx . assert_called_with ( <EOL> key , self . _CONFIG_NAME , <NUM_LIT:0> , self . _winreg_mock . REG_SZ , value ) <EOL> def test_set_config_value_int ( self ) : <EOL> self . _test_set_config_value ( <NUM_LIT:1> ) <EOL> def test_set_config_value_not_int ( self ) : <EOL> self . _test_set_config_value ( '<STR_LIT:1>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_config_value ( self , value , mock_get_config_key_name ) : <EOL> key_name = self . _winutils . _config_key + self . _SECTION + '<STR_LIT:\\>' <EOL> key_name += self . _CONFIG_NAME <EOL> if type ( value ) == int : <EOL> regtype = self . _winreg_mock . REG_DWORD <EOL> else : <EOL> regtype = self . _winreg_mock . REG_SZ <EOL> self . _winreg_mock . QueryValueEx . return_value = ( value , regtype ) <EOL> if value is None : <EOL> mock_get_config_key_name . side_effect = [ <EOL> self . windows_utils . WindowsError ] <EOL> response = self . _winutils . get_config_value ( self . _CONFIG_NAME , <EOL> self . _SECTION ) <EOL> self . assertEqual ( None , response ) <EOL> else : <EOL> mock_get_config_key_name . return_value = key_name <EOL> response = self . _winutils . get_config_value ( self . _CONFIG_NAME , <EOL> self . _SECTION ) <EOL> self . _winreg_mock . OpenKey . assert_called_with ( <EOL> self . _winreg_mock . HKEY_LOCAL_MACHINE , key_name ) <EOL> mock_get_config_key_name . assert_called_with ( self . _SECTION ) <EOL> self . _winreg_mock . QueryValueEx . assert_called_with ( <EOL> self . _winreg_mock . OpenKey ( ) . __enter__ ( ) , self . _CONFIG_NAME ) <EOL> self . assertEqual ( value , response ) <EOL> def test_get_config_value_type_int ( self ) : <EOL> self . _test_get_config_value ( <NUM_LIT:1> ) <EOL> def test_get_config_value_type_str ( self ) : <EOL> self . _test_get_config_value ( '<STR_LIT>' ) <EOL> def test_get_config_value_type_error ( self ) : <EOL> self . _test_get_config_value ( None ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_wait_for_boot_completion ( self , _ , ret_vals = None ) : <EOL> self . _winreg_mock . QueryValueEx . side_effect = ret_vals <EOL> with self . snatcher : <EOL> self . _winutils . wait_for_boot_completion ( ) <EOL> key = self . _winreg_mock . OpenKey . return_value . __enter__ . return_value <EOL> self . _winreg_mock . OpenKey . assert_called_with ( <EOL> self . _winreg_mock . HKEY_LOCAL_MACHINE , <EOL> "<STR_LIT>" , <NUM_LIT:0> , <EOL> self . _winreg_mock . KEY_READ ) <EOL> expected_log = [ ] <EOL> for gen_states in ret_vals : <EOL> gen_state = gen_states [ <NUM_LIT:0> ] <EOL> if gen_state == <NUM_LIT:7> : <EOL> break <EOL> expected_log . append ( '<STR_LIT>' <EOL> '<STR_LIT>' % gen_state ) <EOL> self . _winreg_mock . QueryValueEx . assert_called_with ( <EOL> key , "<STR_LIT>" ) <EOL> self . assertEqual ( expected_log , self . snatcher . output ) <EOL> def test_wait_for_boot_completion ( self ) : <EOL> ret_vals = [ [ <NUM_LIT:7> ] ] <EOL> self . _test_wait_for_boot_completion ( ret_vals = ret_vals ) <EOL> def test_wait_for_boot_completion_wait ( self ) : <EOL> ret_vals = [ [ <NUM_LIT:1> ] , [ <NUM_LIT:7> ] ] <EOL> self . _test_wait_for_boot_completion ( ret_vals = ret_vals ) <EOL> def test_get_service ( self ) : <EOL> conn = self . _wmi_mock . WMI <EOL> conn . return_value . Win32_Service . return_value = [ '<STR_LIT>' ] <EOL> response = self . _winutils . _get_service ( '<STR_LIT>' ) <EOL> conn . assert_called_with ( moniker = '<STR_LIT>' ) <EOL> conn . return_value . Win32_Service . assert_called_with ( Name = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , response ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_check_service_exists ( self , mock_get_service ) : <EOL> mock_get_service . return_value = '<STR_LIT>' <EOL> response = self . _winutils . check_service_exists ( '<STR_LIT>' ) <EOL> self . assertTrue ( response ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_get_service_status ( self , mock_get_service ) : <EOL> mock_service = mock . MagicMock ( ) <EOL> mock_get_service . return_value = mock_service <EOL> response = self . _winutils . get_service_status ( '<STR_LIT>' ) <EOL> self . assertEqual ( mock_service . State , response ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_get_service_start_mode ( self , mock_get_service ) : <EOL> mock_service = mock . MagicMock ( ) <EOL> mock_get_service . return_value = mock_service <EOL> response = self . _winutils . get_service_start_mode ( '<STR_LIT>' ) <EOL> self . assertEqual ( mock_service . StartMode , response ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_set_service_start_mode ( self , mock_get_service , ret_val ) : <EOL> mock_service = mock . MagicMock ( ) <EOL> mock_get_service . return_value = mock_service <EOL> mock_service . ChangeStartMode . return_value = ( ret_val , ) <EOL> if ret_val != <NUM_LIT:0> : <EOL> self . assertRaises ( exception . CloudbaseInitException , <EOL> self . _winutils . set_service_start_mode , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> else : <EOL> self . _winutils . set_service_start_mode ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mock_service . ChangeStartMode . assert_called_once_with ( '<STR_LIT>' ) <EOL> def test_set_service_start_mode ( self ) : <EOL> self . _test_set_service_start_mode ( ret_val = <NUM_LIT:0> ) <EOL> def test_set_service_start_mode_exception ( self ) : <EOL> self . _test_set_service_start_mode ( ret_val = <NUM_LIT:1> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_start_service ( self , mock_get_service , ret_val ) : <EOL> mock_service = mock . MagicMock ( ) <EOL> mock_get_service . return_value = mock_service <EOL> mock_service . StartService . return_value = ( ret_val , ) <EOL> if ret_val != <NUM_LIT:0> : <EOL> self . assertRaises ( exception . CloudbaseInitException , <EOL> self . _winutils . start_service , <EOL> '<STR_LIT>' ) <EOL> else : <EOL> with self . snatcher : <EOL> self . _winutils . start_service ( '<STR_LIT>' ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , <EOL> self . snatcher . output ) <EOL> mock_service . StartService . assert_called_once_with ( ) <EOL> def test_start_service ( self ) : <EOL> self . _test_set_service_start_mode ( ret_val = <NUM_LIT:0> ) <EOL> def test_start_service_exception ( self ) : <EOL> self . _test_set_service_start_mode ( ret_val = <NUM_LIT:1> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_stop_service ( self , mock_get_service , ret_val ) : <EOL> mock_service = mock . MagicMock ( ) <EOL> mock_get_service . return_value = mock_service <EOL> mock_service . StopService . return_value = ( ret_val , ) <EOL> if ret_val != <NUM_LIT:0> : <EOL> self . assertRaises ( exception . CloudbaseInitException , <EOL> self . _winutils . stop_service , <EOL> '<STR_LIT>' ) <EOL> else : <EOL> with self . snatcher : <EOL> self . _winutils . stop_service ( '<STR_LIT>' ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , <EOL> self . snatcher . output ) <EOL> mock_service . StopService . assert_called_once_with ( ) <EOL> def test_stop_service ( self ) : <EOL> self . _test_stop_service ( ret_val = <NUM_LIT:0> ) <EOL> def test_stop_service_exception ( self ) : <EOL> self . _test_stop_service ( ret_val = <NUM_LIT:1> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_terminate ( self , mock_sleep , mock_stop_service ) : <EOL> self . _winutils . terminate ( ) <EOL> mock_stop_service . assert_called_with ( self . _winutils . _service_name ) <EOL> mock_sleep . assert_called_with ( <NUM_LIT:3> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_default_gateway ( self , mock_get_ipv4_routing_table , <EOL> routing_table ) : <EOL> mock_get_ipv4_routing_table . return_value = [ routing_table ] <EOL> response = self . _winutils . get_default_gateway ( ) <EOL> mock_get_ipv4_routing_table . assert_called_once_with ( ) <EOL> if routing_table [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> self . assertEqual ( ( routing_table [ <NUM_LIT:3> ] , routing_table [ <NUM_LIT:2> ] ) , response ) <EOL> else : <EOL> self . assertEqual ( ( None , None ) , response ) <EOL> def test_get_default_gateway ( self ) : <EOL> routing_table = [ '<STR_LIT>' , '<STR_LIT>' , self . _GATEWAY , '<STR_LIT>' ] <EOL> self . _test_get_default_gateway ( routing_table = routing_table ) <EOL> def test_get_default_gateway_error ( self ) : <EOL> routing_table = [ '<STR_LIT>' , '<STR_LIT>' , self . _GATEWAY , '<STR_LIT>' ] <EOL> self . _test_get_default_gateway ( routing_table = routing_table ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_check_static_route_exists ( self , mock_get_ipv4_routing_table , <EOL> routing_table ) : <EOL> mock_get_ipv4_routing_table . return_value = [ routing_table ] <EOL> response = self . _winutils . check_static_route_exists ( self . _DESTINATION ) <EOL> mock_get_ipv4_routing_table . assert_called_once_with ( ) <EOL> if routing_table [ <NUM_LIT:0> ] == self . _DESTINATION : <EOL> self . assertTrue ( response ) <EOL> else : <EOL> self . assertFalse ( response ) <EOL> def test_check_static_route_exists_true ( self ) : <EOL> routing_table = [ self . _DESTINATION , '<STR_LIT>' , self . _GATEWAY , <EOL> '<STR_LIT>' ] <EOL> self . _test_check_static_route_exists ( routing_table = routing_table ) <EOL> def test_check_static_route_exists_false ( self ) : <EOL> routing_table = [ '<STR_LIT>' , '<STR_LIT>' , self . _GATEWAY , '<STR_LIT>' ] <EOL> self . _test_check_static_route_exists ( routing_table = routing_table ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_add_static_route ( self , mock_execute_process , err ) : <EOL> next_hop = '<STR_LIT>' <EOL> interface_index = <NUM_LIT:1> <EOL> metric = <NUM_LIT:9> <EOL> args = [ '<STR_LIT>' , '<STR_LIT>' , self . _DESTINATION , '<STR_LIT>' , self . _NETMASK , <EOL> next_hop ] <EOL> mock_execute_process . return_value = ( None , err , None ) <EOL> if err : <EOL> self . assertRaises ( exception . CloudbaseInitException , <EOL> self . _winutils . add_static_route , <EOL> self . _DESTINATION , self . _NETMASK , next_hop , <EOL> interface_index , metric ) <EOL> else : <EOL> self . _winutils . add_static_route ( self . _DESTINATION , self . _NETMASK , <EOL> next_hop , interface_index , metric ) <EOL> mock_execute_process . assert_called_with ( args ) <EOL> def test_add_static_route ( self ) : <EOL> self . _test_add_static_route ( err = <NUM_LIT> ) <EOL> def test_add_static_route_fail ( self ) : <EOL> self . _test_add_static_route ( err = None ) <EOL> def _test_check_os_version ( self , ret_val = None , fail = False ) : <EOL> params = ( <NUM_LIT:3> , <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> mock_version_info = self . _ntdll . RtlVerifyVersionInfo <EOL> mock_version_info . return_value = ret_val <EOL> mask = mock . Mock ( ) <EOL> mock_condition_mask = self . _kernel32 . VerSetConditionMask <EOL> mock_condition_mask . return_value = mask <EOL> ver_constants = [ <EOL> self . windows_utils . VER_MAJORVERSION , <EOL> self . windows_utils . VER_MINORVERSION , <EOL> self . windows_utils . VER_BUILDNUMBER ] <EOL> if fail : <EOL> with self . assertRaises ( exception . CloudbaseInitException ) : <EOL> self . _winutils . check_os_version ( * params ) <EOL> return <EOL> osversion_struct = self . windows_utils . Win32_OSVERSIONINFOEX_W <EOL> osversion_struct . return_value = osversion_struct ( ) <EOL> osversion_struct . side_effect = None <EOL> vi = osversion_struct ( ) <EOL> mask_calls = [ <EOL> mock . call ( <NUM_LIT:0> , ver_constants [ <NUM_LIT:0> ] , <EOL> self . windows_utils . VER_GREATER_EQUAL ) , <EOL> mock . call ( mask , ver_constants [ <NUM_LIT:1> ] , <EOL> self . windows_utils . VER_GREATER_EQUAL ) , <EOL> mock . call ( mask , ver_constants [ <NUM_LIT:2> ] , <EOL> self . windows_utils . VER_GREATER_EQUAL ) ] <EOL> response = self . _winutils . check_os_version ( * params ) <EOL> self . _ctypes_mock . sizeof . assert_called_once_with ( osversion_struct ) <EOL> mock_condition_mask . assert_has_calls ( mask_calls ) <EOL> type_mask = ver_constants [ <NUM_LIT:0> ] | ver_constants [ <NUM_LIT:1> ] | ver_constants [ <NUM_LIT:2> ] <EOL> self . _ctypes_mock . byref . assert_called_once_with ( vi ) <EOL> mock_version_info . assert_called_once_with ( <EOL> self . _ctypes_mock . byref . return_value , type_mask , mask ) <EOL> expect = None <EOL> if not ret_val : <EOL> expect = True <EOL> elif ret_val == self . _winutils . STATUS_REVISION_MISMATCH : <EOL> expect = False <EOL> self . assertEqual ( expect , response ) <EOL> def test_check_os_version_pass ( self ) : <EOL> self . _test_check_os_version ( ret_val = <NUM_LIT:0> ) <EOL> def test_check_os_version_no_pass ( self ) : <EOL> self . _test_check_os_version ( <EOL> ret_val = self . _winutils . STATUS_REVISION_MISMATCH ) <EOL> def test_check_os_version_fail ( self ) : <EOL> self . _test_check_os_version ( ret_val = mock . Mock ( ) , fail = True ) <EOL> def _test_get_volume_label ( self , ret_val ) : <EOL> label = mock . MagicMock ( ) <EOL> max_label_size = <NUM_LIT> <EOL> drive = '<STR_LIT>' <EOL> self . _ctypes_mock . create_unicode_buffer . return_value = label <EOL> self . _windll_mock . kernel32 . GetVolumeInformationW . return_value = ret_val <EOL> response = self . _winutils . get_volume_label ( drive ) <EOL> if ret_val : <EOL> self . assertTrue ( response is not None ) <EOL> else : <EOL> self . assertTrue ( response is None ) <EOL> self . _ctypes_mock . create_unicode_buffer . assert_called_with ( <EOL> max_label_size ) <EOL> self . _windll_mock . kernel32 . GetVolumeInformationW . assert_called_with ( <EOL> drive , label , max_label_size , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> def test_get_volume_label ( self ) : <EOL> self . _test_get_volume_label ( '<STR_LIT>' ) <EOL> def test_get_volume_label_no_return_value ( self ) : <EOL> self . _test_get_volume_label ( None ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_generate_random_password ( self , mock_generate_random_password , <EOL> mock_search ) : <EOL> length = <NUM_LIT> <EOL> mock_search . return_value = True <EOL> mock_generate_random_password . return_value = '<STR_LIT>' <EOL> response = self . _winutils . generate_random_password ( length ) <EOL> mock_generate_random_password . assert_called_once_with ( length ) <EOL> self . assertEqual ( '<STR_LIT>' , response ) <EOL> def _test_get_logical_drives ( self , buf_length , last_error = None ) : <EOL> mock_buf = mock . MagicMock ( ) <EOL> mock_buf . __getitem__ . side_effect = [ '<STR_LIT:1>' , '<STR_LIT:\x00>' ] <EOL> mock_get_drives = self . _windll_mock . kernel32 . GetLogicalDriveStringsW <EOL> self . _ctypes_mock . create_unicode_buffer . return_value = mock_buf <EOL> mock_get_drives . return_value = buf_length <EOL> if buf_length is None : <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , last_error ) : <EOL> self . _winutils . _get_logical_drives ( ) <EOL> else : <EOL> response = self . _winutils . _get_logical_drives ( ) <EOL> self . _ctypes_mock . create_unicode_buffer . assert_called_with ( <NUM_LIT> ) <EOL> mock_get_drives . assert_called_with ( <NUM_LIT> , mock_buf ) <EOL> self . assertEqual ( [ '<STR_LIT:1>' ] , response ) <EOL> def test_get_logical_drives_exception ( self ) : <EOL> self . _test_get_logical_drives ( buf_length = None , last_error = <NUM_LIT:100> ) <EOL> def test_get_logical_drives ( self ) : <EOL> self . _test_get_logical_drives ( buf_length = <NUM_LIT:2> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_cdrom_drives ( self , mock_kernel32 , mock_get_logical_drives ) : <EOL> mock_get_logical_drives . return_value = [ '<STR_LIT>' ] <EOL> mock_kernel32 . GetDriveTypeW . return_value = self . _winutils . DRIVE_CDROM <EOL> response = self . _winutils . get_cdrom_drives ( ) <EOL> mock_get_logical_drives . assert_called_with ( ) <EOL> self . assertEqual ( [ '<STR_LIT>' ] , response ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_get_physical_disks ( self , mock_sdn , mock_setupapi , mock_kernel32 , <EOL> mock_msvcrt , handle_disks , last_error , <EOL> interface_detail , disk_handle , io_control , <EOL> last_error_code = None ) : <EOL> sizeof_calls = [ <EOL> mock . call ( <EOL> self . windows_utils . Win32_SP_DEVICE_INTERFACE_DATA <EOL> ) , <EOL> mock . call ( <EOL> self . windows_utils . Win32_SP_DEVICE_INTERFACE_DETAIL_DATA_W <EOL> ) , <EOL> mock . call ( mock_sdn ( ) ) <EOL> ] <EOL> device_interfaces_calls = [ <EOL> mock . call ( <EOL> handle_disks , None , self . _ctypes_mock . byref . return_value , <NUM_LIT:0> , <EOL> self . _ctypes_mock . byref . return_value ) , <EOL> mock . call ( handle_disks , None , <EOL> self . _ctypes_mock . byref . return_value , <NUM_LIT:1> , <EOL> self . _ctypes_mock . byref . return_value ) ] <EOL> device_path = self . _ctypes_mock . cast . return_value . contents . DevicePath <EOL> cast_calls = [ mock . call ( mock_msvcrt . malloc ( ) , <EOL> self . _ctypes_mock . POINTER . return_value ) , <EOL> mock . call ( device_path , <EOL> self . _wintypes_mock . LPWSTR ) ] <EOL> mock_setup_interface = mock_setupapi . SetupDiGetDeviceInterfaceDetailW <EOL> mock_setupapi . SetupDiGetClassDevsW . return_value = handle_disks <EOL> mock_kernel32 . GetLastError . return_value = last_error <EOL> mock_setup_interface . return_value = interface_detail <EOL> mock_kernel32 . CreateFileW . return_value = disk_handle <EOL> mock_kernel32 . DeviceIoControl . return_value = io_control <EOL> mock_setupapi . SetupDiEnumDeviceInterfaces . side_effect = [ True , False ] <EOL> if handle_disks == self . _winutils . INVALID_HANDLE_VALUE or ( <EOL> last_error != self . _winutils . ERROR_INSUFFICIENT_BUFFER ) and not ( <EOL> interface_detail ) or ( <EOL> disk_handle == self . _winutils . INVALID_HANDLE_VALUE ) or ( <EOL> not io_control ) : <EOL> if not io_control : <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , last_error_code ) : <EOL> self . _winutils . get_physical_disks ( ) <EOL> elif not interface_detail : <EOL> with self . assert_raises_windows_message ( <EOL> "<STR_LIT>" , <EOL> last_error_code ) : <EOL> self . _winutils . get_physical_disks ( ) <EOL> else : <EOL> self . assertRaises ( exception . CloudbaseInitException , <EOL> self . _winutils . get_physical_disks ) <EOL> else : <EOL> response = self . _winutils . get_physical_disks ( ) <EOL> self . assertEqual ( sizeof_calls , <EOL> self . _ctypes_mock . sizeof . call_args_list ) <EOL> self . assertEqual ( <EOL> device_interfaces_calls , <EOL> mock_setupapi . SetupDiEnumDeviceInterfaces . call_args_list ) <EOL> if not interface_detail : <EOL> mock_kernel32 . GetLastError . assert_called_once_with ( ) <EOL> self . _ctypes_mock . POINTER . assert_called_with ( <EOL> self . windows_utils . Win32_SP_DEVICE_INTERFACE_DETAIL_DATA_W ) <EOL> mock_msvcrt . malloc . assert_called_with ( <EOL> self . _ctypes_mock . c_size_t . return_value <EOL> ) <EOL> self . assertEqual ( cast_calls , self . _ctypes_mock . cast . call_args_list ) <EOL> mock_setup_interface . assert_called_with ( <EOL> handle_disks , self . _ctypes_mock . byref . return_value , <EOL> self . _ctypes_mock . cast . return_value , <EOL> self . _wintypes_mock . DWORD . return_value , <EOL> None , None ) <EOL> mock_kernel32 . CreateFileW . assert_called_with ( <EOL> self . _ctypes_mock . cast . return_value . value , <NUM_LIT:0> , <EOL> self . _winutils . FILE_SHARE_READ , None , <EOL> self . _winutils . OPEN_EXISTING , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> mock_sdn . assert_called_with ( ) <EOL> mock_kernel32 . DeviceIoControl . assert_called_with ( <EOL> disk_handle , self . _winutils . IOCTL_STORAGE_GET_DEVICE_NUMBER , <EOL> None , <NUM_LIT:0> , self . _ctypes_mock . byref . return_value , <EOL> self . _ctypes_mock . sizeof . return_value , <EOL> self . _ctypes_mock . byref . return_value , None ) <EOL> self . assertEqual ( [ "<STR_LIT>" ] , response ) <EOL> mock_setupapi . SetupDiDestroyDeviceInfoList . assert_called_once_with ( <EOL> handle_disks ) <EOL> mock_setupapi . SetupDiGetClassDevsW . assert_called_once_with ( <EOL> self . _ctypes_mock . byref . return_value , None , None , <EOL> self . _winutils . DIGCF_PRESENT | <EOL> self . _winutils . DIGCF_DEVICEINTERFACE ) <EOL> def test_get_physical_disks ( self ) : <EOL> mock_handle_disks = mock . MagicMock ( ) <EOL> mock_disk_handle = mock . MagicMock ( ) <EOL> self . _test_get_physical_disks ( <EOL> handle_disks = mock_handle_disks , <EOL> last_error = self . _winutils . ERROR_INSUFFICIENT_BUFFER , <EOL> interface_detail = '<STR_LIT>' , <EOL> disk_handle = mock_disk_handle , io_control = True ) <EOL> def test_get_physical_disks_other_error_and_no_interface_detail ( self ) : <EOL> mock_handle_disks = mock . MagicMock ( ) <EOL> mock_disk_handle = mock . MagicMock ( ) <EOL> self . _test_get_physical_disks ( <EOL> handle_disks = mock_handle_disks , <EOL> last_error = '<STR_LIT>' , interface_detail = None , <EOL> last_error_code = <NUM_LIT:100> , <EOL> disk_handle = mock_disk_handle , io_control = True ) <EOL> def test_get_physical_disks_invalid_disk_handle ( self ) : <EOL> mock_handle_disks = mock . MagicMock ( ) <EOL> self . _test_get_physical_disks ( <EOL> handle_disks = mock_handle_disks , <EOL> last_error = self . _winutils . ERROR_INSUFFICIENT_BUFFER , <EOL> interface_detail = '<STR_LIT>' , <EOL> disk_handle = self . _winutils . INVALID_HANDLE_VALUE , io_control = True ) <EOL> def test_get_physical_disks_io_control ( self ) : <EOL> mock_handle_disks = mock . MagicMock ( ) <EOL> mock_disk_handle = mock . MagicMock ( ) <EOL> self . _test_get_physical_disks ( <EOL> handle_disks = mock_handle_disks , <EOL> last_error = self . _winutils . ERROR_INSUFFICIENT_BUFFER , <EOL> interface_detail = '<STR_LIT>' , <EOL> last_error_code = <NUM_LIT:100> , <EOL> disk_handle = mock_disk_handle , io_control = False ) <EOL> def test_get_physical_disks_handle_disks_invalid ( self ) : <EOL> mock_disk_handle = mock . MagicMock ( ) <EOL> self . _test_get_physical_disks ( <EOL> handle_disks = self . _winutils . INVALID_HANDLE_VALUE , <EOL> last_error = self . _winutils . ERROR_INSUFFICIENT_BUFFER , <EOL> interface_detail = '<STR_LIT>' , <EOL> disk_handle = mock_disk_handle , io_control = True ) <EOL> def _test_get_volumes ( self , find_first = True , find_next = True ) : <EOL> count = <NUM_LIT:3> <EOL> expected_volumes = [ "<STR_LIT>" . format ( idx ) for idx in range ( count ) ] <EOL> volume_values = mock . PropertyMock ( side_effect = expected_volumes ) <EOL> mock_volume = mock . Mock ( ) <EOL> type ( mock_volume ) . value = volume_values <EOL> self . _ctypes_mock . create_unicode_buffer . return_value = mock_volume <EOL> if not find_first : <EOL> self . _kernel32 . FindFirstVolumeW . return_value = self . _winutils . INVALID_HANDLE_VALUE <EOL> side_effects = [ <NUM_LIT:1> ] * ( count - <NUM_LIT:1> ) <EOL> side_effects . append ( <NUM_LIT:0> ) <EOL> self . _kernel32 . FindNextVolumeW . side_effect = side_effects <EOL> if find_next : <EOL> self . _ctypes_mock . GetLastError . return_value = self . _winutils . ERROR_NO_MORE_FILES <EOL> if not ( find_first and find_next ) : <EOL> with self . assertRaises ( exception . WindowsCloudbaseInitException ) : <EOL> self . _winutils . get_volumes ( ) <EOL> return <EOL> volumes = self . _winutils . get_volumes ( ) <EOL> self . _kernel32 . FindFirstVolumeW . assert_called_once_with ( <EOL> mock_volume , self . _winutils . MAX_PATH ) <EOL> find_next_calls = [ <EOL> mock . call ( self . _kernel32 . FindFirstVolumeW . return_value , <EOL> mock_volume , self . _winutils . MAX_PATH ) ] * count <EOL> self . _kernel32 . FindNextVolumeW . assert_has_calls ( find_next_calls ) <EOL> self . _kernel32 . FindVolumeClose . assert_called_once_with ( <EOL> self . _kernel32 . FindFirstVolumeW . return_value ) <EOL> self . assertEqual ( expected_volumes , volumes ) <EOL> def test_get_volumes ( self ) : <EOL> self . _test_get_volumes ( ) <EOL> def test_get_volumes_first_fail ( self ) : <EOL> self . _test_get_volumes ( find_first = False ) <EOL> def test_get_volumes_next_fail ( self ) : <EOL> self . _test_get_volumes ( find_next = False ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_firewall_create_rule ( self , mock_get_fw_protocol ) : <EOL> expected = [ mock . call ( "<STR_LIT>" ) , <EOL> mock . call ( "<STR_LIT>" ) ] <EOL> self . _winutils . firewall_create_rule ( <EOL> name = '<STR_LIT>' , port = <NUM_LIT> , protocol = self . _winutils . PROTOCOL_TCP ) <EOL> self . assertEqual ( expected , self . _client_mock . Dispatch . call_args_list ) <EOL> mock_get_fw_protocol . assert_called_once_with ( <EOL> self . _winutils . PROTOCOL_TCP ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_firewall_remove_rule ( self , mock_get_fw_protocol ) : <EOL> self . _winutils . firewall_remove_rule ( <EOL> name = '<STR_LIT>' , port = <NUM_LIT> , protocol = self . _winutils . PROTOCOL_TCP ) <EOL> self . _client_mock . Dispatch . assert_called_once_with ( "<STR_LIT>" ) <EOL> mock_get_fw_protocol . assert_called_once_with ( <EOL> self . _winutils . PROTOCOL_TCP ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_system32_dir ( self , mock_expandvars ) : <EOL> path = "<STR_LIT>" <EOL> mock_expandvars . return_value = path <EOL> response = self . _winutils . get_system32_dir ( ) <EOL> mock_expandvars . assert_called_once_with ( '<STR_LIT>' . format ( path ) ) <EOL> self . assertEqual ( path , response ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_syswow64_dir ( self , mock_expandvars ) : <EOL> path = "<STR_LIT>" <EOL> mock_expandvars . return_value = path <EOL> response = self . _winutils . get_syswow64_dir ( ) <EOL> mock_expandvars . assert_called_once_with ( '<STR_LIT>' . format ( path ) ) <EOL> self . assertEqual ( path , response ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_sysnative_dir ( self , mock_expandvars ) : <EOL> path = "<STR_LIT>" <EOL> mock_expandvars . return_value = path <EOL> response = self . _winutils . get_sysnative_dir ( ) <EOL> mock_expandvars . assert_called_once_with ( '<STR_LIT>' . format ( path ) ) <EOL> self . assertEqual ( path , response ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_check_sysnative_dir_exists ( self , mock_isdir , <EOL> mock_get_sysnative_dir , <EOL> mock_is_wow64 , <EOL> exists = True , wow64 = False ) : <EOL> mock_get_sysnative_dir . return_value = '<STR_LIT>' <EOL> mock_isdir . return_value = exists <EOL> mock_is_wow64 . return_value = wow64 <EOL> with self . snatcher : <EOL> response = self . _winutils . check_sysnative_dir_exists ( ) <EOL> expected_log = [ ] <EOL> if not exists and wow64 : <EOL> expected_log = [ '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ] <EOL> self . assertEqual ( expected_log , self . snatcher . output ) <EOL> self . assertEqual ( exists , response ) <EOL> def test_check_sysnative_dir_exists ( self ) : <EOL> self . _test_check_sysnative_dir_exists ( ) <EOL> def test_check_sysnative_dir_does_not_exist ( self ) : <EOL> self . _test_check_sysnative_dir_exists ( exists = False , wow64 = True ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_system_dir ( self , mock_get_system32_dir , <EOL> mock_get_syswow64_dir , <EOL> mock_is_64bit_arch , <EOL> mock_get_sysnative_dir , <EOL> mock_check_sysnative_dir_exists , <EOL> sysnative , arches ) : <EOL> mock_get_system32_dir . return_value = "<STR_LIT>" <EOL> mock_get_syswow64_dir . return_value = "<STR_LIT>" <EOL> mock_get_sysnative_dir . return_value = "<STR_LIT>" <EOL> mock_is_64bit_arch . return_value = arches . startswith ( "<STR_LIT>" ) <EOL> mock_check_sysnative_dir_exists . return_value = ( arches == "<STR_LIT>" ) <EOL> expect_dict = { <EOL> "<STR_LIT>" : { <EOL> False : ( <EOL> "<STR_LIT>" , <EOL> [ mock_is_64bit_arch , mock_get_system32_dir ] <EOL> ) , <EOL> True : ( <EOL> "<STR_LIT>" , <EOL> [ mock_check_sysnative_dir_exists , <EOL> mock_get_system32_dir ] <EOL> ) <EOL> } , <EOL> "<STR_LIT>" : { <EOL> False : ( <EOL> "<STR_LIT>" , <EOL> [ mock_is_64bit_arch , mock_get_system32_dir ] <EOL> ) , <EOL> True : ( <EOL> "<STR_LIT>" , <EOL> [ mock_check_sysnative_dir_exists , <EOL> mock_get_sysnative_dir ] <EOL> ) <EOL> } , <EOL> "<STR_LIT>" : { <EOL> False : ( <EOL> "<STR_LIT>" , <EOL> [ mock_is_64bit_arch , mock_get_syswow64_dir ] <EOL> ) , <EOL> True : ( <EOL> "<STR_LIT>" , <EOL> [ mock_check_sysnative_dir_exists , <EOL> mock_get_system32_dir ] <EOL> ) <EOL> } <EOL> } <EOL> response = self . _winutils . _get_system_dir ( sysnative = sysnative ) <EOL> expect , calls = expect_dict [ arches ] [ sysnative ] <EOL> self . assertEqual ( expect , response ) <EOL> for call in calls : <EOL> call . assert_called_once_with ( ) <EOL> def test_get_system_dir_32on32 ( self ) : <EOL> arches = "<STR_LIT>" <EOL> self . _test_get_system_dir ( sysnative = False , arches = arches ) <EOL> self . _test_get_system_dir ( sysnative = True , arches = arches ) <EOL> def test_get_system_dir_32on64 ( self ) : <EOL> arches = "<STR_LIT>" <EOL> self . _test_get_system_dir ( sysnative = False , arches = arches ) <EOL> self . _test_get_system_dir ( sysnative = True , arches = arches ) <EOL> def test_get_system_dir_64on64 ( self ) : <EOL> arches = "<STR_LIT>" <EOL> self . _test_get_system_dir ( sysnative = False , arches = arches ) <EOL> self . _test_get_system_dir ( sysnative = True , arches = arches ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_is_nano_server ( self , mock_check_server_level ) : <EOL> expect = mock . Mock ( ) <EOL> mock_check_server_level . return_value = expect <EOL> response = self . _winutils . is_nano_server ( ) <EOL> mock_check_server_level . assert_called_once_with ( "<STR_LIT>" ) <EOL> self . assertEqual ( expect , response ) <EOL> def _test_check_server_level ( self , fail = False , success = True ) : <EOL> server_level = mock . Mock ( ) <EOL> open_key = self . _winreg_mock . OpenKey <EOL> if not success : <EOL> eclass = Exception <EOL> error = eclass ( ) <EOL> self . windows_utils . WindowsError = eclass <EOL> open_key . side_effect = [ error ] <EOL> if fail : <EOL> with self . assertRaises ( eclass ) : <EOL> self . _winutils . _check_server_level ( server_level ) <EOL> return <EOL> error . winerror = <NUM_LIT:2> <EOL> response = self . _winutils . _check_server_level ( server_level ) <EOL> self . assertEqual ( False , response ) <EOL> return <EOL> self . used_hive , self . used_path = None , None <EOL> mock_key = mock . Mock ( ) <EOL> @ contextlib . contextmanager <EOL> def open_key_context ( hive , path ) : <EOL> self . used_hive , self . used_path = hive , path <EOL> yield mock_key <EOL> self . _winreg_mock . OpenKey = open_key_context <EOL> self . _winreg_mock . QueryValueEx . return_value = [ <NUM_LIT:1> ] <EOL> response = self . _winutils . _check_server_level ( server_level ) <EOL> self . assertEqual ( self . _winreg_mock . HKEY_LOCAL_MACHINE , <EOL> self . used_hive ) <EOL> self . assertEqual ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> self . used_path ) <EOL> self . _winreg_mock . QueryValueEx . assert_called_once_with ( <EOL> mock_key , server_level ) <EOL> self . assertEqual ( True , response ) <EOL> def test_check_server_level_fail ( self ) : <EOL> self . _test_check_server_level ( fail = True , success = False ) <EOL> def test_check_server_level_no_success ( self ) : <EOL> self . _test_check_server_level ( fail = False , success = False ) <EOL> def test_check_server_level_success ( self ) : <EOL> self . _test_check_server_level ( ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_execute_powershell_script ( self , mock_execute_process , <EOL> mock_get_system32_dir , <EOL> mock_get_sysnative_dir , <EOL> mock_check_sysnative_dir_exists , <EOL> mock_is_nano_server , <EOL> ret_val = None , nano = False ) : <EOL> mock_check_sysnative_dir_exists . return_value = ret_val <EOL> mock_is_nano_server . return_value = nano <EOL> mock_get_sysnative_dir . return_value = '<STR_LIT>' <EOL> mock_get_system32_dir . return_value = '<STR_LIT>' <EOL> fake_path = os . path . join ( '<STR_LIT>' , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> args = [ fake_path ] <EOL> if not nano : <EOL> args . extend ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> args . append ( '<STR_LIT>' ) <EOL> response = self . _winutils . execute_powershell_script ( <EOL> script_path = '<STR_LIT>' ) <EOL> if ret_val : <EOL> mock_get_sysnative_dir . assert_called_once_with ( ) <EOL> else : <EOL> mock_get_system32_dir . assert_called_once_with ( ) <EOL> mock_execute_process . assert_called_with ( args , shell = False ) <EOL> self . assertEqual ( mock_execute_process . return_value , response ) <EOL> def test_execute_powershell_script_sysnative ( self ) : <EOL> self . _test_execute_powershell_script ( ret_val = True ) <EOL> def test_execute_powershell_script_system32 ( self ) : <EOL> self . _test_execute_powershell_script ( ret_val = False ) <EOL> def test_execute_powershell_script_sysnative_nano ( self ) : <EOL> self . _test_execute_powershell_script ( ret_val = True , nano = True ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_dhcp_hosts_in_use ( self , mock_get_adapter_addresses ) : <EOL> net_addr = { } <EOL> net_addr [ "<STR_LIT>" ] = '<STR_LIT>' <EOL> net_addr [ "<STR_LIT>" ] = '<STR_LIT>' <EOL> net_addr [ "<STR_LIT>" ] = True <EOL> mock_get_adapter_addresses . return_value = [ net_addr ] <EOL> response = self . _winutils . get_dhcp_hosts_in_use ( ) <EOL> mock_get_adapter_addresses . assert_called_once_with ( ) <EOL> self . assertEqual ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , response ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_set_ntp_client_config ( self , mock_execute_process , <EOL> mock_get_system32_dir , <EOL> mock_get_sysnative_dir , <EOL> mock_check_sysnative_dir_exists , <EOL> sysnative , ret_val ) : <EOL> fake_base_dir = '<STR_LIT>' <EOL> ntp_hosts = [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> mock_check_sysnative_dir_exists . return_value = sysnative <EOL> mock_execute_process . return_value = ( None , None , ret_val ) <EOL> w32tm_path = os . path . join ( fake_base_dir , "<STR_LIT>" ) <EOL> mock_get_sysnative_dir . return_value = fake_base_dir <EOL> mock_get_system32_dir . return_value = fake_base_dir <EOL> args = [ w32tm_path , '<STR_LIT>' , <EOL> '<STR_LIT>' % "<STR_LIT:U+002C>" . join ( ntp_hosts ) , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> if ret_val : <EOL> self . assertRaises ( exception . CloudbaseInitException , <EOL> self . _winutils . set_ntp_client_config , <EOL> '<STR_LIT>' ) <EOL> else : <EOL> self . _winutils . set_ntp_client_config ( ntp_hosts = ntp_hosts ) <EOL> if sysnative : <EOL> mock_get_sysnative_dir . assert_called_once_with ( ) <EOL> else : <EOL> mock_get_system32_dir . assert_called_once_with ( ) <EOL> mock_execute_process . assert_called_once_with ( args , shell = False ) <EOL> def test_set_ntp_client_config_sysnative_true ( self ) : <EOL> self . _test_set_ntp_client_config ( sysnative = True , ret_val = None ) <EOL> def test_set_ntp_client_config_sysnative_false ( self ) : <EOL> self . _test_set_ntp_client_config ( sysnative = False , ret_val = None ) <EOL> def test_set_ntp_client_config_sysnative_exception ( self ) : <EOL> self . _test_set_ntp_client_config ( sysnative = False , <EOL> ret_val = '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_set_network_adapter_mtu ( self , <EOL> mock_check_os_version , <EOL> mock_get_adapter_addresses , <EOL> mock_get_system_dir , <EOL> mock_execute_process , <EOL> fail = False , os_version_ret = True , <EOL> mac_address_match = True , <EOL> execute_process_val = <NUM_LIT:0> ) : <EOL> mac_address = "<STR_LIT>" <EOL> mtu = "<STR_LIT>" <EOL> base_dir = "<STR_LIT>" <EOL> mock_check_os_version . return_value = os_version_ret <EOL> mock_get_adapter_addresses . return_value = [ mock . MagicMock ( ) <EOL> for _ in range ( <NUM_LIT:3> ) ] <EOL> if mac_address_match : <EOL> mock_get_adapter_addresses . return_value [ <NUM_LIT:1> ] . __getitem__ . return_value = mac_address <EOL> mock_get_system_dir . return_value = base_dir <EOL> mock_execute_process . return_value = [ None , None , execute_process_val ] <EOL> if fail : <EOL> with self . assertRaises ( exception . CloudbaseInitException ) : <EOL> self . _winutils . set_network_adapter_mtu ( mac_address , mtu ) <EOL> return <EOL> with self . snatcher : <EOL> self . _winutils . set_network_adapter_mtu ( mac_address , mtu ) <EOL> expected_log = [ '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> { '<STR_LIT>' : mac_address , '<STR_LIT>' : mtu } ] <EOL> args = [ os . path . join ( base_dir , "<STR_LIT>" ) , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> mac_address , "<STR_LIT>" % mtu , "<STR_LIT>" ] <EOL> self . assertEqual ( expected_log , self . snatcher . output ) <EOL> mock_check_os_version . assert_called_once_with ( <NUM_LIT:6> , <NUM_LIT:0> ) <EOL> mock_get_adapter_addresses . assert_called_once_with ( ) <EOL> mock_get_system_dir . assert_called_once_with ( ) <EOL> mock_execute_process . assert_called_once_with ( args , shell = False ) <EOL> def test_set_network_adapter_mtu_not_supported ( self ) : <EOL> self . _test_set_network_adapter_mtu ( fail = True , os_version_ret = False ) <EOL> def test_set_network_adapter_mtu_no_mac_match ( self ) : <EOL> self . _test_set_network_adapter_mtu ( fail = True , <EOL> mac_address_match = False ) <EOL> def test_set_network_adapter_mtu_execute_fail ( self ) : <EOL> self . _test_set_network_adapter_mtu ( fail = True , <EOL> execute_process_val = <NUM_LIT:1> ) <EOL> def test_set_network_adapter_mtu ( self ) : <EOL> self . _test_set_network_adapter_mtu ( ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_execute_system32_process ( self , mock_execute_process , <EOL> mock_get_system_dir ) : <EOL> mock_get_system_dir . return_value = '<STR_LIT>' <EOL> mock_execute_process . return_value = mock . sentinel . execute_process <EOL> args = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> result = self . _winutils . execute_system32_process ( args ) <EOL> mock_execute_process . assert_called_once_with ( <EOL> [ os . path . join ( '<STR_LIT>' , args [ <NUM_LIT:0> ] ) ] + args [ <NUM_LIT:1> : ] , <EOL> decode_output = False , <EOL> shell = True ) <EOL> self . assertEqual ( mock . sentinel . execute_process , result ) <EOL> def test_get_password_maximum_length ( self ) : <EOL> self . assertEqual ( <NUM_LIT:20> , self . _winutils . get_maximum_password_length ( ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_set_timezone_fails ( self , mock_windows_tz ) : <EOL> mock_windows_tz . tz_win . get . return_value = None <EOL> with self . assertRaises ( exception . CloudbaseInitException ) as cm : <EOL> self . _winutils . set_timezone ( mock . sentinel . timezone ) <EOL> expected = ( <EOL> "<STR_LIT>" <EOL> % mock . sentinel . timezone <EOL> ) <EOL> self . assertEqual ( expected , str ( cm . exception ) ) <EOL> mock_windows_tz . tz_win . get . assert_called_once_with ( <EOL> mock . sentinel . timezone ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_set_timezone ( self , mock_windows_tz , mock_timezone ) : <EOL> mock_windows_tz . tz_win . get . return_value = ( <EOL> mock . sentinel . windows_timezone ) <EOL> self . _winutils . set_timezone ( mock . sentinel . timezone ) <EOL> mock_windows_tz . tz_win . get . assert_called_once_with ( <EOL> mock . sentinel . timezone ) <EOL> mock_timezone . Timezone . assert_called_once_with ( <EOL> mock . sentinel . windows_timezone ) <EOL> mock_timezone . Timezone . return_value . set . assert_called_once_with ( <EOL> self . _winutils ) <EOL> def _test__heap_alloc ( self , fail ) : <EOL> mock_heap = mock . Mock ( ) <EOL> mock_size = mock . Mock ( ) <EOL> if fail : <EOL> self . _kernel32 . HeapAlloc . return_value = None <EOL> with self . assertRaises ( exception . CloudbaseInitException ) as cm : <EOL> self . _winutils . _heap_alloc ( mock_heap , mock_size ) <EOL> self . assertEqual ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> str ( cm . exception ) ) <EOL> else : <EOL> result = self . _winutils . _heap_alloc ( mock_heap , mock_size ) <EOL> self . assertEqual ( self . _kernel32 . HeapAlloc . return_value , result ) <EOL> self . _kernel32 . HeapAlloc . assert_called_once_with ( <EOL> mock_heap , <NUM_LIT:0> , self . _ctypes_mock . c_size_t ( mock_size . value ) ) <EOL> def test__heap_alloc_error ( self ) : <EOL> self . _test__heap_alloc ( fail = True ) <EOL> def test__heap_alloc_no_error ( self ) : <EOL> self . _test__heap_alloc ( fail = False ) <EOL> def test__get_forward_table_no_memory ( self ) : <EOL> self . _winutils . _heap_alloc = mock . Mock ( ) <EOL> error_msg = '<STR_LIT>' <EOL> exc = exception . CloudbaseInitException ( error_msg ) <EOL> self . _winutils . _heap_alloc . side_effect = exc <EOL> with self . assertRaises ( exception . CloudbaseInitException ) as cm : <EOL> with self . _winutils . _get_forward_table ( ) : <EOL> pass <EOL> self . assertEqual ( error_msg , str ( cm . exception ) ) <EOL> self . _winutils . _heap_alloc . assert_called_once_with ( <EOL> self . _kernel32 . GetProcessHeap . return_value , <EOL> self . _ctypes_mock . wintypes . ULONG . return_value ) <EOL> def test__get_forward_table_insufficient_buffer_no_memory ( self ) : <EOL> self . _kernel32 . HeapAlloc . side_effect = ( mock . sentinel . table_mem , None ) <EOL> self . _iphlpapi . GetIpForwardTable . return_value = ( <EOL> self . _winutils . ERROR_INSUFFICIENT_BUFFER ) <EOL> with self . assertRaises ( exception . CloudbaseInitException ) : <EOL> with self . _winutils . _get_forward_table ( ) : <EOL> pass <EOL> table = self . _ctypes_mock . cast . return_value <EOL> self . _iphlpapi . GetIpForwardTable . assert_called_once_with ( <EOL> table , <EOL> self . _ctypes_mock . byref . return_value , <NUM_LIT:0> ) <EOL> heap_calls = [ <EOL> mock . call ( self . _kernel32 . GetProcessHeap . return_value , <NUM_LIT:0> , table ) , <EOL> mock . call ( self . _kernel32 . GetProcessHeap . return_value , <NUM_LIT:0> , table ) <EOL> ] <EOL> self . assertEqual ( heap_calls , self . _kernel32 . HeapFree . mock_calls ) <EOL> def _test__get_forward_table ( self , reallocation = False , <EOL> insufficient_buffer = False , <EOL> fail = False ) : <EOL> if fail : <EOL> with self . assertRaises ( exception . CloudbaseInitException ) as cm : <EOL> with self . _winutils . _get_forward_table ( ) : <EOL> pass <EOL> msg = ( '<STR_LIT>' <EOL> % mock . sentinel . error ) <EOL> self . assertEqual ( msg , str ( cm . exception ) ) <EOL> else : <EOL> with self . _winutils . _get_forward_table ( ) as table : <EOL> pass <EOL> pointer = self . _ctypes_mock . POINTER ( <EOL> self . _iphlpapi . Win32_MIB_IPFORWARDTABLE ) <EOL> expected_forward_table = self . _ctypes_mock . cast ( <EOL> self . _kernel32 . HeapAlloc . return_value , pointer ) <EOL> self . assertEqual ( expected_forward_table , table ) <EOL> heap_calls = [ <EOL> mock . call ( self . _kernel32 . GetProcessHeap . return_value , <NUM_LIT:0> , <EOL> self . _ctypes_mock . cast . return_value ) <EOL> ] <EOL> forward_calls = [ <EOL> mock . call ( self . _ctypes_mock . cast . return_value , <EOL> self . _ctypes_mock . byref . return_value , <NUM_LIT:0> ) , <EOL> ] <EOL> if insufficient_buffer : <EOL> forward_calls . append ( forward_calls [ <NUM_LIT:0> ] ) <EOL> if reallocation : <EOL> heap_calls . append ( heap_calls [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( heap_calls , self . _kernel32 . HeapFree . mock_calls ) <EOL> self . assertEqual ( forward_calls , <EOL> self . _iphlpapi . GetIpForwardTable . mock_calls ) <EOL> def test__get_forward_table_sufficient_buffer ( self ) : <EOL> self . _iphlpapi . GetIpForwardTable . return_value = None <EOL> self . _test__get_forward_table ( ) <EOL> def test__get_forward_table_insufficient_buffer_reallocate ( self ) : <EOL> self . _kernel32 . HeapAlloc . side_effect = ( <EOL> mock . sentinel . table_mem , mock . sentinel . table_mem ) <EOL> self . _iphlpapi . GetIpForwardTable . side_effect = ( <EOL> self . _winutils . ERROR_INSUFFICIENT_BUFFER , None ) <EOL> self . _test__get_forward_table ( reallocation = True , <EOL> insufficient_buffer = True ) <EOL> def test__get_forward_table_insufficient_buffer_other_error ( self ) : <EOL> self . _kernel32 . HeapAlloc . side_effect = ( <EOL> mock . sentinel . table_mem , mock . sentinel . table_mem ) <EOL> self . _iphlpapi . GetIpForwardTable . side_effect = ( <EOL> self . _winutils . ERROR_INSUFFICIENT_BUFFER , mock . sentinel . error ) <EOL> self . _test__get_forward_table ( reallocation = True , <EOL> insufficient_buffer = True , <EOL> fail = True ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_routes ( self , mock_forward_table ) : <EOL> def _same ( arg ) : <EOL> return arg . _mock_name . encode ( ) <EOL> route = mock . MagicMock ( ) <EOL> mock_cast_result = mock . Mock ( ) <EOL> mock_cast_result . contents = [ route ] <EOL> self . _ctypes_mock . cast . return_value = mock_cast_result <EOL> self . windows_utils . Ws2_32 . inet_ntoa . side_effect = _same <EOL> route . dwForwardIfIndex = '<STR_LIT>' <EOL> route . dwForwardProto = '<STR_LIT>' <EOL> route . dwForwardMetric1 = '<STR_LIT>' <EOL> routes = self . _winutils . _get_ipv4_routing_table ( ) <EOL> mock_forward_table . assert_called_once_with ( ) <EOL> enter = mock_forward_table . return_value . __enter__ <EOL> enter . assert_called_once_with ( ) <EOL> exit_ = mock_forward_table . return_value . __exit__ <EOL> exit_ . assert_called_once_with ( None , None , None ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( routes ) ) <EOL> given_route = routes [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , given_route [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , given_route [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , given_route [ <NUM_LIT:2> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , given_route [ <NUM_LIT:3> ] ) <EOL> self . assertEqual ( '<STR_LIT>' , given_route [ <NUM_LIT:4> ] ) </s>
