Implement RAM Transitions Tests

.github/workflows/pytest_transitions_tests_coverage.yaml

@@ -0,0 +1,29 @@
+name: Pytest with Coverage
+
+on:
+  pull_request:
+    branches: [humble-devel]
+  push:
+    branches: [humble-devel, testing]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.10.12'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r test/requirements.txt
+          pip install pytest pytest-cov
+
+      - name: Run tests with coverage
+        run: |
+          PYTHONPATH=. pytest --cov=manager/manager --cov-report=term

.gitignore

@@ -3,4 +3,8 @@ __pycache__/
 /.idea
 
 # IDEs
-.vscode
+.vscode
+
+# Log Files
+*.log
+*.coverage

manager/comms/new_consumer.py

@@ -1,3 +1,9 @@
+"""
+WebSocket consumer module for the Robotics Application Manager (RAM).
+
+Handles client connections, message processing, and communication with manager queue.
+"""
+
 import json
 import logging
 from queue import Queue
@@ -8,25 +14,37 @@
     ManagerConsumerMessageException,
     ManagerConsumerMessage,
 )
-from manager.comms.websocker_server import WebsocketServer
+from manager.comms.websocket_server import WebsocketServer
 from manager.ram_logging.log_manager import LogManager
 
 
 class Client:
+    """Represents a client connected to the WebSocket server."""
+
     def __init__(self, **kwargs):
+        """Initialize a Client instance with id, handler, and address."""
         self.id = kwargs["id"]
         self.handler = kwargs["handler"]
         self.address = kwargs["address"]
 
 
 class ManagerConsumer:
     """
-    Websocket server consumer for new Robotics Application Manager aka. RAM
+    Websocket server consumer for new Robotics Application Manager aka: RAM.
+
     Supports single client connection to RAM
     TODO: Better handling of single client connections, closing and redirecting
     """
 
     def __init__(self, host, port, manager_queue: Queue):
+        """
+        Initialize the ManagerConsumer with host, port, and manager_queue.
+
+        Args:
+            host (str): The host address for the WebSocket server.
+            port (int): The port number for the WebSocket server.
+            manager_queue (Queue): The queue for communication with the manager.
+        """
         self.host = host
         self.port = port
         self.server = WebsocketServer(host=host, port=port, loglevel=logging.INFO)
@@ -37,7 +55,8 @@ def __init__(self, host, port, manager_queue: Queue):
         ws_logger.setLevel(logging.INFO)
         ws_logger.handlers.clear()
         ws_formatter = logging.Formatter(
-            "%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] (%(name)s)  %(message)s",
+            "%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] "
+            "(%(name)s)  %(message)s",
             "%H:%M:%S",
         )
         ws_console_handler = logging.StreamHandler()
@@ -51,11 +70,25 @@ def __init__(self, host, port, manager_queue: Queue):
         self.manager_queue = manager_queue
 
     def handle_client_new(self, client, server):
+        """
+        Handle a new client connection event.
+
+        Args:
+            client: The client object representing the connected client.
+            server: The WebSocket server instance.
+        """
         LogManager.logger.info(f"client connected: {client}")
         self.client = client
         self.server.deny_new_connections()
 
     def handle_client_disconnect(self, client, server):
+        """
+        Handle a client disconnection event.
+
+        Args:
+            client: The client object representing the disconnected client.
+            server: The WebSocket server instance.
+        """
         if client is None:
             return
         LogManager.logger.info(f"client disconnected: {client}")
@@ -70,6 +103,14 @@ def handle_client_disconnect(self, client, server):
         self.server.allow_new_connections()
 
     def handle_message_received(self, client, server, websocket_message):
+        """
+        Handle a message received from a client.
+
+        Args:
+            client: The client object that sent the message.
+            server: The WebSocket server instance.
+            websocket_message (str): The message received from the client.
+        """
         LogManager.logger.info(
             f"message received length: {len(websocket_message)} from client {client}"
         )
@@ -90,6 +131,15 @@ def handle_message_received(self, client, server, websocket_message):
             raise e
 
     def send_message(self, message_data, command=None):
+        """
+        Send a message to the connected client.
+
+        Args:
+            message_data: The message data to send, can be a ManagerConsumerMessage,
+                ManagerConsumerMessageException, or other data.
+            command (str, optional): The command associated with the message,
+                used if message_data is not a ManagerConsumerMessage.
+        """
         if self.client is not None and self.server is not None:
             if isinstance(message_data, ManagerConsumerMessage):
                 message = message_data
