454 add unit tests

.github/workflows/test.yml

@@ -0,0 +1,34 @@
+name: Test
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: 🛎 Checkout
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.head_ref }}
+
+      - name: 🔭 Install UV
+        uses: astral-sh/setup-uv@v6
+        with:
+          enable-cache: true
+          cache-dependency-glob: "**/pyproject.toml"
+
+      - name: 🐍 Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.13"
+
+      - name: 📦 Install Hatch
+        uses: pypa/hatch@install
+
+      - name: 🧪 Type Tests
+        run: hatch run 

pyproject.toml

@@ -59,14 +59,20 @@ exclude = ["/.github", "/.idea", "/docs"]
 installer = "uv"
 python = "3.13"
 dependencies = [
-    "pyinstaller==6.12.0",
-    "basedpyright==1.28.5",
+    "pyinstaller==6.13.0",
+    "basedpyright==1.29.1",
+    "pytest==8.3.5",
+    "pytest-cov==6.1.1",
+    "pytest-mock==3.14.0",
+    "pytest-asyncio==0.26.0"
 ]
 [tool.hatch.envs.default.scripts]
 exe = "pyinstaller.exe ephys_link.spec -y -- -d && pyinstaller.exe ephys_link.spec -y"
 exe-clean = "pyinstaller.exe ephys_link.spec -y --clean"
 check = "basedpyright"
 check-watched = "basedpyright --watch"
+tests = "pytest"
+cov = "pytest --cov=ephys_link --cov-report=html --cov-report=term-missing"
 
 [tool.hatch.envs.docs]
 installer = "uv"
@@ -88,5 +94,16 @@ exclude = ["typings"]
 unsafe-fixes = true
 
 [tool.basedpyright]
