diff --git a/tests/test_ankaios.py b/tests/test_ankaios.py index 165a1a5..e7502b3 100644 --- a/tests/test_ankaios.py +++ b/tests/test_ankaios.py @@ -215,12 +215,12 @@ def test_get_reponse_by_id(): assert ankaios._get_response_by_id("3344") == response assert ankaios._responses.empty() + ankaios._responses.put(Response(MESSAGE_BUFFER_COMPLETE_STATE)) with pytest.raises(TimeoutError): - ankaios._responses.put(Response(MESSAGE_BUFFER_COMPLETE_STATE)) ankaios._get_response_by_id("3344", timeout=0.01) + ankaios._responses.put(Response(MESSAGE_BUFFER_CONNECTION_CLOSED)) with pytest.raises(ConnectionClosedException): - ankaios._responses.put(Response(MESSAGE_BUFFER_CONNECTION_CLOSED)) ankaios._get_response_by_id("1122") diff --git a/tests/test_control_interface.py b/tests/test_control_interface.py index c91d2a5..566ef8f 100644 --- a/tests/test_control_interface.py +++ b/tests/test_control_interface.py @@ -395,30 +395,28 @@ def test_handle_response_connection_closed(): # Got connection closed response as initial response response_callback = MagicMock() + ci = ControlInterface( + add_response_callback=response_callback, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.INITIALIZED with pytest.raises( ConnectionClosedException, match="Connection closed reason" ): - ci = ControlInterface( - add_response_callback=response_callback, - add_log_callback=lambda _: None, - add_event_callback=lambda _: None, - ) - ci._state = ControlInterfaceState.INITIALIZED - ci._handle_response(conn_closed_response) # Got connection closed response while already connected response_callback = MagicMock() + ci = ControlInterface( + add_response_callback=response_callback, + add_log_callback=lambda _: None, + add_event_callback=lambda _: None, + ) + ci._state = ControlInterfaceState.CONNECTED with pytest.raises( ConnectionClosedException, match="Connection closed reason" ): - ci = ControlInterface( - add_response_callback=response_callback, - add_log_callback=lambda _: None, - add_event_callback=lambda _: None, - ) - ci._state = ControlInterfaceState.CONNECTED - ci._handle_response(conn_closed_response) @@ -540,16 +538,17 @@ def test_write_request(): add_log_callback=lambda _: None, add_event_callback=lambda _: None, ) + test_request = generate_test_request() ci._state = ControlInterfaceState.TERMINATED with pytest.raises( ControlInterfaceException, match="Could not write to pipe" ): - ci.write_request(generate_test_request()) + ci.write_request(test_request) ci._state = ControlInterfaceState.CONNECTED with patch("ankaios_sdk.ControlInterface._write_to_pipe") as mock_write: - ci.write_request(generate_test_request()) + ci.write_request(test_request) mock_write.assert_called_once() ci._state = ControlInterfaceState.CONNECTION_CLOSED @@ -557,7 +556,7 @@ def test_write_request(): ConnectionClosedException, match="Could not write to pipe, connection closed.", ): - ci.write_request(generate_test_request()) + ci.write_request(test_request) def test_send_initial_hello(): diff --git a/tests/workload/test_access_right_rule.py b/tests/workload/test_access_right_rule.py index db8454f..b9543a0 100644 --- a/tests/workload/test_access_right_rule.py +++ b/tests/workload/test_access_right_rule.py @@ -16,7 +16,8 @@ This module contains unit tests for the AccessRightRule class. """ -from ankaios_sdk import AccessRightRule +import pytest +from ankaios_sdk import AccessRightRule, WorkloadFieldException from ankaios_sdk._protos import _ank_base @@ -42,6 +43,9 @@ def test_state_rule(): "filterMask": ["mask_1", "mask_2"], } + with pytest.raises(WorkloadFieldException): + _ = AccessRightRule.state_rule("Invalid", ["mask"]) + def test_log_rule(): """ diff --git a/tests/workload/test_workload.py b/tests/workload/test_workload.py index a97bb61..e0db6b3 100644 --- a/tests/workload/test_workload.py +++ b/tests/workload/test_workload.py @@ -239,16 +239,6 @@ def test_rules(workload: Workload): # pylint: disable=redefined-outer-name assert len(allow_rules) == 1 assert len(deny_rules) == 1 - with pytest.raises(WorkloadFieldException): - workload.update_allow_rules( - [AccessRightRule.state_rule("Invalid", ["mask"])] - ) - - with pytest.raises(WorkloadFieldException): - workload.update_deny_rules( - [AccessRightRule.state_rule("Invalid", ["mask"])] - ) - allow_rules.append( AccessRightRule.state_rule( "Write", [f"{WORKLOADS_PREFIX}.another_workload"] diff --git a/tests/workload_state/test_workload_execution_state.py b/tests/workload_state/test_workload_execution_state.py index 4aba382..bae388d 100644 --- a/tests/workload_state/test_workload_execution_state.py +++ b/tests/workload_state/test_workload_execution_state.py @@ -51,7 +51,6 @@ def test_interpret_state_error(): Test the handling of an invalid execution state in the WorkloadExecutionState class, ensuring it raises a ValueError. """ + invalid_state = _ank_base.ExecutionState(additionalInfo="No state present") with pytest.raises(ValueError, match="Invalid state for workload."): - WorkloadExecutionState( - _ank_base.ExecutionState(additionalInfo="No state present") - ) + WorkloadExecutionState(invalid_state)