Coverage for test\test_HUD_main.py: 99%
345 statements
« prev ^ index » next coverage.py v7.6.3, created at 2024-10-15 19:33 +0000
1import pytest
2from unittest.mock import MagicMock, patch
3from PyQt5.QtWidgets import QApplication
4import sys
5import types
6import zmq
8from pathlib import Path
10sys.path.append(str(Path(__file__).parent.parent))
12# Create a mock 'WinTables' module
13win_tables_module = types.ModuleType("WinTables")
14win_tables_module.Table = MagicMock()
16with patch.dict("sys.modules", {"WinTables": win_tables_module}):
17 import HUD_main
20@pytest.fixture
21def app(qtbot):
22 return QApplication.instance()
25@pytest.fixture
26def hud_main(app, qtbot):
27 # Crate mock
28 options = MagicMock()
29 options.dbname = "test_db"
30 options.config = None
31 options.errorsToConsole = False
32 options.xloc = None
33 options.yloc = None
35 import tempfile
37 with (
38 patch("HUD_main.Configuration.Config") as mock_config,
39 patch("HUD_main.Configuration.set_logfile"),
40 patch("HUD_main.Database.Database"),
41 patch("HUD_main.ZMQReceiver"),
42 patch("sys.exit"),
43 patch("PyQt5.QtCore.QCoreApplication.quit"),
44 ):
45 mock_config_instance = MagicMock()
46 mock_config.return_value = mock_config_instance
48 mock_config_instance.dir_log = tempfile.gettempdir()
49 mock_config_instance.os_family = "Win7"
50 mock_config_instance.get_hud_ui_parameters.return_value = {
51 "deck_type": "default",
52 "card_back": "blue",
53 "card_wd": 72,
54 "card_ht": 96,
55 "hud_days": 30,
56 "h_hud_days": 90,
57 }
58 mock_config_instance.graphics_path = tempfile.gettempdir()
59 mock_config_instance.hhcs = {"test_site": MagicMock(converter="some_converter")}
60 mock_config_instance.get_site_parameters.return_value = {
61 "layout_set": "some_layout",
62 "param1": "value1",
63 }
64 mock_config_instance.get_layout.return_value = "some_layout"
66 hm = HUD_main.HUD_main(options, db_name=options.dbname)
67 qtbot.addWidget(hm.main_window)
68 yield hm
70 qtbot.waitExposed(hm.main_window)
71 hm.main_window.close()
74# Verifies that all necessary attributes of the HUD_main instance are correctly initialized.
75def test_hud_main_initialization(hud_main):
76 assert hud_main.db_name == "test_db"
77 assert hasattr(hud_main, "config")
78 assert hasattr(hud_main, "db_connection")
79 assert hasattr(hud_main, "hud_dict")
80 assert hasattr(hud_main, "blacklist")
81 assert hasattr(hud_main, "hud_params")
82 assert hasattr(hud_main, "deck")
83 assert hasattr(hud_main, "cache")
84 assert hasattr(hud_main, "zmq_receiver")
85 assert hasattr(hud_main, "zmq_worker")
86 assert hasattr(hud_main, "main_window")
89# Ensures that the handle_message method correctly calls read_stdin when provided with a hand ID.
90def test_handle_message(hud_main):
91 with patch.object(hud_main, "read_stdin") as mock_read_stdin:
92 hud_main.handle_message("test_hand_id")
93 mock_read_stdin.assert_called_once_with("test_hand_id")
96# Checks that the destroy method properly closes connections and stops processes.
97def test_destroy(hud_main):
98 with (
99 patch.object(hud_main.zmq_receiver, "close") as mock_close,
100 patch.object(hud_main.zmq_worker, "stop") as mock_stop,
101 patch("PyQt5.QtCore.QCoreApplication.quit") as mock_quit,
102 ):
103 hud_main.destroy()
104 mock_close.assert_called_once()
105 mock_stop.assert_called_once()
106 mock_quit.assert_called_once()
109# Verifies that check_tables calls the correct methods (client_destroyed, client_moved, client_resized) based on the table's status.
110@pytest.mark.parametrize(
111 "status, expected_method",
112 [
113 ("client_destroyed", "client_destroyed"),
114 ("client_moved", "client_moved"),
115 ("client_resized", "client_resized"),
116 ],
117)
118def test_check_tables(hud_main, status, expected_method):
119 mock_hud = MagicMock()
120 mock_hud.table.check_table.return_value = status
121 hud_main.hud_dict = {"test_table": mock_hud}
123 with patch.object(hud_main, expected_method) as mock_method:
124 hud_main.check_tables()
125 mock_method.assert_called_once_with(None, mock_hud)
128# Ensures that create_HUD creates a new HUD and adds it to the hud_dict.
129def test_create_hud(hud_main):
130 with (
131 patch.object(HUD_main.Hud, "Hud") as mock_hud,
132 patch.object(hud_main, "idle_create") as mock_idle_create,
133 patch.object(
134 hud_main.config, "get_site_parameters", return_value={"layout_set": "some_layout", "param1": "value1"}
135 ),
136 patch.object(hud_main.config, "get_layout", return_value="some_layout"),
137 ):
138 mock_table = MagicMock()
139 mock_table.site = "test_site"
140 hud_main.create_HUD("new_hand_id", mock_table, "temp_key", 9, "poker_game", "cash", {}, {})
142 assert "temp_key" in hud_main.hud_dict
143 mock_hud.assert_called_once()
144 mock_idle_create.assert_called_once()
147# Verifies that update_HUD properly calls idle_update.
148def test_update_hud(hud_main):
149 with patch.object(hud_main, "idle_update") as mock_idle_update:
150 hud_main.update_HUD("new_hand_id", "table_name", hud_main.config)
151 mock_idle_update.assert_called_once_with("new_hand_id", "table_name", hud_main.config)
154# Ensures that cached data is processed correctly in read_stdin and calls update_HUD.
155def test_read_stdin_cached(hud_main):
156 # Configuration env
157 hud_main.config = MagicMock()
158 hud_main.config.get_supported_sites.return_value = ["test_site"]
159 hud_main.config.supported_sites = {"test_site": MagicMock(screen_name="test_hero")}
160 test_hand_id = "test_hand_id"
161 hud_main.cache[test_hand_id] = (
162 "table_name",
163 9,
164 "poker_game",
165 "cash",
166 False,
167 1,
168 "test_site",
169 9,
170 "tour_number",
171 "tab_number",
172 )
173 temp_key = "table_name"
174 hud_main.hud_dict[temp_key] = MagicMock()
175 hud_main.hud_dict[temp_key].hud_params = {"hud_days": 30, "h_hud_days": 90}
176 hud_main.hud_dict[temp_key].poker_game = "poker_game"
177 hud_main.hud_dict[temp_key].max = 9
178 hud_main.hud_dict[temp_key].aux_windows = []
180 with (
181 patch.object(hud_main.db_connection, "get_site_id", return_value=[(1,)]),
182 patch.object(hud_main.db_connection, "get_player_id", return_value=123),
183 patch.object(hud_main.db_connection, "init_hud_stat_vars"),
184 patch.object(
185 hud_main.db_connection, "get_stats_from_hand", return_value={"player1": {"screen_name": "test_hero"}}
186 ),
187 patch.object(hud_main, "get_cards", return_value={}),
188 patch.object(hud_main, "update_HUD") as mock_update_hud,
189 ):
190 hud_main.read_stdin(test_hand_id)
191 assert mock_update_hud.called, "update_HUD n'a pas été appelé"
194# Confirms that cached data is used if available when calling read_stdin.
195def test_read_stdin_cache_used(hud_main):
196 hud_main.cache = {"test_hand_id": ("table_name", 9, "poker_game", "cash", False, 1, "test_site", 9, 123, "tab")}
197 with patch("HUD_main.log.debug") as mock_log_debug:
198 hud_main.read_stdin("test_hand_id")
199 mock_log_debug.assert_any_call("Using cached data for hand test_hand_id")
202# Tests the behavior of read_stdin when no cached data is available, ensuring it calls create_HUD
203def test_read_stdin_not_cached(hud_main):
204 hud_main.config = MagicMock()
205 hud_main.config.get_supported_sites.return_value = ["test_site"]
206 hud_main.config.supported_sites = {"test_site": MagicMock(screen_name="test_hero")}
207 hud_main.config.get_site_parameters.return_value = {"aux_enabled": True}
209 hud_main.Tables = MagicMock()
210 hud_main.Tables.Table.return_value = MagicMock()
211 test_hand_id = "test_hand_id"
213 hud_main.cache = {}
215 table_info = ("table_name", 9, "poker_game", "tour", False, 1, "test_site", 9, 123456, "Table 789")
217 with (
218 patch.object(hud_main.db_connection, "get_site_id", return_value=[(1,)]),
219 patch.object(hud_main.db_connection, "get_player_id", return_value=123),
220 patch.object(hud_main.db_connection, "get_table_info", return_value=table_info),
221 patch.object(hud_main.db_connection, "init_hud_stat_vars"),
222 patch.object(
223 hud_main.db_connection, "get_stats_from_hand", return_value={"player1": {"screen_name": "test_hero"}}
224 ),
225 patch.object(hud_main, "get_cards", return_value={}),
226 patch.object(hud_main.Tables, "Table", return_value=MagicMock()),
227 patch.object(hud_main, "create_HUD") as mock_create_hud,
228 ):
229 hud_main.read_stdin(test_hand_id)
230 assert mock_create_hud.called, "create_HUD n'a pas été appelé"
233# Ensures that get_cards retrieves both player and community cards correctly.
234def test_get_cards(hud_main):
235 mock_db = MagicMock()
236 mock_db.get_cards.return_value = {"player1": ["As", "Kh"]}
237 mock_db.get_common_cards.return_value = {"common": ["Jd", "Qc", "Tc"]}
238 hud_main.db_connection = mock_db
240 cards = hud_main.get_cards("test_hand_id", "holdem")
241 assert "player1" in cards
242 assert "common" in cards
245# Verifies that idle_kill removes the HUD from hud_dict and cleans up widgets.
246def test_idle_kill(hud_main):
247 mock_hud = MagicMock()
248 hud_main.hud_dict["test_table"] = mock_hud
249 hud_main.vb = MagicMock()
251 hud_main.idle_kill("test_table")
253 assert "test_table" not in hud_main.hud_dict
254 mock_hud.kill.assert_called_once()
255 hud_main.vb.removeWidget.assert_called_once()
258# Checks exception handling in read_stdin when an error occurs in get_table_info.
259def test_read_stdin_exception_handling(hud_main):
260 hud_main.config = MagicMock()
261 hud_main.config.get_supported_sites.return_value = ["test_site"]
262 hud_main.config.get_site_parameters.return_value = {"aux_enabled": True}
263 hud_main.hero = {}
264 hud_main.hero_ids = {}
266 hud_main.cache = {}
268 test_hand_id = "test_hand_id"
270 with (
271 patch.object(hud_main.db_connection, "get_table_info", side_effect=Exception("Database error")),
272 patch.object(hud_main, "destroy") as mock_destroy,
273 ):
274 hud_main.read_stdin(test_hand_id)
276 mock_destroy.assert_not_called()
279# Ensures that ZMQWorker.stop stops the worker properly.
280def test_zmqworker_stop():
281 zmq_receiver = MagicMock()
282 worker = HUD_main.ZMQWorker(zmq_receiver)
283 worker.wait = MagicMock()
284 worker.is_running = True
286 worker.stop()
287 assert not worker.is_running
288 worker.wait.assert_called_once()
291# Verifies that a heartbeat log is generated when no messages are received.
292def test_process_message_heartbeat(hud_main):
293 zmq_receiver = HUD_main.ZMQReceiver()
294 zmq_receiver.socket = MagicMock()
295 zmq_receiver.poller = MagicMock()
297 zmq_receiver.poller.poll.return_value = {}
299 with patch("HUD_main.log.debug") as mock_log_debug:
300 zmq_receiver.process_message()
301 mock_log_debug.assert_called_with("Heartbeat: No message received")
304# Tests the run loop of ZMQWorker, ensuring it stops after processing a message.
305def test_zmqworker_run(hud_main):
306 zmq_receiver = MagicMock()
307 worker = HUD_main.ZMQWorker(zmq_receiver)
309 # Limit the loop to avoid an infinite blockage
310 worker.is_running = True
312 # Use of `side_effect` to stop the loop after the first call to `process_message`.
313 def stop_after_one_iteration(*args, **kwargs):
314 worker.is_running = False
316 with (
317 patch("time.sleep", return_value=None),
318 patch.object(zmq_receiver, "process_message", side_effect=stop_after_one_iteration) as mock_process_message,
319 ):
320 worker.run()
321 mock_process_message.assert_called_once()
324# Ensures that process_message logs the correct hand ID received from the socket.
325def test_process_message(hud_main):
326 zmq_receiver = HUD_main.ZMQReceiver()
327 zmq_receiver.socket = MagicMock()
328 zmq_receiver.poller = MagicMock()
330 zmq_receiver.poller.poll.return_value = {zmq_receiver.socket: zmq.POLLIN}
331 zmq_receiver.socket.recv_string.return_value = "hand_id"
333 with patch("HUD_main.log.debug") as mock_log_debug:
334 zmq_receiver.process_message()
335 mock_log_debug.assert_called_with("Received hand ID: hand_id")
338# Verifies that table_title_changed calls kill_hud when the table's title changes.
339def test_table_title_changed_calls_kill_hud(hud_main):
340 mock_hud = MagicMock()
341 mock_hud.table.key = "test_table"
342 hud_main.hud_dict["test_table"] = mock_hud
344 with patch.object(hud_main, "kill_hud") as mock_kill_hud:
345 hud_main.table_title_changed(None, mock_hud)
346 mock_kill_hud.assert_called_once_with(None, "test_table")
349# Ensures that table_is_stale calls kill_hud for stale tables.
350def test_table_is_stale_calls_kill_hud(hud_main):
351 mock_hud = MagicMock()
352 mock_hud.table.key = "test_table"
353 hud_main.hud_dict["test_table"] = mock_hud
355 with patch.object(hud_main, "kill_hud") as mock_kill_hud:
356 hud_main.table_is_stale(mock_hud)
357 mock_kill_hud.assert_called_once_with(None, "test_table")
360# Verifies that blacklist_hud correctly removes a HUD from hud_dict and adds it to the blacklist.
361def test_blacklist_hud(hud_main):
362 mock_hud = MagicMock()
363 mock_hud.tablenumber = 123
364 hud_main.hud_dict["test_table"] = mock_hud
365 hud_main.vb = MagicMock()
367 hud_main.blacklist_hud(None, "test_table")
369 assert 123 in hud_main.blacklist
370 assert "test_table" not in hud_main.hud_dict
371 mock_hud.kill.assert_called_once()
372 hud_main.vb.removeWidget.assert_called_once()
375# Ensures that handle_worker_error logs an error message.
376def test_handle_worker_error(hud_main):
377 with patch("HUD_main.log.error") as mock_log_error:
378 hud_main.handle_worker_error("Test error message")
379 mock_log_error.assert_called_once_with("ZMQWorker encountered an error: Test error message")
382# Verifies that close_event_handler calls destroy and accepts the event.
383def test_close_event_handler(hud_main):
384 mock_event = MagicMock()
385 with patch.object(hud_main, "destroy") as mock_destroy:
386 hud_main.close_event_handler(mock_event)
387 mock_destroy.assert_called_once()
388 mock_event.accept.assert_called_once()
391# Ensures that idle_move moves the table and auxiliary windows.
392def test_idle_move(hud_main):
393 mock_hud = MagicMock()
394 mock_hud.aux_windows = [MagicMock()]
395 hud_main.idle_move(mock_hud)
397 mock_hud.move_table_position.assert_called_once()
398 for aw in mock_hud.aux_windows:
399 aw.move_windows.assert_called_once()
402# Verifies that idle_resize resizes the table and auxiliary windows.
403def test_idle_resize(hud_main):
404 mock_hud = MagicMock()
405 mock_hud.aux_windows = [MagicMock()]
406 hud_main.idle_resize(mock_hud)
408 mock_hud.resize_windows.assert_called_once()
409 for aw in mock_hud.aux_windows:
410 aw.resize_windows.assert_called_once()
413# Checks that kill_hud removes the HUD from hud_dict and cleans up associated widgets.
414def test_kill_hud(hud_main):
415 mock_hud = MagicMock()
416 hud_main.hud_dict["test_table"] = mock_hud
417 hud_main.vb = MagicMock()
419 hud_main.kill_hud(None, "test_table")
421 assert "test_table" not in hud_main.hud_dict
422 mock_hud.kill.assert_called_once()
423 hud_main.vb.removeWidget.assert_called_once()
426# Verifies that client_moved calls idle_move for the given HUD.
427def test_client_moved(hud_main):
428 mock_hud = MagicMock()
429 with patch.object(hud_main, "idle_move") as mock_idle_move:
430 hud_main.client_moved(None, mock_hud)
431 mock_idle_move.assert_called_once_with(mock_hud)
434# Ensures that client_resized calls idle_resize for the given HUD.
435def test_client_resized(hud_main):
436 mock_hud = MagicMock()
437 with patch.object(hud_main, "idle_resize") as mock_idle_resize:
438 hud_main.client_resized(None, mock_hud)
439 mock_idle_resize.assert_called_once_with(mock_hud)
442# Checks that client_destroyed calls kill_hud for the appropriate HUD.
443def test_client_destroyed(hud_main):
444 mock_hud = MagicMock()
445 mock_hud.table.key = "test_table"
446 with patch.object(hud_main, "kill_hud") as mock_kill_hud:
447 hud_main.client_destroyed(None, mock_hud)
448 mock_kill_hud.assert_called_once_with(None, "test_table")
451# Verifies that idle_create creates a new HUD and adds it to hud_dict, along with logging.
452@pytest.mark.parametrize("import_path", ["HUD_main.QLabel", "PyQt5.QtWidgets.QLabel"])
453def test_idle_create(import_path, hud_main):
454 with patch(import_path) as mock_qlabel, patch("HUD_main.log") as mock_log:
455 # Configuration
456 mock_hud = MagicMock()
457 mock_hud.tablehudlabel = MagicMock()
458 hud_main.hud_dict = {"test_table": mock_hud}
459 hud_main.vb = MagicMock()
461 new_hand_id = "new_hand_id"
462 table = MagicMock()
463 table.site = "test_site"
464 table.number = 123
465 temp_key = "test_table"
466 max = 9
467 poker_game = "holdem"
468 hud_type = "cash"
469 stat_dict = {}
470 cards = {}
472 with (
473 patch.object(hud_main, "get_cards", return_value=cards),
474 patch.object(hud_main.hud_dict["test_table"], "create") as mock_create,
475 patch.object(hud_main.hud_dict["test_table"], "aux_windows", []),
476 ):
477 print("Before calling idle_create")
479 # Call idle_create
480 hud_main.idle_create(new_hand_id, table, temp_key, max, poker_game, hud_type, stat_dict, cards)
482 print("After calling idle_create")
483 print(f"mock_qlabel.call_count: {mock_qlabel.call_count}")
484 print(f"mock_qlabel.call_args_list: {mock_qlabel.call_args_list}")
486 # Checks
487 tablehudlabel = hud_main.hud_dict[temp_key].tablehudlabel
488 print(f"tablehudlabel: {tablehudlabel}")
489 print(f"Type of tablehudlabel: {tablehudlabel.__class__}")
490 print(f"Is instance of mocked QLabel: {isinstance(tablehudlabel, mock_qlabel.return_value.__class__)}")
492 # Check taht vb.addWidget is called
493 try:
494 hud_main.vb.addWidget.assert_called_once()
495 print("vb.addWidget assertion passed")
496 except AssertionError as e:
497 print(f"vb.addWidget assertion failed: {e}")
499 # Check attributs
500 assert hud_main.hud_dict[temp_key].tablehudlabel is not None, "tablehudlabel is None"
501 assert hud_main.hud_dict[temp_key].tablenumber == table.number, "tablenumber mismatch"
502 print("hud_dict assertions passed")
504 # Check call
505 try:
506 mock_create.assert_called_once_with(new_hand_id, hud_main.config, stat_dict)
507 print("create method assertion passed")
508 except AssertionError as e:
509 print(f"create method assertion failed: {e}")
511 # Check logs
512 expected_log_message = f"adding label {table.site} - {temp_key}"
513 print(f"Expected log message: {expected_log_message}")
514 print(f"Actual log calls: {mock_log.debug.call_args_list}")
516 log_message_found = any(call[0][0] == expected_log_message for call in mock_log.debug.call_args_list)
517 if log_message_found:
518 print("Log assertion passed")
519 else:
520 print("Log assertion failed: Expected message not found in debug logs")
523# Ensures that idle_update updates the HUD and auxiliary windows.
524def test_idle_update(hud_main):
525 # Create a mock HUD
526 temp_key = "table_name"
527 mock_hud = MagicMock()
528 hud_main.hud_dict[temp_key] = mock_hud
529 hud_main.hud_dict[temp_key].aux_windows = [MagicMock()]
531 # Call idle_update
532 hud_main.idle_update("new_hand_id", temp_key, hud_main.config)
534 # Assert update method was called
535 mock_hud.update.assert_called_once_with("new_hand_id", hud_main.config)
537 # Assert aux_windows update_gui was called
538 for aw in hud_main.hud_dict[temp_key].aux_windows:
539 aw.update_gui.assert_called_once_with("new_hand_id")
542# Confirms that idle_kill removes widgets from the layout and calls the HUD's kill method.
543def test_idle_kill_widget_removal(hud_main):
544 mock_hud = MagicMock()
545 hud_main.hud_dict["test_table"] = mock_hud
546 hud_main.vb = MagicMock()
548 # Call idle_kill
549 hud_main.idle_kill("test_table")
551 # Assert widget was removed from layout
552 hud_main.vb.removeWidget.assert_called_once_with(mock_hud.tablehudlabel)
554 # Assert kill method on HUD was called
555 mock_hud.kill.assert_called_once()
557 # Assert HUD is removed from the dictionary
558 assert "test_table" not in hud_main.hud_dict
561# Ensures that check_tables calls the correct methods for different table statuses.
562@pytest.mark.parametrize("status", ["client_destroyed", "client_moved", "client_resized"])
563def test_check_tables_full_coverage(hud_main, status):
564 mock_hud = MagicMock()
565 mock_hud.table.check_table.return_value = status
566 hud_main.hud_dict = {"test_table": mock_hud}
568 # Map status to expected method
569 method_map = {
570 "client_destroyed": "client_destroyed",
571 "client_moved": "client_moved",
572 "client_resized": "client_resized",
573 }
575 with patch.object(hud_main, method_map[status]) as mock_method:
576 hud_main.check_tables()
577 mock_method.assert_called_once_with(None, mock_hud)
580# Verifies that the appropriate idle methods (idle_move, idle_resize, kill_hud) are called for different client actions.
581@pytest.mark.parametrize(
582 "method_name, expected_args",
583 [
584 ("client_moved", (MagicMock(),)),
585 ("client_resized", (MagicMock(),)),
586 ("client_destroyed", (None, MagicMock().table.key)),
587 ],
588)
589def test_client_methods(hud_main, method_name, expected_args):
590 mock_hud = MagicMock()
592 # Map method to expected idle method
593 idle_method = {"client_moved": "idle_move", "client_resized": "idle_resize", "client_destroyed": "kill_hud"}[
594 method_name
595 ]
597 with patch.object(hud_main, idle_method) as mock_idle_method:
598 getattr(hud_main, method_name)(None, mock_hud)
599 if method_name == "client_destroyed":
600 mock_idle_method.assert_called_once_with(expected_args[0], mock_hud.table.key)
601 else:
602 mock_idle_method.assert_called_once_with(mock_hud)
605# Ensures that auxiliary windows are created and updated properly.
606def test_aux_windows_creation_and_update(hud_main):
607 mock_aux_window = MagicMock()
608 hud_main.hud_dict["test_key"] = MagicMock()
609 hud_main.hud_dict["test_key"].aux_windows = [mock_aux_window]
611 with patch("HUD_main.log.debug") as mock_log_debug:
612 hud_main.idle_create("new_hand_id", MagicMock(), "test_key", 9, "poker_game", "cash", {}, {})
614 mock_aux_window.create.assert_called_once()
615 mock_aux_window.update_gui.assert_called_once_with("new_hand_id")
616 mock_log_debug.assert_called_with("idle_create new_hand_id new_hand_id")
619# Verifies that ZMQReceiver.close properly closes the socket and context, and logs the closure.
620def test_zmqreceiver_close(hud_main):
621 zmq_receiver = HUD_main.ZMQReceiver(port="5555")
623 with (
624 patch.object(zmq_receiver.socket, "close") as mock_socket_close,
625 patch.object(zmq_receiver.context, "term") as mock_context_term,
626 patch("HUD_main.log.info") as mock_log_info,
627 ):
628 zmq_receiver.close()
630 # Ensure socket.close and context.term were called
631 mock_socket_close.assert_called_once()
632 mock_context_term.assert_called_once()
634 # Ensure the closure was logged
635 mock_log_info.assert_called_with("ZMQ receiver closed")