Merge pull request #276 from Apolo151/ram_tests_fix

fix: ram tests structure

Author: javizqh

.github/workflows/pytest_transitions_tests_coverage.yaml

@@ -26,4 +26,4 @@ jobs:
 
       - name: Run tests with coverage
         run: |
-          PYTHONPATH=. pytest --cov=manager/manager --cov-report=term
+          PYTHONPATH=. pytest --cov=robotics_application_manager --cov-report=term

test/conftest.py

@@ -1,7 +1,7 @@
 """Utility functions to transition the Manager to a specific state for testing."""
 
 import pytest
-from manager.manager.manager import Manager
+from robotics_application_manager.manager.manager import Manager
 
 
 class DummyServer:
@@ -76,8 +76,8 @@ def stop(self):
 def manager(monkeypatch):
     """Fixture to provide a Manager instance with patched dependencies for testing."""
 
-    monkeypatch.setattr("manager.comms.websocket_server.WebsocketServer", DummyServer)
-    monkeypatch.setattr("manager.manager.manager.ManagerConsumer", DummyConsumer)
+    monkeypatch.setattr("robotics_application_manager.comms.websocket_server.WebsocketServer", DummyServer)
+    monkeypatch.setattr("robotics_application_manager.manager.manager.ManagerConsumer", DummyConsumer)
 
     # Patch subprocess.check_output for ROS_DISTRO and IMAGE_TAG
     def fake_check_output(cmd, *a, **k):
@@ -91,14 +91,14 @@ def fake_check_output(cmd, *a, **k):
 
     # Patch check_gpu_acceleration where it is used
     monkeypatch.setattr(
-        "manager.manager.manager.check_gpu_acceleration", lambda x=None: "OFF"
+        "robotics_application_manager.manager.manager.check_gpu_acceleration", lambda x=None: "OFF"
     )
 
     def dummy_run(self, start_pose=None):
         print("run around")
 
     monkeypatch.setattr(
-        "manager.manager.launcher.launcher_robot.LauncherRobot.run", dummy_run
+        "robotics_application_manager.manager.launcher.launcher_robot.LauncherRobot.run", dummy_run
     )
 
     # Patch os.makedirs and os.path.isdir to avoid real FS operations
@@ -121,7 +121,7 @@ def run(self):
         def terminate(self):
             pass
 
-    monkeypatch.setattr("manager.manager.manager.LauncherWorld", DummyLauncherWorld)
+    monkeypatch.setattr("robotics_application_manager.manager.manager.LauncherWorld", DummyLauncherWorld)
 
     class DummyFileWatchdog:
         def __init__(self, path, update_callback):
@@ -146,10 +146,10 @@ def run(self, consumer):
         def terminate(self):
             pass
 
-    monkeypatch.setattr("manager.manager.manager.LauncherTools", DummyToolsLauncher)
+    monkeypatch.setattr("robotics_application_manager.manager.manager.LauncherTools", DummyToolsLauncher)
     # Deprecated
-    # monkeypatch.setattr("manager.manager.manager.Server", DummyServer)
-    # monkeypatch.setattr("manager.manager.manager.FileWatchdog", DummyFileWatchdog)
+    # monkeypatch.setattr("robotics_application_manager.manager.manager.Server", DummyServer)
+    # monkeypatch.setattr("robotics_application_manager.manager.manager.FileWatchdog", DummyFileWatchdog)
 
     # Setup Manager with dummy consumer
     m = Manager(host="localhost", port=12345)

test/test_connect_disconnect_transitions.py

@@ -84,7 +84,7 @@ def test_disconnect_invalid_process(manager, monkeypatch):
     # Mock needed methods and attributes
     monkeypatch.setattr("os.execl", lambda *args: None)
     monkeypatch.setattr(
-        "manager.manager.manager.stop_process_and_children",
+        "robotics_application_manager.manager.manager.stop_process_and_children",
         lambda proc: (_ for _ in ()).throw(Exception("Process cannot be stopped")),
     )
     manager.application_process = DummyProc()

test/test_connected_to_world_ready.py

@@ -1,9 +1,9 @@
 """Tests for transitioning Manager from 'connected' to 'world_ready' state."""
 
 import pytest
-from manager.libs.launch_world_model import ConfigurationModel
+from robotics_application_manager.libs.launch_world_model import ConfigurationModel
 from test_utils import setup_manager_to_connected
