chore: update

Author: MrBlenny

platform_cli/groups/ros.py

@@ -5,7 +5,14 @@
 
 
 from platform_cli.groups.base import PlatformCliGroup
-from platform_cli.helpers import get_ros_env, echo, call, start_watcher
+from platform_cli.helpers import (
+    get_ros_env,
+    echo,
+    call,
+    start_watcher,
+    get_changed_files,
+    find_packages_for_changed_files,
+)
 
 
 def get_ros_poetry_packages(path: Path) -> List[Path]:
@@ -23,6 +30,20 @@ def get_ros_poetry_packages(path: Path) -> List[Path]:
     return ros_poetry_packages
 
 
+def get_changed_packages(change_base: str) -> List[str]:
+    """Get list of ROS packages that have changes since base branch"""
+    workspace_path = Path.cwd()
+
+    if not (workspace_path / ".git").exists():
+        return []
+
+    changed_files = get_changed_files(change_base, workspace_path)
+    if not changed_files:
+        return []
+
+    return find_packages_for_changed_files(changed_files, workspace_path)
+
+
 def collect_xunit_xmls(destination: Path, package: str):  # type: ignore
     """Collects all pytest, gtest XML files from the test results"""
 
@@ -101,30 +122,69 @@ def command():
             default=2,
             help="Number of times to retest until pass",
         )
+        @click.option(
+            "--changed",
+            type=bool,
+            is_flag=True,
+            default=False,
+            help="Only test packages with changes since base branch",
+        )
+        @click.option(
+            "--change-base",
+            type=str,
+            default="main",
+            help="Base branch to compare changes against (used with --changed)",
+        )
+        @click.option(
+            "--include-dependents",
+            type=bool,
+            is_flag=True,
+            default=False,
+            help="Also test packages that depend on changed packages (used with --changed)",
+        )
         @click.argument("args", nargs=-1)
         def test(
             package: str,
             results_dir: Path,
             watch: bool,
             build: bool,
             retest_until_pass: int,
+            changed: bool,
+            change_base: str,
+            include_dependents: bool,
             args: List[str],
         ):
             """Runs colcon test on all ROS packages"""
 
             env = get_ros_env()
             args_str = " ".join(args)
 
-            # Some args only apply to colcon test
-            args_str_test = args_str
-            args_str_build = args_str
+            # Section 1: Determine packages to test
+            packages_to_test = []
             if package:
-                args_str_test += f" --packages-select {package}"
-                args_str_build += f" --package {package}"
+                packages_to_test = [package]
+            elif changed:
+                packages_to_test = get_changed_packages(change_base)
+
+            if not packages_to_test:
+                return
+
+            # Section 2: Choose testing strategy
+            args_str_test = args_str
+            packages_str = " ".join(packages_to_test)
+
+            if include_dependents and changed:
+                args_str_test += f" --packages-up-to {packages_str}"
+            else:
+                args_str_test += f" --packages-select {packages_str}"
 
             def command():
                 if build:
+                    args_str_build = args_str
+                    if package:
+                        args_str_build += f" --package {package}"
                     call(" ".join(["platform ros build", args_str_build]))
+
                 p1 = call(
                     " ".join(
                         [

platform_cli/helpers.py

@@ -11,6 +11,50 @@
 import signal
 
 
+def get_changed_files(base_branch: str = "main", cwd: Optional[Path] = None) -> List[Path]:
+    """Get list of files changed between current HEAD and base branch"""
+    try:
+        result = stdout_call(f"git diff --name-only {base_branch}...HEAD", cwd=cwd, abort=False)
+        if not result.strip():
+            return []
+
+        workspace = Path(cwd) if cwd else Path.cwd()
+        return [
+            (workspace / line.strip()).resolve()
+            for line in result.strip().splitlines()
+            if line.strip()
+        ]
+    except Exception:
+        return []
+
+
+def find_packages_for_changed_files(changed_files: List[Path], workspace_path: Path) -> List[str]:
+    """Find ROS packages that contain the changed files"""
+    from platform_cli.groups.packaging import find_packages_with_colcon
+
+    if not changed_files:
+        return []
+
+    try:
+        all_packages = find_packages_with_colcon(workspace_path)
+        if not all_packages:
+            return []
+
+        affected_packages = set()
+        for changed_file in changed_files:
+            for package_name, package_path in all_packages.items():
+                try:
+                    changed_file.resolve().relative_to(package_path.resolve())
+                    affected_packages.add(package_name)
+                    break
+                except ValueError:
+                    continue
+
+        return list(affected_packages)
+    except Exception:
+        return []
+
+
 class FileSystemEventHandlerDebounced(FileSystemEventHandler):
     def __init__(
         self,