Connect and disconnect

Author: kennethjyang

src/ephys_link/back_end/server.py

@@ -33,7 +33,13 @@
 from ephys_link.__about__ import __version__
 from ephys_link.back_end.platform_handler import PlatformHandler
 from ephys_link.front_end.console import Console
-from ephys_link.utils.constants import PORT, PROXY_CLIENT_NOT_INITIALIZED_ERROR, SERVER_NOT_INITIALIZED_ERROR
+from ephys_link.utils.constants import (
+    PORT,
+    PROXY_CLIENT_NOT_INITIALIZED_ERROR,
+    SERVER_NOT_INITIALIZED_ERROR,
+    cannot_connect_as_client_is_already_connected_error,
+    client_disconnected_without_being_connected_error,
+)
 
 # Server message generic types.
 INPUT_TYPE = TypeVar("INPUT_TYPE", bound=VBLBaseModel)
@@ -197,7 +203,7 @@ async def connect(self, sid: str, _: str) -> bool:
             return True
 
         self._console.error_print(
-            "CONNECTION REFUSED", f"Cannot connect {sid} as {self._client_sid} is already connected."
+            "CONNECTION REFUSED", cannot_connect_as_client_is_already_connected_error(sid, self._client_sid)
         )
         return False
 
@@ -207,13 +213,14 @@ async def disconnect(self, sid: str) -> None:
         Args:
             sid: Socket session ID.
         """
-        self._console.info_print("DISCONNECTED", sid)
+        self._console.info_print("DISCONNECTION REQUEST", sid)
 
         # Reset client SID if it matches.
         if self._client_sid == sid:
             self._client_sid = ""
+            self._console.info_print("DISCONNECTED", sid)
         else:
-            self._console.error_print("DISCONNECTION", f"Client {sid} disconnected without being connected.")
+            self._console.error_print("DISCONNECTION", client_disconnected_without_being_connected_error(sid))
 
     async def platform_event_handler(self, event: str, *args: tuple[Any]) -> str:  # pyright: ignore [reportExplicitAny]
         """Handle events from the server.

src/ephys_link/utils/constants.py

@@ -61,3 +61,24 @@ def did_not_reach_target_depth_error(request: SetDepthRequest, final_unified_dep
 
 SERVER_NOT_INITIALIZED_ERROR = "Server not initialized."
 PROXY_CLIENT_NOT_INITIALIZED_ERROR = "Proxy client not initialized."
+
+
+def cannot_connect_as_client_is_already_connected_error(new_client_sid: str, current_client_sid: str) -> str:
+    """Generate an error message for when the client is already connected.
+    Args:
+        new_client_sid: The SID of the new client.
+        current_client_sid: The SID of the current client.
+    Returns:
+        str: The error message.
+    """
+    return f"Cannot connect {new_client_sid} as {current_client_sid} is already connected."
+
+
+def client_disconnected_without_being_connected_error(client_sid: str) -> str:
+    """Generate an error message for when the client is disconnected without being connected.
+    Args:
+        client_sid: The SID of the client.
+    Returns:
+        str: The error message.
+    """
+    return f"Client {client_sid} disconnected without being connected."

tests/back_end/test_server.py

@@ -10,7 +10,12 @@
 from ephys_link.back_end.platform_handler import PlatformHandler
 from ephys_link.back_end.server import Server
 from ephys_link.front_end.console import Console
-from ephys_link.utils.constants import PROXY_CLIENT_NOT_INITIALIZED_ERROR, SERVER_NOT_INITIALIZED_ERROR
+from ephys_link.utils.constants import (
+    PROXY_CLIENT_NOT_INITIALIZED_ERROR,
+    SERVER_NOT_INITIALIZED_ERROR,
+    cannot_connect_as_client_is_already_connected_error,
+    client_disconnected_without_being_connected_error,
+)
 from tests.conftest import DUMMY_STRING, DUMMY_STRING_LIST
 
 
@@ -176,3 +181,71 @@ def run_coroutine(coroutine: Awaitable[None]) -> None:
         assert init_error.value.args[0] == PROXY_CLIENT_NOT_INITIALIZED_ERROR
         patched_connect.assert_not_awaited()  # pyright: ignore[reportUnusedCallResult]
         patched_wait.assert_not_awaited()  # pyright: ignore[reportUnusedCallResult]
+
+    @pytest.mark.asyncio
+    async def test_connect_success(self, server: Server, console: Console, mocker: MockerFixture) -> None:
+        """Server should allow connection if there is no existing connection."""
+        # Spy console.
+        spied_info_print = mocker.spy(console, "info_print")
+
+        # Act.
+        result = await server.connect(DUMMY_STRING, DUMMY_STRING)
+
+        # Assert.
+        spied_info_print.assert_any_call("CONNECTION REQUEST", DUMMY_STRING)
+        assert server._client_sid == DUMMY_STRING  # noqa: SLF001 # pyright: ignore[reportPrivateUsage]
+        spied_info_print.assert_any_call("CONNECTION GRANTED", DUMMY_STRING)
+        assert result
+
+    @pytest.mark.asyncio
+    async def test_connect_failure(self, server: Server, console: Console, mocker: MockerFixture) -> None:
+        """Server should not allow connection if there is an existing connection."""
+        # Spy console.
+        spied_info_print = mocker.spy(console, "info_print")
+        spied_error_print = mocker.spy(console, "error_print")
+
+        # Mock client sid.
+        _ = mocker.patch.object(server, "_client_sid", new=DUMMY_STRING)
+
+        # Act.
+        result = await server.connect(DUMMY_STRING, DUMMY_STRING)
+
+        # Assert.
+        spied_info_print.assert_any_call("CONNECTION REQUEST", DUMMY_STRING)
+        spied_error_print.assert_called_once_with(
+            "CONNECTION REFUSED", cannot_connect_as_client_is_already_connected_error(DUMMY_STRING, DUMMY_STRING)
+        )
+        assert not result
+
+    @pytest.mark.asyncio
+    async def test_disconnect_success(self, server: Server, console: Console, mocker: MockerFixture) -> None:
+        """Server should allow disconnection if the SID matches the existing connection."""
+        # Spy console.
+        spied_info_print = mocker.spy(console, "info_print")
+
+        # Mock client sid.
+        _ = mocker.patch.object(server, "_client_sid", new=DUMMY_STRING)
+
+        # Act.
+        await server.disconnect(DUMMY_STRING)
+
+        # Assert.
+        spied_info_print.assert_any_call("DISCONNECTION REQUEST", DUMMY_STRING)
+        assert server._client_sid == ""  # noqa: SLF001 # pyright: ignore[reportPrivateUsage]
+        spied_info_print.assert_any_call("DISCONNECTED", DUMMY_STRING)
+
+    @pytest.mark.asyncio
+    async def test_disconnect_failure(self, server: Server, console: Console, mocker: MockerFixture) -> None:
+        """Server should not allow disconnection if there is no existing connection."""
+        # Spy console.
+        spied_info_print = mocker.spy(console, "info_print")
+        spied_error_print = mocker.spy(console, "error_print")
+
+        # Act.
+        await server.disconnect(DUMMY_STRING)
+
+        # Assert.
+        spied_info_print.assert_any_call("DISCONNECTION REQUEST", DUMMY_STRING)
+        spied_error_print.assert_called_once_with(
+            "DISCONNECTION", client_disconnected_without_being_connected_error(DUMMY_STRING)
+        )