Added tests.

Author: brittleru

drone_sim_runner/sphx/sim_config.py

@@ -123,13 +123,3 @@ def _prepare_temp_config(original_config_path: Path) -> Path:
     def _validate_path(path_: Path):
         if not path_.exists():
             raise FileNotFoundError(f"File does not exist: {path_}")
-
-
-if __name__ == '__main__':
-    base_dir = Path(__file__).parent.parent.parent
-    path_config = PathConfig.from_yaml(
-        "",
-        base_dir / "examples" / "example-config.yaml",
-        base_dir,
-        "")
-    print(path_config)

tests/core/__init__.py

@@ -0,0 +1,99 @@
+import subprocess
+import unittest
+from unittest.mock import MagicMock, patch
+
+import psutil
+
+from drone_sim_runner.core.process_manager import ProcessManager
+
+
+class TestProcessManager(unittest.TestCase):
+
+    def test_verify_process_running_none(self):
+        """Should return False if process is None."""
+        result = ProcessManager.verify_process_running(None, "test_proc")
+        self.assertFalse(result)
+
+    def test_verify_process_running_still_active(self):
+        """Should return True if poll() returns None (process still running)."""
+        mock_proc = MagicMock(spec=subprocess.Popen)
+        mock_proc.poll.return_value = None
+
+        result = ProcessManager.verify_process_running(mock_proc, "test_proc")
+        self.assertTrue(result)
+
+    def test_verify_process_running_terminated(self):
+        """Should return False and print error if process has terminated."""
+        mock_proc = MagicMock(spec=subprocess.Popen)
+
+        mock_stderr = MagicMock()
+        mock_stderr.read.return_value = b"Some error occurred"
+        mock_proc.stderr = mock_stderr
+
+        mock_proc.poll.return_value = 1
+
+        result = ProcessManager.verify_process_running(mock_proc, "test_proc")
+
+        self.assertFalse(result)
+        mock_stderr.read.assert_called_once()
+
+    @patch('psutil.process_iter')
+    def test_clean_remaining_processes(self, mock_iter):
+        """Should call kill() on processes that match the name."""
+        proc1 = MagicMock()
+        proc1.info = {'name': 'my_app_process'}
+
+        proc2 = MagicMock()
+        proc2.info = {'name': 'other_process'}
+
+        mock_iter.return_value = [proc1, proc2]
+
+        ProcessManager.clean_remaining_processes("my_app")
+
+        proc1.kill.assert_called_once()
+        proc2.kill.assert_not_called()
+
+    @patch('psutil.process_iter')
+    def test_clean_remaining_processes_access_denied(self, mock_iter):
+        """Should handle AccessDenied exceptions gracefully."""
+        proc1 = MagicMock()
+        proc1.info = {'name': 'admin_proc'}
+        proc1.kill.side_effect = psutil.AccessDenied()
+
+        mock_iter.return_value = [proc1]
+
+        ProcessManager.clean_remaining_processes("admin_proc")
+        proc1.kill.assert_called_once()
+
+    def test_kill_process_tree_none_pid(self):
+        """Should return immediately if PID is None."""
+        with patch('psutil.Process') as mock_ps:
+            ProcessManager.kill_process_tree(None)
+            mock_ps.assert_not_called()
+
+    @patch('psutil.Process')
+    def test_kill_process_tree_recursive(self, mock_ps_class):
+        """Should kill children first, then the parent."""
+        mock_parent = MagicMock()
+        mock_child1 = MagicMock()
+        mock_child2 = MagicMock()
+
+        mock_ps_class.return_value = mock_parent
+        mock_parent.children.return_value = [mock_child1, mock_child2]
+
+        ProcessManager.kill_process_tree(1234)
+
+        mock_child1.kill.assert_called_once()
+        mock_child2.kill.assert_called_once()
+        mock_parent.kill.assert_called_once()
+
+    @patch('psutil.Process')
+    def test_kill_process_tree_not_found(self, mock_ps_class):
+        """Should handle NoSuchProcess if the PID doesn't exist."""
+        mock_ps_class.side_effect = psutil.NoSuchProcess(pid=9999)
+
+        ProcessManager.kill_process_tree(9999)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/core/test_process_manager.py

