Loading csp-lmc-mid/tests/unit/midcspsubarray_unit_test.py +60 −1 Original line number Original line Diff line number Diff line Loading @@ -236,7 +236,7 @@ def test_removereceptors_command_WHEN_subarray_is_in_wrong_state(): tango_context.device.RemoveAllReceptors() tango_context.device.RemoveAllReceptors() if df: if df: err_msg = str(df.value.args[0].desc) err_msg = str(df.value.args[0].desc) assert "Error executing command RemoveAllReceptorsCommand" in err_msg assert "Error executing command RemoveAllReceptorsCommand" in err_msg def test_midcspsubarray_obsstate_WHEN_cbfsubarray_raises_an_exception(): def test_midcspsubarray_obsstate_WHEN_cbfsubarray_raises_an_exception(): """ """ Loading Loading @@ -665,6 +665,65 @@ def test_midcspsubarray_obsstate_WHEN_cbfsubarray_is_in_fault_during_abort_reque prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) Poller(3, 0.1).check(prober_obs_state) assert tango_context.device.obsState == ObsState.FAULT assert tango_context.device.obsState == ObsState.FAULT def test_midcspsubarray_obsstate_WHEN_abort_invoked_in_resetting(): """ Test the behavior of the MidCspSubarray when the command Abort is invoked when the subarray is in RESETTING Expected result: state = ON obsState=ABORTED 28-07-2020: Obtained: the command raises an exception. Modified the test to catch the exception. """ device_under_test = MidCspSubarrayBase cbf_subarray_fqdn = 'mid_csp_cbf/sub_elt/subarray_01' pss_subarray_fqdn = 'mid_csp_pss/sub_elt/subarray_01' cbf_subarray_state_attr = 'obsState' dut_properties = { 'CbfSubarray': cbf_subarray_fqdn, 'PssSubarray': pss_subarray_fqdn, 'CspMaster':'mid_csp/elt/master', 'SubID': '1', } event_subscription_map = {} cbf_subarray_device_proxy_mock = Mock() pss_subarray_device_proxy_mock = Mock() cbf_subarray_device_proxy_mock.subscribe_event.side_effect = ( lambda attr_name, event_type, callback, *args, **kwargs: event_subscription_map.update({attr_name: callback})) proxies_to_mock = { cbf_subarray_fqdn: cbf_subarray_device_proxy_mock, pss_subarray_fqdn: cbf_subarray_device_proxy_mock, } receptor_list = [1,2,3] with fake_tango_system(device_under_test, initial_dut_properties=dut_properties, proxies_to_mock=proxies_to_mock) as tango_context: #cbf_subarray_device_proxy_mock.Abort.side_effect = return_failed with mock.patch('csp_lmc_mid.MidCspSubarrayBase.MidCspSubarrayBase.AddReceptorsCommand.do') as mock_do,\ mock.patch('csp_lmc_mid.MidCspSubarrayBase.MidCspSubarrayBase.OnCommand.do') as mock_on_do,\ mock.patch('csp_lmc_common.CspSubarray.CspSubarray._get_expected_delay') as mock_delay_expected,\ mock.patch('csp_lmc_mid.MidCspSubarrayBase.MidCspSubarrayBase.__len__') as mock_len: mock_len.return_value = len(receptor_list) mock_do.return_value = (ResultCode.OK, "AddReceptors OK") mock_on_do.return_value = (ResultCode.OK, "On command OK") mock_delay_expected.return_value = 2 tango_context.device.On() tango_context.device.AddReceptors(receptor_list) configuration_string = load_json_file("test_ConfigureScan_ADR4.json") tango_context.device.Configure(configuration_string) dummy_event = create_dummy_event(cbf_subarray_fqdn, 'obsstate',ObsState.FAULT) event_subscription_map[cbf_subarray_state_attr](dummy_event) prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) tango_context.device.ObsReset() prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.RESETTING, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) with pytest.raises(tango.DevFailed) as df: tango_context.device.Abort() if df: err_msg = str(df.value.args[0].desc) assert "Error executing command AbortCommand" in err_msg prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) ''' ''' def test_midcspsubarray_obsstate_WHEN_cbfsubarray_abort_returns_failed(): def test_midcspsubarray_obsstate_WHEN_cbfsubarray_abort_returns_failed(): """ """ Loading Loading
csp-lmc-mid/tests/unit/midcspsubarray_unit_test.py +60 −1 Original line number Original line Diff line number Diff line Loading @@ -236,7 +236,7 @@ def test_removereceptors_command_WHEN_subarray_is_in_wrong_state(): tango_context.device.RemoveAllReceptors() tango_context.device.RemoveAllReceptors() if df: if df: err_msg = str(df.value.args[0].desc) err_msg = str(df.value.args[0].desc) assert "Error executing command RemoveAllReceptorsCommand" in err_msg assert "Error executing command RemoveAllReceptorsCommand" in err_msg def test_midcspsubarray_obsstate_WHEN_cbfsubarray_raises_an_exception(): def test_midcspsubarray_obsstate_WHEN_cbfsubarray_raises_an_exception(): """ """ Loading Loading @@ -665,6 +665,65 @@ def test_midcspsubarray_obsstate_WHEN_cbfsubarray_is_in_fault_during_abort_reque prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) Poller(3, 0.1).check(prober_obs_state) assert tango_context.device.obsState == ObsState.FAULT assert tango_context.device.obsState == ObsState.FAULT def test_midcspsubarray_obsstate_WHEN_abort_invoked_in_resetting(): """ Test the behavior of the MidCspSubarray when the command Abort is invoked when the subarray is in RESETTING Expected result: state = ON obsState=ABORTED 28-07-2020: Obtained: the command raises an exception. Modified the test to catch the exception. """ device_under_test = MidCspSubarrayBase cbf_subarray_fqdn = 'mid_csp_cbf/sub_elt/subarray_01' pss_subarray_fqdn = 'mid_csp_pss/sub_elt/subarray_01' cbf_subarray_state_attr = 'obsState' dut_properties = { 'CbfSubarray': cbf_subarray_fqdn, 'PssSubarray': pss_subarray_fqdn, 'CspMaster':'mid_csp/elt/master', 'SubID': '1', } event_subscription_map = {} cbf_subarray_device_proxy_mock = Mock() pss_subarray_device_proxy_mock = Mock() cbf_subarray_device_proxy_mock.subscribe_event.side_effect = ( lambda attr_name, event_type, callback, *args, **kwargs: event_subscription_map.update({attr_name: callback})) proxies_to_mock = { cbf_subarray_fqdn: cbf_subarray_device_proxy_mock, pss_subarray_fqdn: cbf_subarray_device_proxy_mock, } receptor_list = [1,2,3] with fake_tango_system(device_under_test, initial_dut_properties=dut_properties, proxies_to_mock=proxies_to_mock) as tango_context: #cbf_subarray_device_proxy_mock.Abort.side_effect = return_failed with mock.patch('csp_lmc_mid.MidCspSubarrayBase.MidCspSubarrayBase.AddReceptorsCommand.do') as mock_do,\ mock.patch('csp_lmc_mid.MidCspSubarrayBase.MidCspSubarrayBase.OnCommand.do') as mock_on_do,\ mock.patch('csp_lmc_common.CspSubarray.CspSubarray._get_expected_delay') as mock_delay_expected,\ mock.patch('csp_lmc_mid.MidCspSubarrayBase.MidCspSubarrayBase.__len__') as mock_len: mock_len.return_value = len(receptor_list) mock_do.return_value = (ResultCode.OK, "AddReceptors OK") mock_on_do.return_value = (ResultCode.OK, "On command OK") mock_delay_expected.return_value = 2 tango_context.device.On() tango_context.device.AddReceptors(receptor_list) configuration_string = load_json_file("test_ConfigureScan_ADR4.json") tango_context.device.Configure(configuration_string) dummy_event = create_dummy_event(cbf_subarray_fqdn, 'obsstate',ObsState.FAULT) event_subscription_map[cbf_subarray_state_attr](dummy_event) prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) tango_context.device.ObsReset() prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.RESETTING, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) with pytest.raises(tango.DevFailed) as df: tango_context.device.Abort() if df: err_msg = str(df.value.args[0].desc) assert "Error executing command AbortCommand" in err_msg prober_obs_state = Probe(tango_context.device, 'obsState', ObsState.FAULT, f"Wrong CspSubarray state") Poller(3, 0.1).check(prober_obs_state) ''' ''' def test_midcspsubarray_obsstate_WHEN_cbfsubarray_abort_returns_failed(): def test_midcspsubarray_obsstate_WHEN_cbfsubarray_abort_returns_failed(): """ """ Loading