From bd3bc1f6999126b4dbfdbb0733712cbd59357770 Mon Sep 17 00:00:00 2001 From: xjules Date: Mon, 20 Jan 2025 13:29:19 +0100 Subject: [PATCH] Revert tests to previous state --- .../test_ensemble_client.py | 2 +- .../test_event_reporter.py | 37 +++++++------------ tests/ert/utils.py | 7 +++- 3 files changed, 20 insertions(+), 26 deletions(-) diff --git a/tests/ert/unit_tests/ensemble_evaluator/test_ensemble_client.py b/tests/ert/unit_tests/ensemble_evaluator/test_ensemble_client.py index ca9a8ef362e..b62a37a1f22 100644 --- a/tests/ert/unit_tests/ensemble_evaluator/test_ensemble_client.py +++ b/tests/ert/unit_tests/ensemble_evaluator/test_ensemble_client.py @@ -57,7 +57,7 @@ async def test_reconnect_when_missing_ping(unused_tcp_port, monkeypatch): monkeypatch.setattr(_ert.forward_model_runner.client, "PING_TIMEOUT", 0.1) async with ( - MockZMQServer(unused_tcp_port) as mock_server, + MockZMQServer(unused_tcp_port, signal=3) as mock_server, Client(url) as client, ): await client.send("start", retries=1) diff --git a/tests/ert/unit_tests/forward_model_runner/test_event_reporter.py b/tests/ert/unit_tests/forward_model_runner/test_event_reporter.py index 99a23804e3f..441c1dad181 100644 --- a/tests/ert/unit_tests/forward_model_runner/test_event_reporter.py +++ b/tests/ert/unit_tests/forward_model_runner/test_event_reporter.py @@ -43,8 +43,8 @@ def test_report_with_successful_start_message_argument(unused_tcp_port): reporter.report(Start(fmstep1)) reporter.report(Finish()) - assert len(mock_server.messages) == 3 - event = event_from_json(mock_server.messages[1]) + assert len(mock_server.messages) == 1 + event = event_from_json(mock_server.messages[0]) assert type(event) is ForwardModelStepStart assert event.ensemble == "ens_id" assert event.real == "0" @@ -70,8 +70,8 @@ def test_report_with_failed_start_message_argument(unused_tcp_port): reporter.report(msg) reporter.report(Finish()) - assert len(mock_server.messages) == 4 # 2 msgs + connect, disconnect - event = event_from_json(mock_server.messages[2]) + assert len(mock_server.messages) == 2 + event = event_from_json(mock_server.messages[1]) assert type(event) is ForwardModelStepFailure assert event.error_msg == "massive_failure" @@ -89,8 +89,8 @@ async def test_report_with_successful_exit_message_argument(unused_tcp_port): reporter.report(Exited(fmstep1, 0)) reporter.report(Finish().with_error("failed")) - assert len(mock_server.messages) == 3 - event = event_from_json(mock_server.messages[1]) + assert len(mock_server.messages) == 1 + event = event_from_json(mock_server.messages[0]) assert type(event) is ForwardModelStepSuccess @@ -107,10 +107,8 @@ def test_report_with_failed_exit_message_argument(unused_tcp_port): reporter.report(Exited(fmstep1, 1).with_error("massive_failure")) reporter.report(Finish()) - assert len(mock_server.messages) == 3 - assert mock_server.messages[0] == "CONNECT" - assert mock_server.messages[2] == "DISCONNECT" - event = event_from_json(mock_server.messages[1]) + assert len(mock_server.messages) == 1 + event = event_from_json(mock_server.messages[0]) assert type(event) is ForwardModelStepFailure assert event.error_msg == "massive_failure" @@ -128,11 +126,8 @@ def test_report_with_running_message_argument(unused_tcp_port): reporter.report(Running(fmstep1, ProcessTreeStatus(max_rss=100, rss=10))) reporter.report(Finish()) - assert len(mock_server.messages) == 3 - - assert mock_server.messages[0] == "CONNECT" - assert mock_server.messages[2] == "DISCONNECT" - event = event_from_json(mock_server.messages[1]) + assert len(mock_server.messages) == 1 + event = event_from_json(mock_server.messages[0]) assert type(event) is ForwardModelStepRunning assert event.max_memory_usage == 100 assert event.current_memory_usage == 10 @@ -151,7 +146,7 @@ def test_report_only_job_running_for_successful_run(unused_tcp_port): reporter.report(Running(fmstep1, ProcessTreeStatus(max_rss=100, rss=10))) reporter.report(Finish()) - assert len(mock_server.messages) == 3 + assert len(mock_server.messages) == 1 def test_report_with_failed_finish_message_argument(unused_tcp_port): @@ -167,7 +162,7 @@ def test_report_with_failed_finish_message_argument(unused_tcp_port): reporter.report(Running(fmstep1, ProcessTreeStatus(max_rss=100, rss=10))) reporter.report(Finish().with_error("massive_failure")) - assert len(mock_server.messages) == 3 + assert len(mock_server.messages) == 1 def test_report_inconsistent_events(unused_tcp_port): @@ -217,9 +212,7 @@ def test_report_with_failed_reporter_but_finished_jobs(unused_tcp_port): def test_report_with_reconnected_reporter_but_finished_jobs(unused_tcp_port): # this is to show when the reporter fails but reconnects # reporter still manages to send events and completes fine - # see assert reporter._timeout_timestamp is not None - # meaning Finish event initiated _timeout but timeout wasn't reached since - # it finished succesfully + # see reporter._event_publisher for more details. host = "localhost" url = f"tcp://{host}:{unused_tcp_port}" @@ -239,6 +232,4 @@ def test_report_with_reconnected_reporter_but_finished_jobs(unused_tcp_port): if reporter._event_publisher_thread.is_alive(): reporter._event_publisher_thread.join() assert reporter._done.is_set() - assert len(mock_server.messages) == 5, ( - "expected 3 Job running messages and CONNECT and DISCONNECT" - ) + assert len(mock_server.messages) == 3, "expected 3 Job running messages" diff --git a/tests/ert/utils.py b/tests/ert/utils.py index d62756933d5..e80ff3957e4 100644 --- a/tests/ert/utils.py +++ b/tests/ert/utils.py @@ -69,9 +69,10 @@ def wait_until(func, interval=0.5, timeout=30): class MockZMQServer: def __init__(self, port, signal=0): """Mock ZMQ server for testing - signal = 0: normal operation + signal = 0: normal operation, receive messages but don't store CONNECT and DISCONNECT messages signal = 1: don't send ACK and don't receive messages signal = 2: don't send ACK, but receive messages + signal = 3: normal operation, and store also CONNECT and DISCONNECT messages """ self.port = port self.messages = [] @@ -138,7 +139,9 @@ async def _handler(self): self.dealers.discard(dealer) elif self.value == 0: await self.router_socket.send_multipart([dealer, b"", ACK_MSG]) - if self.value != 1: + if ( + self.value in {0, 2} and frame not in {CONNECT_MSG, DISCONNECT_MSG} + ) or self.value == 3: self.messages.append(frame.decode("utf-8")) except asyncio.CancelledError: break