@@ -0,0 +1,41 @@
+import unittest
+from unittest.mock import patch
+
+from drone_sim_runner.core.script_config import ScriptConfig
+
+
+class TestScriptConfig(unittest.TestCase):
+
+    @patch('pathlib.Path.exists')
+    def test_valid_config(self, mock_exists):
+        """Should initialize correctly with valid path and args."""
+        mock_exists.return_value = True
+        config = ScriptConfig(script_path="/fake/path/script.sh", script_args=["--speed", "10"])
+        self.assertEqual(config.script_path, "/fake/path/script.sh")
+        self.assertEqual(config.script_args, ["--speed", "10"])
+
+    @patch('pathlib.Path.exists')
+    def test_missing_file_raises_error(self, mock_exists):
+        """Should raise FileNotFoundError if path doesn't exist."""
+        mock_exists.return_value = False
+        with self.assertRaises(FileNotFoundError):
+            ScriptConfig(script_path="invalid_path.py")
+
+    @patch('pathlib.Path.exists')
+    def test_invalid_args_type(self, mock_exists):
+        """Should raise ValueError if script_args is not a list."""
+        mock_exists.return_value = True
+        with self.assertRaises(ValueError) as cm:
+            ScriptConfig(script_path="script.py", script_args="not a list")
+        self.assertEqual(str(cm.exception), "script_args must be a list")
+
+    @patch('pathlib.Path.exists')
+    def test_empty_args_list(self, mock_exists):
+        """Should raise ValueError if script_args is empty."""
+        mock_exists.return_value = True
+        with self.assertRaises(ValueError):
+            ScriptConfig(script_path="script.py", script_args=[])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/core/test_script_config.py

@@ -0,0 +1,98 @@
+import subprocess
+import unittest
+from unittest.mock import MagicMock, patch
+
+from drone_sim_runner.core.process_type import ProcessType
+from drone_sim_runner.core.simulation_process import SimulationProcess
+
+
+class TestSimulationProcess(unittest.TestCase):
+
+    def setUp(self):
+        self.cmd = ["python3", "sim.py"]
+        self.sim = SimulationProcess(
+            cmd=self.cmd,
+            name=ProcessType.SCRIPT,
+            init_indicator="READY",
+            wait_time=0,
+            timeout=5
+        )
+
+    @patch('subprocess.Popen')
+    @patch('threading.Thread')
+    @patch('time.sleep')
+    def test_start_script_process(self, mock_sleep, mock_thread, mock_popen):
+        """Should start the process and launch output thread for SCRIPTS."""
+        mock_popen.return_value = MagicMock(spec=subprocess.Popen)
+
+        self.sim.start()
+
+        self.assertTrue(self.sim.is_running)
+        mock_popen.assert_called_once()
+        mock_thread.assert_called_once_with(target=self.sim._log_output, daemon=True)
+
+    @patch('subprocess.Popen')
+    @patch('drone_sim_runner.core.process_manager.ProcessManager.verify_process_running')
+    @patch('time.sleep')
+    def test_wait_for_initialization_success(self, mock_sleep, mock_verify, mock_popen):
+        """Should continue when init_indicator is found in stdout."""
+        self.sim.name = ProcessType.SPHINX
+
+        mock_proc = MagicMock(spec=subprocess.Popen)
+        mock_stdout = MagicMock()
+        mock_stdout.readline.side_effect = [b"Starting...\n", b"System READY\n"]
+        mock_proc.stdout = mock_stdout
+
+        mock_popen.return_value = mock_proc
+        mock_verify.return_value = True
+
+        self.sim.start()
+
+        self.assertEqual(mock_stdout.readline.call_count, 2)
+
+    @patch('subprocess.Popen')
+    @patch('time.time')
+    @patch('drone_sim_runner.core.process_manager.ProcessManager.verify_process_running')
+    def test_wait_for_initialization_timeout(self, mock_verify, mock_time, mock_popen):
+        """Should raise TimeoutError if indicator never appears."""
+        self.sim.name = ProcessType.FIRMWARE
+        mock_proc = MagicMock(spec=subprocess.Popen)
+
+        mock_stdout = MagicMock()
+        mock_stdout.readline.return_value = b"just some noise\n"
+        mock_proc.stdout = mock_stdout
+
+        mock_popen.return_value = mock_proc
+        mock_verify.return_value = True
+
+        mock_time.side_effect = [0, 10]
+
+        with self.assertRaises(TimeoutError):
+            self.sim._wait_for_initialization()
+
+    @patch('drone_sim_runner.core.process_manager.ProcessManager.kill_process_tree')
+    def test_stop_process(self, mock_kill):
+        """Should kill the process tree and stop running."""
+        mock_proc = MagicMock()
+        mock_proc.pid = 1234
+        self.sim.process = mock_proc
+        self.sim.is_running = True
+
+        self.sim.stop()
+
+        self.assertFalse(self.sim.is_running)
+        self.assertIsNone(self.sim.process)
+        mock_kill.assert_called_once_with(1234)
+
+    @patch('subprocess.Popen')
+    def test_start_failure_raises_runtime_error(self, mock_popen):
+        """Should wrap exceptions in a RuntimeError."""
+        mock_popen.side_effect = OSError("Binary not found")
+
+        with self.assertRaises(RuntimeError) as cm:
+            self.sim.start()
+        self.assertIn("Failed to start 'Script'", str(cm.exception))
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/core/test_simulation_process.py

