Create a loop to make requests (#492)

* Create a loop to make requests

* Update src/ephys_link/back_end/server.py

Co-authored-by: Copilot <[email protected]>

* Update src/ephys_link/back_end/server.py

Co-authored-by: Copilot <[email protected]>

* Hatch static analysis

* Update tests to mock new_event_loop instead of deprecated get_event_loop (#493)

* Initial plan

* Update tests to mock new_event_loop instead of get_event_loop

Co-authored-by: kjy5 <[email protected]>

---------

Co-authored-by: copilot-swe-agent[bot] <[email protected]>
Co-authored-by: Kenneth Yang <[email protected]>

---------

Co-authored-by: Copilot <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

src/ephys_link/back_end/server.py

@@ -12,7 +12,7 @@
     ```
 """
 
-from asyncio import get_event_loop, run
+from asyncio import new_event_loop, run
 from collections.abc import Callable, Coroutine
 from json import JSONDecodeError, dumps, loads
 from typing import Any, TypeVar, final
@@ -96,10 +96,16 @@ def launch(self) -> None:
 
         # List platform and available manipulators.
         self._console.info_print("PLATFORM", self._platform_handler.get_display_name())
-        self._console.info_print(
-            "MANIPULATORS",
-            str(get_event_loop().run_until_complete(self._platform_handler.get_manipulators()).manipulators),
-        )
+
+        # Create a temporary event loop for getting manipulators
+        loop = new_event_loop()
+        try:
+            self._console.info_print(
+                "MANIPULATORS",
+                str(loop.run_until_complete(self._platform_handler.get_manipulators()).manipulators),
+            )
+        finally:
+            loop.close()
 
         # Launch server
         if self._options.use_proxy:

tests/back_end/test_server.py

@@ -88,8 +88,8 @@ def test_launch_server(
         patched_run_until_complete = mocker.patch.object(
             asyncio_loop, "run_until_complete", return_value=GetManipulatorsResponse(manipulators=DUMMY_STRING_LIST)
         )
-        patched_get_event_loop = mocker.patch.object(
-            ephys_link.back_end.server, "get_event_loop", return_value=asyncio_loop
+        patched_new_event_loop = mocker.patch.object(
+            ephys_link.back_end.server, "new_event_loop", return_value=asyncio_loop
         )
 
         # Mock out run_app.
@@ -102,7 +102,7 @@ def test_launch_server(
         patched_get_display_name.assert_called_once()
         patched_get_manipulators.assert_called_once()
         patched_run_until_complete.assert_called_once()
-        patched_get_event_loop.assert_called_once()
+        patched_new_event_loop.assert_called_once()
         spied_info_print.assert_any_call("PLATFORM", platform_handler.get_display_name())
         spied_info_print.assert_any_call("MANIPULATORS", str(DUMMY_STRING_LIST))
         mocked_run_app.assert_called_once()
@@ -123,8 +123,8 @@ def test_launch_proxy_client(
         patched_run_until_complete = mocker.patch.object(
             asyncio_loop, "run_until_complete", return_value=GetManipulatorsResponse(manipulators=DUMMY_STRING_LIST)
         )
-        patched_get_event_loop = mocker.patch.object(
-            ephys_link.back_end.server, "get_event_loop", return_value=asyncio_loop
+        patched_new_event_loop = mocker.patch.object(
+            ephys_link.back_end.server, "new_event_loop", return_value=asyncio_loop
         )
 
         # Mock out run.
@@ -143,7 +143,7 @@ def run_coroutine(coroutine: Awaitable[None]) -> None:
         patched_get_display_name.assert_called_once()
         patched_get_manipulators.assert_called_once()
         patched_run_until_complete.assert_called_once()
-        patched_get_event_loop.assert_called_once()
+        patched_new_event_loop.assert_called_once()
         spied_info_print.assert_any_call("PLATFORM", platform_handler.get_display_name())
         spied_info_print.assert_any_call("MANIPULATORS", str(DUMMY_STRING_LIST))
         spied_info_print.assert_any_call("PINPOINT ID", mocker.ANY)
@@ -166,8 +166,8 @@ def test_launch_proxy_client_failed_init(
         patched_run_until_complete = mocker.patch.object(
             asyncio_loop, "run_until_complete", return_value=GetManipulatorsResponse(manipulators=DUMMY_STRING_LIST)
         )
-        patched_get_event_loop = mocker.patch.object(
-            ephys_link.back_end.server, "get_event_loop", return_value=asyncio_loop
+        patched_new_event_loop = mocker.patch.object(
+            ephys_link.back_end.server, "new_event_loop", return_value=asyncio_loop
         )
 
         # Mock out run.
@@ -191,7 +191,7 @@ def run_coroutine(coroutine: Awaitable[None]) -> None:
         patched_get_display_name.assert_called_once()
         patched_get_manipulators.assert_called_once()
         patched_run_until_complete.assert_called_once()
-        patched_get_event_loop.assert_called_once()
+        patched_new_event_loop.assert_called_once()
         spied_info_print.assert_any_call("PLATFORM", platform_handler.get_display_name())
         spied_info_print.assert_any_call("MANIPULATORS", str(DUMMY_STRING_LIST))
         spied_info_print.assert_any_call("PINPOINT ID", mocker.ANY)