-include = ["src/ephys_link"]
-strict = ["src/ephys_link"]
+include = ["src/ephys_link", "tests"]
+strict = ["src/ephys_link", "tests"]
+
+[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "function"
+
+[tool.coverage.run]
+source_pkgs = ["ephys_link"]
+branch = true
+omit = [
+    "tests/*",
+    "scripts/*",
+]

scripts/move_tester.py

@@ -4,7 +4,7 @@
 from vbl_aquarium.models.unity import Vector4
 
 from ephys_link.back_end.platform_handler import PlatformHandler
-from ephys_link.utils.console import Console
+from ephys_link.front_end.console import Console
 
 c = Console(enable_debug=True)
 p = PlatformHandler(EphysLinkOptions(type="pathfinder-mpm"), c)

src/ephys_link/__main__.py

@@ -16,9 +16,9 @@
 from ephys_link.back_end.platform_handler import PlatformHandler
 from ephys_link.back_end.server import Server
 from ephys_link.front_end.cli import CLI
+from ephys_link.front_end.console import Console
 from ephys_link.front_end.gui import GUI
-from ephys_link.utils.console import Console
-from ephys_link.utils.startup import check_for_updates, preamble
+from ephys_link.utils.startup import check_for_updates, get_binding_instance, preamble
 
 
 def main() -> None:
@@ -37,13 +37,16 @@ def main() -> None:
     if not options.ignore_updates:
         check_for_updates(console)
 
-    # 4. Instantiate the Platform Handler with the appropriate platform bindings.
-    platform_handler = PlatformHandler(options, console)
+    # 4. Instantiate the requested platform binding.
+    binding = get_binding_instance(options, console)
 
-    # 5. Add hotkeys for emergency stop.
+    # 5. Instantiate the Platform Handler with the appropriate platform bindings.
+    platform_handler = PlatformHandler(binding, console)
+
+    # 6. Add hotkeys for emergency stop.
     _ = add_hotkey("ctrl+alt+shift+q", lambda: run(platform_handler.emergency_stop()))
 
-    # 6. Start the server.
+    # 7. Start the server.
     Server(options, platform_handler, console).launch()
 
 

src/ephys_link/back_end/platform_handler.py

@@ -8,12 +8,10 @@
 """
 
 from typing import final
-from uuid import uuid4
 
 from vbl_aquarium.models.ephys_link import (
     AngularResponse,
     BooleanStateResponse,
-    EphysLinkOptions,
     GetManipulatorsResponse,
     PlatformInfo,
     PositionalResponse,
@@ -23,81 +21,38 @@
     SetPositionRequest,
     ShankCountResponse,
 )
-from vbl_aquarium.models.unity import Vector4
 
-from ephys_link.bindings.mpm_binding import MPMBinding
+from ephys_link.front_end.console import Console
 from ephys_link.utils.base_binding import BaseBinding
-from ephys_link.utils.console import Console
+from ephys_link.utils.constants import (
+    EMERGENCY_STOP_MESSAGE,
+    NO_SET_POSITION_WHILE_INSIDE_BRAIN_ERROR,
+    did_not_reach_target_depth_error,
+    did_not_reach_target_position_error,
+)
 from ephys_link.utils.converters import vector4_to_array
-from ephys_link.utils.startup import get_bindings
 
 
 @final
 class PlatformHandler:
     """Handler for platform commands."""
 
-    def __init__(self, options: EphysLinkOptions, console: Console) -> None:
+    def __init__(self, binding: BaseBinding, console: Console) -> None:
         """Initialize platform handler.
 
         Args:
-            options: CLI options.
+            binding: Binding instance for the platform.
             console: Console instance.
         """
-        # Store the CLI options.
-        self._options = options
-
         # Store the console.
         self._console = console
 
         # Define bindings based on platform type.
-        self._bindings = self._get_binding_instance(options)
+        self._bindings = binding
 
         # Record which IDs are inside the brain.
         self._inside_brain: set[str] = set()
 
-        # Generate a Pinpoint ID for proxy usage.
-        self._pinpoint_id = str(uuid4())[:8]
-
-    def _get_binding_instance(self, options: EphysLinkOptions) -> BaseBinding:
-        """Match the platform type to the appropriate bindings.
-
-        Args:
-            options: CLI options.
-
-        Raises:
-            ValueError: If the platform type is not recognized.
-
-        Returns:
-            Bindings for the specified platform type.
-        """
-
-        # What the user supplied.
-        selected_type = options.type
-
-        for binding_type in get_bindings():
-            binding_cli_name = binding_type.get_cli_name()
-
-            # Notify deprecation of "ump-4" and "ump-3" CLI options and fix.
-            if selected_type in ("ump-4", "ump-3"):
-                self._console.error_print(
-                    "DEPRECATION",
-                    f"CLI option '{selected_type}' is deprecated and will be removed in v3.0.0. Use 'ump' instead.",
-                )
-                selected_type = "ump"
-
-            if binding_cli_name == selected_type:
-                # Pass in HTTP port for Pathfinder MPM.
-                if binding_cli_name == "pathfinder-mpm":
-                    return MPMBinding(options.mpm_port)
-
-                # Otherwise just return the binding.
-                return binding_type()
-
-        # Raise an error if the platform type is not recognized.
-        error_message = f'Platform type "{options.type}" not recognized.'
-        self._console.critical_print(error_message)
-        raise ValueError(error_message)
-
     # Platform metadata.
 
     def get_display_name(self) -> str:
@@ -152,7 +107,7 @@ async def get_position(self, manipulator_id: str) -> PositionalResponse:
             )
         except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Get Position", e)
-            return PositionalResponse(error=str(e))
+            return PositionalResponse(error=self._console.pretty_exception(e))
         else:
             return PositionalResponse(position=unified_position)
 
@@ -202,9 +157,8 @@ async def set_position(self, request: SetPositionRequest) -> PositionalResponse:
         try:
             # Disallow setting manipulator position while inside the brain.
             if request.manipulator_id in self._inside_brain:
-                error_message = 'Can not move manipulator while inside the brain. Set the depth ("set_depth") instead.'
-                self._console.error_print("Set Position", error_message)
-                return PositionalResponse(error=error_message)
+                self._console.error_print("Set Position", NO_SET_POSITION_WHILE_INSIDE_BRAIN_ERROR)
+                return PositionalResponse(error=NO_SET_POSITION_WHILE_INSIDE_BRAIN_ERROR)
 
             # Move to the new position.
             final_platform_position = await self._bindings.set_position(
@@ -222,11 +176,7 @@ async def set_position(self, request: SetPositionRequest) -> PositionalResponse:
 
                 # Check if the axis is within the movement tolerance.
                 if abs(axis) > self._bindings.get_movement_tolerance():
-                    error_message = (
-                        f"Manipulator {request.manipulator_id} did not reach target"
-                        f" position on axis {list(Vector4.model_fields.keys())[index]}."
-                        f" Requested: {request.position}, got: {final_unified_position}."
-                    )
+                    error_message = did_not_reach_target_position_error(request, index, final_unified_position)
                     self._console.error_print("Set Position", error_message)
                     return PositionalResponse(error=error_message)
         except Exception as e:  # noqa: BLE001
@@ -246,26 +196,22 @@ async def set_depth(self, request: SetDepthRequest) -> SetDepthResponse:
         """
         try:
             # Move to the new depth.
-            final_platform_depth = await self._bindings.set_depth(
+            final_depth = await self._bindings.set_depth(
                 manipulator_id=request.manipulator_id,
-                depth=self._bindings.unified_space_to_platform_space(Vector4(w=request.depth)).w,
+                depth=request.depth,
                 speed=request.speed,
             )
-            final_unified_depth = self._bindings.platform_space_to_unified_space(Vector4(w=final_platform_depth)).w
 
             # Return error if movement did not reach target within tolerance.
-            if abs(final_unified_depth - request.depth) > self._bindings.get_movement_tolerance():
-                error_message = (
-                    f"Manipulator {request.manipulator_id} did not reach target depth."
-                    f" Requested: {request.depth}, got: {final_unified_depth}."
-                )
+            if abs(final_depth - request.depth) > self._bindings.get_movement_tolerance():
+                error_message = did_not_reach_target_depth_error(request, final_depth)
                 self._console.error_print("Set Depth", error_message)
                 return SetDepthResponse(error=error_message)
         except Exception as e:  # noqa: BLE001
             self._console.exception_error_print("Set Depth", e)
             return SetDepthResponse(error=self._console.pretty_exception(e))
         else:
-            return SetDepthResponse(depth=final_unified_depth)
+            return SetDepthResponse(depth=final_depth)
 
     async def set_inside_brain(self, request: SetInsideBrainRequest) -> BooleanStateResponse:
         """Mark a manipulator as inside the brain or not.
@@ -278,16 +224,11 @@ async def set_inside_brain(self, request: SetInsideBrainRequest) -> BooleanState
         Returns:
             Inside brain state of the manipulator and an error message if any.
         """
-        try:
-            if request.inside:
-                self._inside_brain.add(request.manipulator_id)
-            else:
-                self._inside_brain.discard(request.manipulator_id)
-        except Exception as e:  # noqa: BLE001
-            self._console.exception_error_print("Set Inside Brain", e)
-            return BooleanStateResponse(error=self._console.pretty_exception(e))
+        if request.inside:
+            self._inside_brain.add(request.manipulator_id)
         else:
-            return BooleanStateResponse(state=request.inside)
+            self._inside_brain.discard(request.manipulator_id)
+        return BooleanStateResponse(state=request.inside)
 
     async def stop(self, manipulator_id: str) -> str:
         """Stop a manipulator.
@@ -316,12 +257,12 @@ async def stop_all(self) -> str:
             for manipulator_id in await self._bindings.get_manipulators():
                 await self._bindings.stop(manipulator_id)
         except Exception as e:  # noqa: BLE001
-            self._console.exception_error_print("Stop", e)
+            self._console.exception_error_print("Stop All", e)
             return self._console.pretty_exception(e)
         else:
             return ""
 
     async def emergency_stop(self) -> None:
         """Stops all manipulators with a message."""
-        self._console.critical_print("Emergency Stopping All Manipulators...")
+        self._console.critical_print(EMERGENCY_STOP_MESSAGE)
         _ = await self.stop_all()