@@ -0,0 +1,108 @@
+import unittest
+from pathlib import Path
+from unittest.mock import patch, mock_open
+
+from drone_sim_runner.sphx.sim_config import SimulationConfig, PathConfig
+
+
+class TestSimulationConfig(unittest.TestCase):
+    def test_valid_init(self):
+        """Should initialize correctly with valid types."""
+        config = SimulationConfig(firmware_command="run_fw", sphinx_command="/bin/sphinx")
+        self.assertEqual(config.sphinx_cmd, "/bin/sphinx")
+
+    def test_invalid_types(self):
+        """Should raise ValueError for incorrect types."""
+        with self.assertRaises(ValueError):
+            SimulationConfig(firmware_command=123, sphinx_command="path")
+        with self.assertRaises(ValueError):
+            SimulationConfig(firmware_command="cmd", sphinx_command=123)
+
+    def test_negative_waits(self):
+        """Should raise ValueError if any time-based attribute is negative."""
+        with self.assertRaises(ValueError):
+            SimulationConfig(firmware_command="cmd", sphinx_command="cmd", sphinx_init_wait=-1)
+
+    def test_negative_values_raise_error(self):
+        """Post-init should catch negative timeouts."""
+        with self.assertRaises(ValueError):
+            SimulationConfig(
+                firmware_command="run",
+                sphinx_command="path",
+                timeout_seconds=-10
+            )
+
+
+class TestPathConfig(unittest.TestCase):
+
+    def setUp(self):
+        self.yaml_data = """
+paths:
+  sphinx:
+    command: "sphinx-binary"
+    level: "low"
+    quality: "high"
+    firmware:
+      base_command: "launch"
+      drone_path: "/drones/anafi"
+      firmware_url: "http://fw.local"
+      pose: "0,0,0"
+  scripts:
+    controller: "main.py"
+    save_dir_name: "test_run"
+    args:
+      --out: "${output_dir}"
+"""
+
+    @patch('pathlib.Path.exists', return_value=True)
+    @patch('builtins.open', new_callable=mock_open)
+    def test_from_yaml_parsing(self, mock_file, mock_exists):
+        """Verify YAML data is correctly transformed into shell commands."""
+        mock_file.side_effect = [
+            mock_open(read_data=self.yaml_data).return_value,
+            mock_open(read_data="").return_value
+        ]
+
+        config = PathConfig.from_yaml(
+            sphinx_base_dir="/opt/sphinx",
+            config_path="config.yaml",
+            project_base_dir="/home/user/project",
+            output_dir="/tmp/results"
+        )
+
+        # 1. Verify Firmware Command string assembly
+        expected_fw = 'launch "/drones/anafi::pose=0,0,0::firmware=http://fw.local"'
+        self.assertEqual(config.firmware_command, expected_fw)
+
+        # 2. Verify Sphinx Command string assembly
+        self.assertIn("/opt/sphinx/sphinx-binary", config.sphinx_command)
+        self.assertIn("-level=low", config.sphinx_command)
+        self.assertIn("-quality=high", config.sphinx_command)
+
+        # 3. Verify Script Args Replacement
+        self.assertIn("--out=/tmp/results/test_run", config.script_args)
+
+    @patch('tempfile.mkstemp')
+    @patch('builtins.open', new_callable=mock_open, read_data="${MODELS_DIR}/drone.cpp")
+    @patch('pathlib.Path.exists', return_value=True)
+    def test_prepare_temp_config_regex(self, mock_exists, mock_file, mock_temp):
+        """Verify that ${MODELS_DIR} is correctly replaced in config files."""
+        mock_temp.return_value = (None, "/tmp/sphinx_config_123.yaml")
+
+        # original_path: /home/user/project/config/file.yaml
+        # bunker_dir (parent.parent): /home/user/project
+        # models_dir: /home/user/project/models
+        original_path = Path("/home/user/project/config/file.yaml")
+
+        temp_file = PathConfig._prepare_temp_config(original_path)
+
+        write_handle = mock_file.return_value
+        written_content = "".join(call.args[0] for call in write_handle.write.call_args_list)
+
+        self.assertEqual(temp_file, Path("/tmp/sphinx_config_123.yaml"))
+        self.assertIn("/home/user/project/models", written_content)
+        self.assertNotIn("${MODELS_DIR}", written_content)
+
+
+if __name__ == '__main__':
+    unittest.main()