Revert tests to previous state

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)

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"

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