<s> import eventlet <EOL> import mock <EOL> from os_win import constants <EOL> from os_win import exceptions as os_win_exc <EOL> from os_win import utilsfactory <EOL> from nova . tests . unit . virt . hyperv import test_base <EOL> from nova import utils <EOL> from nova . virt . hyperv import eventhandler <EOL> class EventHandlerTestCase ( test_base . HyperVBaseTestCase ) : <EOL> _FAKE_POLLING_INTERVAL = <NUM_LIT:3> <EOL> _FAKE_EVENT_CHECK_TIMEFRAME = <NUM_LIT:15> <EOL> @ mock . patch . object ( utilsfactory , '<STR_LIT>' ) <EOL> def setUp ( self , mock_get_vmutils ) : <EOL> super ( EventHandlerTestCase , self ) . setUp ( ) <EOL> self . _state_change_callback = mock . Mock ( ) <EOL> self . _running_state_callback = mock . Mock ( ) <EOL> self . flags ( <EOL> power_state_check_timeframe = self . _FAKE_EVENT_CHECK_TIMEFRAME , <EOL> group = '<STR_LIT>' ) <EOL> self . flags ( <EOL> power_state_event_polling_interval = self . _FAKE_POLLING_INTERVAL , <EOL> group = '<STR_LIT>' ) <EOL> self . _event_handler = eventhandler . InstanceEventHandler ( <EOL> self . _state_change_callback , <EOL> self . _running_state_callback ) <EOL> @ mock . patch . object ( eventhandler , '<STR_LIT>' , create = True ) <EOL> @ mock . patch . object ( eventhandler . InstanceEventHandler , '<STR_LIT>' ) <EOL> @ mock . patch . object ( eventlet , '<STR_LIT>' ) <EOL> def _test_poll_events ( self , mock_sleep , mock_dispatch , <EOL> mock_wmi , event_found = True ) : <EOL> fake_listener = mock . Mock ( ) <EOL> mock_wmi . x_wmi_timed_out = Exception <EOL> fake_listener . side_effect = ( mock . sentinel . event if event_found <EOL> else mock_wmi . x_wmi_timed_out , <EOL> KeyboardInterrupt ) <EOL> self . _event_handler . _listener = fake_listener <EOL> self . assertRaises ( KeyboardInterrupt , <EOL> self . _event_handler . _poll_events ) <EOL> if event_found : <EOL> mock_dispatch . assert_called_once_with ( mock . sentinel . event ) <EOL> else : <EOL> mock_sleep . assert_called_once_with ( self . _FAKE_POLLING_INTERVAL ) <EOL> def test_poll_having_events ( self ) : <EOL> self . _test_poll_events ( ) <EOL> def test_poll_no_event_found ( self ) : <EOL> self . _test_poll_events ( event_found = False ) <EOL> @ mock . patch . object ( eventhandler . InstanceEventHandler , <EOL> '<STR_LIT>' ) <EOL> @ mock . patch . object ( eventhandler . InstanceEventHandler , '<STR_LIT>' ) <EOL> def _test_dispatch_event ( self , mock_emit_event , mock_get_uuid , <EOL> missing_uuid = False ) : <EOL> mock_get_uuid . return_value = ( <EOL> mock . sentinel . instance_uuid if not missing_uuid else None ) <EOL> self . _event_handler . _vmutils . get_vm_power_state . return_value = ( <EOL> mock . sentinel . power_state ) <EOL> event = mock . Mock ( ) <EOL> event . ElementName = mock . sentinel . instance_name <EOL> event . EnabledState = mock . sentinel . enabled_state <EOL> self . _event_handler . _dispatch_event ( event ) <EOL> if not missing_uuid : <EOL> mock_emit_event . assert_called_once_with ( <EOL> mock . sentinel . instance_name , <EOL> mock . sentinel . instance_uuid , <EOL> mock . sentinel . power_state ) <EOL> else : <EOL> self . assertFalse ( mock_emit_event . called ) <EOL> def test_dispatch_event_new_final_state ( self ) : <EOL> self . _test_dispatch_event ( ) <EOL> def test_dispatch_event_missing_uuid ( self ) : <EOL> self . _test_dispatch_event ( missing_uuid = True ) <EOL> @ mock . patch . object ( eventhandler . InstanceEventHandler , '<STR_LIT>' ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' ) <EOL> def test_emit_event ( self , mock_spawn , mock_get_event ) : <EOL> self . _event_handler . _emit_event ( mock . sentinel . instance_name , <EOL> mock . sentinel . instance_uuid , <EOL> constants . HYPERV_VM_STATE_ENABLED ) <EOL> virt_event = mock_get_event . return_value <EOL> mock_spawn . assert_has_calls ( <EOL> [ mock . call ( self . _state_change_callback , virt_event ) , <EOL> mock . call ( self . _running_state_callback , <EOL> mock . sentinel . instance_name , <EOL> mock . sentinel . instance_uuid ) ] ) <EOL> def _test_get_instance_uuid ( self , instance_found = True , <EOL> missing_uuid = False ) : <EOL> if instance_found : <EOL> side_effect = ( mock . sentinel . instance_uuid <EOL> if not missing_uuid else None , ) <EOL> else : <EOL> side_effect = os_win_exc . HyperVVMNotFoundException ( <EOL> vm_name = mock . sentinel . instance_name ) <EOL> mock_get_uuid = self . _event_handler . _vmutils . get_instance_uuid <EOL> mock_get_uuid . side_effect = side_effect <EOL> instance_uuid = self . _event_handler . _get_instance_uuid ( <EOL> mock . sentinel . instance_name ) <EOL> expected_uuid = ( mock . sentinel . instance_uuid <EOL> if instance_found and not missing_uuid else None ) <EOL> self . assertEqual ( expected_uuid , instance_uuid ) <EOL> def test_get_nova_created_instance_uuid ( self ) : <EOL> self . _test_get_instance_uuid ( ) <EOL> def test_get_deleted_instance_uuid ( self ) : <EOL> self . _test_get_instance_uuid ( instance_found = False ) <EOL> def test_get_instance_uuid_missing_notes ( self ) : <EOL> self . _test_get_instance_uuid ( missing_uuid = True ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_virt_event ( self , mock_lifecycle_event ) : <EOL> instance_state = constants . HYPERV_VM_STATE_ENABLED <EOL> expected_transition = self . _event_handler . _TRANSITION_MAP [ <EOL> instance_state ] <EOL> virt_event = self . _event_handler . _get_virt_event ( <EOL> mock . sentinel . instance_uuid , instance_state ) <EOL> self . assertEqual ( mock_lifecycle_event . return_value , <EOL> virt_event ) <EOL> mock_lifecycle_event . assert_called_once_with ( <EOL> uuid = mock . sentinel . instance_uuid , <EOL> transition = expected_transition ) </s>