@@ -103,7 +153,9 @@ def send_message(self, message_data, command=None):
             self.server.send_message(self.client, str(message))
 
     def start(self):
+        """Start the WebSocket server in a separate thread."""
         self.server.run_forever(threaded=True)
 
     def stop(self):
+        """Stop the WebSocket server gracefully."""
         self.server.shutdown_gracefully()

manager/libs/launch_world_model.py

@@ -1,3 +1,5 @@
+"""Module for managing and validating robotics application launch world configs."""
+
 from dataclasses import dataclass
 from typing import Optional
 from pydantic import BaseModel, ValidationError
@@ -6,6 +8,8 @@
 
 
 class ConfigurationModel(BaseModel):
+    """Pydantic model for robotics application world type and launch file config."""
+
     type: str
     launch_file_path: str
 
@@ -15,10 +19,23 @@ class ConfigurationModel(BaseModel):
 
 @dataclass
 class ConfigurationManager:
+    """Manager for robotics application configuration validation and storage."""
+
     configuration: ConfigurationModel
 
     @staticmethod
     def validate(configuration: dict):
+        """Validate the given configuration dictionary using the ConfigurationModel.
+
+        Args:
+            configuration (dict): The configuration data to validate.
+
+        Returns:
+            ConfigurationModel: The validated configuration model.
+
+        Raises:
+            ValueError: If the configuration is invalid.
+        """
         try:
             return ConfigurationModel(**configuration)
         except ValidationError as e:

manager/manager/launcher/launcher_robot.py

@@ -1,3 +1,5 @@
+"""LauncherRobot module for managing robot launchers in different simulation worlds."""
+
 from typing import Optional
 from pydantic import BaseModel
 
@@ -66,6 +68,8 @@
 
 
 class LauncherRobot(BaseModel):
+    """Class for managing robot launchers in different simulation worlds."""
+
     type: str
     launch_file_path: str
     module: str = ".".join(__name__.split(".")[:-1])
@@ -74,7 +78,8 @@ class LauncherRobot(BaseModel):
     start_pose: Optional[list] = []
 
     def run(self, start_pose=None):
-        if start_pose != None:
+        """Run the robot launcher with an optional start pose."""
+        if start_pose is not None:
             self.start_pose = start_pose
         for module in worlds[self.type][str(self.ros_version)]:
             module["launch_file"] = self.launch_file_path
@@ -83,13 +88,16 @@ def run(self, start_pose=None):
         LogManager.logger.info(self.launchers)
 
     def terminate(self):
+        """Terminate all robot launchers and clear the launchers list."""
         LogManager.logger.info("Terminating robot launcher")
         if self.launchers:
             for launcher in self.launchers:
                 launcher.terminate()
         self.launchers = []
 
     def launch_module(self, configuration):
+        """Launch a robot module based on the provided configuration."""
+
         def process_terminated(name, exit_code):
             LogManager.logger.info(
                 f"LauncherEngine: {name} exited with code {exit_code}"
@@ -98,17 +106,24 @@ def process_terminated(name, exit_code):
                 self.terminated_callback(name, exit_code)
 
         launcher_module_name = configuration["module"]
-        launcher_module = f"{self.module}.launcher_{launcher_module_name}.Launcher{class_from_module(launcher_module_name)}"
+        launcher_module = (
+            f"{self.module}.launcher_{launcher_module_name}."
+            f"Launcher{class_from_module(launcher_module_name)}"
+        )
         launcher_class = get_class(launcher_module)
         launcher = launcher_class.from_config(launcher_class, configuration)
 
         launcher.run(self.start_pose, process_terminated)
         return launcher
 
     def launch_command(self, configuration):
+        """Launch a robot command based on the provided configuration."""
         pass
 
 
 class LauncherRobotException(Exception):
+    """Exception class for errors related to LauncherRobot."""
+
     def __init__(self, message):
+        """Initialize the LauncherRobotException with a message."""
         super(LauncherRobotException, self).__init__(message)

manager/manager/lint/linter.py

@@ -1,12 +1,28 @@
+"""Linter module for evaluating and cleaning Python code using pylint."""
+
 import glob
 import re
-import os
+import os  # noqa: F401
 import subprocess
 import tempfile
 
 
 class Lint:
+    """Class for evaluating and cleaning Python code using pylint."""
+
     def clean_pylint_output(self, result, warnings=False):
+        """
+        Clean the output from pylint.
+
+        By removing unwanted messages and formatting errors.
+
+        Args:
+            result (str): The output string from pylint.
+            warnings (bool): Whether to include warnings in the output.
+
+        Returns:
+            str: The cleaned and formatted output.
+        """
 
         # result = result.replace(os.path.basename(code_file_name), 'user_code')
         # Define the patterns to remove
@@ -16,13 +32,15 @@ def clean_pylint_output(self, result, warnings=False):
             r":[0-9]+:[0-9]+: R[0-9]{4}:.*",  # Refactor messages
             r":[0-9]+:[0-9]+: error.*EOF.*",  # Unexpected EOF error
             r":[0-9]+:[0-9]+: E1101:.*Module 'ompl.*",  # ompl E1101 error
-            r":[0-9]+:[0-9]+:.*value.*argument.*unbound.*method.*",  # No value for argument 'self' error
+            (
+                r":[0-9]+:[0-9]+:.*value.*argument.*unbound.*method.*"
+            ),  # No value for argument 'self' error
             r":[0-9]+:[0-9]+: E1111:.*",  # Assignment from no return error
             r":[0-9]+:[0-9]+: E1136:.*",  # E1136 until issue is resolved
         ]
 
         if not warnings:
