Add test for connection aliveness and remove duplicate from test_connection_manager

Author: vrslev

packages/stompman/test_stompman/test_aliveness.py

@@ -0,0 +1,21 @@
+import time
+
+import pytest
+from stompman import Client, ConnectedFrame, ReceiptFrame
+
+from test_stompman.conftest import EnrichedClient, build_dataclass, create_spying_connection
+
+pytestmark = [pytest.mark.anyio, pytest.mark.usefixtures("mock_sleep")]
+
+
+@pytest.mark.parametrize("is_alive", [True, False])
+async def test_connection_alive(is_alive: bool) -> None:  # noqa: FBT001
+    connection_class, _ = create_spying_connection(
+        [ConnectedFrame(headers={"version": Client.PROTOCOL_VERSION, "heart-beat": "1000,1000"})],
+        [],
+        [build_dataclass(ReceiptFrame)],
+    )
+    client = await EnrichedClient(connection_class=connection_class).__aenter__()
+    assert client._connection_manager._active_connection_state
+    client._connection_manager._active_connection_state.connection.last_read_time = time.time() - 1 + is_alive
+    assert client.is_alive() == is_alive

packages/stompman/test_stompman/test_connection_manager.py

@@ -355,16 +355,3 @@ class MockConnection(BaseMockConnection):
 async def test_maybe_write_frame_ok() -> None:
     async with EnrichedConnectionManager(connection_class=BaseMockConnection) as manager:
         assert await manager.maybe_write_frame(build_dataclass(ConnectFrame))
-
-
-@pytest.mark.parametrize("is_alive", [True, False])
-async def test_connection_alive(is_alive: bool) -> None:  # noqa: FBT001
-    connection_class, _ = create_spying_connection(
-        [ConnectedFrame(headers={"version": Client.PROTOCOL_VERSION, "heart-beat": "1000,1000"})],
-        [],
-        [build_dataclass(ReceiptFrame)],
-    )
-    connection_manager = await EnrichedConnectionManager(connection_class=connection_class).__aenter__()
-    assert connection_manager._active_connection_state
-    connection_manager._active_connection_state.connection.last_read_time = time.time() - 1 + is_alive
-    assert connection_manager._active_connection_state.is_alive() == is_alive