-from manager.manager.launcher.launcher_robot import worlds
+from robotics_application_manager.manager.launcher.launcher_robot import worlds
 
 valid_world_cfg = ConfigurationModel(
     type=next(iter(worlds)), launch_file_path="/path/to/launch_file.launch"
@@ -62,7 +62,8 @@ def fake_prepare_custom_universe(cfg):
         raise ValueError("Invalid world configuration")
 
     monkeypatch.setattr(
-        "manager.libs.launch_world_model.ConfigurationManager.validate", fake_validate
+        "robotics_application_manager.libs.launch_world_model.ConfigurationManager.validate",
+        fake_validate,
     )
     manager.prepare_custom_universe = fake_prepare_custom_universe
 
@@ -94,7 +95,8 @@ def fake_validate(cfg):
         return DummyConfig()
 
     monkeypatch.setattr(
-        "manager.libs.launch_world_model.ConfigurationManager.validate", fake_validate
+        "robotics_application_manager.libs.launch_world_model.ConfigurationManager.validate",
+        fake_validate,
     )
 
     invalid_robot_cfg = {"name": "", "type": ""}  # Invalid robot config

test/test_resume_and_pause_transitions.py

@@ -9,6 +9,10 @@
 class DummyProc:
     """Dummy process class for testing suspend and resume methods."""
 
+    def children(self, recursive=True):
+        """Simulate psutil child-process lookup."""
+        return []
+
     def suspend(self):
         """Simulate suspending the process."""
         pass

test/test_terminate_transitions.py

@@ -46,7 +46,7 @@ def dummy_stop_process_and_children(proc):
         pass
 
     monkeypatch.setattr(
-        "manager.manager.manager.stop_process_and_children",
+        "robotics_application_manager.manager.manager.stop_process_and_children",
         dummy_stop_process_and_children,
     )
     manager.pause_sim = lambda: None
@@ -72,7 +72,7 @@ def dummy_stop_process_and_children(proc):
         pass
 
     monkeypatch.setattr(
-        "manager.manager.manager.stop_process_and_children",
+        "robotics_application_manager.manager.manager.stop_process_and_children",
         dummy_stop_process_and_children,
     )
     manager.pause_sim = lambda: None
@@ -106,7 +106,7 @@ def test_terminate_tools_invalid_machine_error(manager, monkeypatch):
     Ensure that the transition raises an error when executed from an invalid state.
     """
     monkeypatch.setattr(
-        "manager.libs.applications.compatibility.server.Server", DummyServer
+        "robotics_application_manager.libs.server.Server", DummyServer
     )
     # Ensure the manager is in a state where it can stop
     setup_manager_to_application_running(manager, monkeypatch)

test/test_tools_ready_to_application_running.py

@@ -38,6 +38,8 @@ def fake_open(file, mode="r", *args, **kwargs):
                 return io.BytesIO()
             elif "r" in mode:
                 return io.BytesIO(b"fake zip content")
+        if isinstance(file, str) and file.startswith("/dev/pts/"):
+            return io.BytesIO() if "b" in mode else io.StringIO()
         return original_open(file, mode, *args, **kwargs)
 
     # Mock file system and subprocess operations
@@ -47,7 +49,7 @@ def fake_open(file, mode="r", *args, **kwargs):
     monkeypatch.setattr("subprocess.Popen", lambda *a, **k: DummyProc())
     monkeypatch.setattr("os.mkdir", lambda path: None)
     monkeypatch.setattr("os.path.exists", lambda path: True)
-    monkeypatch.setattr("shutil.rmtree", lambda path: None)
+    monkeypatch.setattr("shutil.rmtree", lambda path, ignore_errors=False: None)
     monkeypatch.setattr(
         "zipfile.ZipFile",
         lambda *a, **k: type(
@@ -58,7 +60,8 @@ def fake_open(file, mode="r", *args, **kwargs):
     )
     monkeypatch.setattr("base64.b64decode", lambda s: b"print('hello')")
     monkeypatch.setattr(
-        "manager.manager.manager.Manager.unpause_sim", lambda self: None
+        "robotics_application_manager.manager.manager.Manager.unpause_sim",
+        lambda self: None,
     )
     # Mock linter to return no errors
     manager.linter.evaluate_code = lambda code, ros_version: ""
@@ -85,6 +88,8 @@ def fake_open(file, mode="r", *args, **kwargs):
             import io
 
             return io.BytesIO()
+        if isinstance(file, str) and file.startswith("/dev/pts/"):
+            return io.BytesIO() if "b" in mode else io.StringIO()
         return original_open(file, mode, *args, **kwargs)
 
     monkeypatch.setattr("builtins.open", fake_open)
@@ -93,7 +98,7 @@ def fake_open(file, mode="r", *args, **kwargs):
     monkeypatch.setattr("subprocess.Popen", lambda *a, **k: None)
     monkeypatch.setattr("os.mkdir", lambda path: None)
     monkeypatch.setattr("os.path.exists", lambda path: True)
-    monkeypatch.setattr("shutil.rmtree", lambda path: None)
+    monkeypatch.setattr("shutil.rmtree", lambda path, ignore_errors=False: None)
     monkeypatch.setattr(
         "zipfile.ZipFile",
         lambda *a, **k: type(
@@ -104,7 +109,8 @@ def fake_open(file, mode="r", *args, **kwargs):
     )
     monkeypatch.setattr("base64.b64decode", lambda s: b"print('hello')")
     monkeypatch.setattr(
-        "manager.manager.manager.Manager.unpause_sim", lambda self: None
+        "robotics_application_manager.manager.manager.Manager.unpause_sim",
+        lambda self: None,
     )
     # Mock linter to return no errors
     manager.linter.evaluate_code = lambda code, ros_version: ""
@@ -127,7 +133,7 @@ def test_on_run_application_corrupt_zip(manager, monkeypatch):
     monkeypatch.setattr("os.path.exists", lambda path: True)
     monkeypatch.setattr("os.mkdir", lambda path: None)
     monkeypatch.setattr("os.listdir", lambda path: ["0", "1", "2"])
-    monkeypatch.setattr("shutil.rmtree", lambda path: None)
+    monkeypatch.setattr("shutil.rmtree", lambda path, ignore_errors=False: None)
     # Mock open for app.zip to avoid FileNotFoundError
     original_open = builtins.open
 
@@ -136,6 +142,8 @@ def fake_open(file, mode="r", *args, **kwargs):
             import io
 
             return io.BytesIO()
+        if isinstance(file, str) and file.startswith("/dev/pts/"):
+            return io.BytesIO() if "b" in mode else io.StringIO()
         return original_open(file, mode, *args, **kwargs)
 
     monkeypatch.setattr("builtins.open", fake_open)
@@ -153,7 +161,8 @@ def fake_open(file, mode="r", *args, **kwargs):
         )(),
     )
     monkeypatch.setattr(
-        "manager.manager.manager.Manager.unpause_sim", lambda self: None
+        "robotics_application_manager.manager.manager.Manager.unpause_sim",
+        lambda self: None,
     )
     manager.linter.evaluate_code = lambda code, ros_version: ""
     data = valid_app_data

test/test_utils.py

@@ -3,7 +3,7 @@
 import builtins
 import io
 
-from manager.manager.launcher.launcher_robot import worlds
+from robotics_application_manager.manager.launcher.launcher_robot import worlds
 
 
 def setup_manager_to_connected(manager, monkeypatch):
@@ -18,7 +18,7 @@ def setup_manager_to_world_ready(manager, monkeypatch):
     setup_manager_to_connected(manager, monkeypatch)
 
     # Use ConfigurationModel for valid world config
-    from manager.libs.launch_world_model import ConfigurationModel
+    from robotics_application_manager.libs.launch_world_model import ConfigurationModel
 
     valid_world_cfg = ConfigurationModel(
         type=next(iter(worlds)),  # Use the first world type
@@ -70,7 +70,7 @@ def run(self, consumer=None):
         def terminate(self):
             pass
 
-    monkeypatch.setattr("manager.manager.manager.LauncherTools", DummyToolsLauncher)
+    monkeypatch.setattr("robotics_application_manager.manager.manager.LauncherTools", DummyToolsLauncher)
 
     # Trigger visualization ready state
     manager.trigger(
@@ -107,6 +107,8 @@ def fake_open(file, mode="r", *args, **kwargs):
                 return io.BytesIO()
             elif "r" in mode:
                 return io.BytesIO(b"fake zip content")
+        if isinstance(file, str) and file.startswith("/dev/pts/"):
+            return io.BytesIO() if "b" in mode else io.StringIO()
         return original_open(file, mode, *args, **kwargs)
 
     # Mock file system and subprocess operations
@@ -116,7 +118,7 @@ def fake_open(file, mode="r", *args, **kwargs):
     monkeypatch.setattr("subprocess.Popen", lambda *a, **k: DummyProc())
     monkeypatch.setattr("os.mkdir", lambda path: None)
     monkeypatch.setattr("os.path.exists", lambda path: True)
-    monkeypatch.setattr("shutil.rmtree", lambda path: None)
+    monkeypatch.setattr("shutil.rmtree", lambda path, ignore_errors=False: None)
     monkeypatch.setattr(
         "zipfile.ZipFile",
         lambda *a, **k: type(
@@ -127,7 +129,7 @@ def fake_open(file, mode="r", *args, **kwargs):
     )
     monkeypatch.setattr("base64.b64decode", lambda s: b"print('hello')")
     monkeypatch.setattr(
-        "manager.manager.manager.Manager.unpause_sim", lambda self: None
+        "robotics_application_manager.manager.manager.Manager.unpause_sim", lambda self: None
     )
     # Mock linter to return no errors
     manager.linter.evaluate_code = lambda code, ros_version: ""