-            # Remove convention, refactor, and warning messages if warnings are not desired
+            # Remove convention, refactor, and warning msgs if warnings are not desired
             for pattern in patterns[:3]:
                 result = re.sub(r"^[^:]*" + pattern, "", result, flags=re.MULTILINE)
 
@@ -43,6 +61,15 @@ def clean_pylint_output(self, result, warnings=False):
         return result
 
     def append_rating_if_missing(self, result):
+        """
+        Append a default rating message to the result if it is missing.
+
+        Args:
+            result (str): The output string from pylint.
+
+        Returns:
+            str: The result string with the rating message appended if necessary.
+        """
         rating_message = (
             "-----------------------------------\nYour code has been rated at 0.00/10"
         )
@@ -58,13 +85,28 @@ def append_rating_if_missing(self, result):
     def evaluate_code(
         self, code, ros_version, warnings=False, py_lint_source="pylint_checker.py"
     ):
+        """
+        Evaluate the provided Python code using pylint and return the cleaned output.
+
+        Args:
+            code (str): The Python code to evaluate.
+            ros_version (str): The ROS version to determine environment settings.
+            warnings (bool, optional): Whether to include warnings in the output.
+                Defaults to False.
+            py_lint_source (str, optional): The pylint checker source file.
+                Defaults to "pylint_checker.py".
+
+        Returns:
+            str: The cleaned and formatted pylint output.
+        """
         try:
             code = re.sub(r"from HAL import HAL", "from hal import HAL", code)
             code = re.sub(r"from GUI import GUI", "from gui import GUI", code)
             code = re.sub(r"from MAP import MAP", "from map import MAP", code)
             code = re.sub(r"\nimport cv2\n", "\nfrom cv2 import cv2\n", code)
 
-            # Avoids EOF error when iterative code is empty (which prevents other errors from showing)
+            # Avoids EOF error when iterative code is empty
+            # (which prevents other errors from showing)
             while_position = re.search(
                 r"[^ ]while\s*\(\s*True\s*\)\s*:|[^ ]while\s*True\s*:|[^ ]while\s*1\s*:|[^ ]while\s*\(\s*1\s*\)\s*:",
                 code,
@@ -89,9 +131,17 @@ def evaluate_code(
 
             command = ""
             if "humble" in str(ros_version):
-                command = f"export PYTHONPATH=$PYTHONPATH:/workspace/code; python3 /RoboticsApplicationManager/manager/manager/lint/{py_lint_source}"
+                command = (
+                    f"export PYTHONPATH=$PYTHONPATH:/workspace/code; "
+                    f"python3 /RoboticsApplicationManager/manager/manager/lint/"
+                    f"{py_lint_source}"
+                )
             else:
-                command = f"export PYTHONPATH=$PYTHONPATH:/workspace/code; python3 /RoboticsApplicationManager/manager/manager/lint/{py_lint_source}"
+                command = (
+                    f"export PYTHONPATH=$PYTHONPATH:/workspace/code; "
+                    f"python3 /RoboticsApplicationManager/manager/manager/lint/"
+                    f"{py_lint_source}"
+                )
 
             ret = subprocess.run(command, capture_output=True, text=True, shell=True)