ansible-test - Support target testing with black (#84744)

test/lib/ansible_test/_internal/commands/sanity/__init__.py

@@ -872,6 +872,7 @@ class SanityScript(SanityTest, metaclass=abc.ABCMeta):
 
             self.__all_targets: bool = self.config.get('all_targets')
             self.__no_targets: bool = self.config.get('no_targets')
+            self.__split_targets: bool = self.config.get('split_targets')
             self.__include_directories: bool = self.config.get('include_directories')
             self.__include_symlinks: bool = self.config.get('include_symlinks')
             self.__error_code: str | None = self.config.get('error_code', None)
@@ -889,6 +890,7 @@ class SanityScript(SanityTest, metaclass=abc.ABCMeta):
 
             self.__all_targets = False
             self.__no_targets = True
+            self.__split_targets = False
             self.__include_directories = False
             self.__include_symlinks = False
             self.__error_code = None
@@ -925,6 +927,11 @@ class SanityScript(SanityTest, metaclass=abc.ABCMeta):
         """True if the test does not use test targets. Mutually exclusive with all_targets."""
         return self.__no_targets
 
+    @property
+    def split_targets(self) -> bool:
+        """True if the test requires target paths to be split between controller-only and target paths."""
+        return self.__split_targets
+
     @property
     def include_directories(self) -> bool:
         """True if the test targets should include directories."""
@@ -1019,6 +1026,12 @@ class SanityScript(SanityTest, metaclass=abc.ABCMeta):
                 raise ApplicationError('Unsupported output type: %s' % self.output)
 
         if not self.no_targets:
+            if self.split_targets:
+                target_paths = set(target.path for target in self.filter_remote_targets(list(targets.targets)))
+                controller_path_list = sorted(set(paths) - target_paths)
+                target_path_list = sorted(set(paths) & target_paths)
+                paths = controller_path_list + ['--'] + target_path_list
+
             data = '\n'.join(paths)
 
             if data:

test/sanity/code-smell/black.json

@@ -1,10 +1,12 @@
 {
     "prefixes": [
+        "lib/ansible/utils/collection_loader/__init__.py",
         "test/sanity/code-smell/black."
     ],
     "extensions": [
         ".py"
     ],
+    "split_targets": true,
     "error_code": "ansible-test",
     "output": "path-message"
 }

test/sanity/code-smell/black.py

@@ -13,12 +13,26 @@ def main() -> None:
     """Main program entry point."""
     paths = sys.argv[1:] or sys.stdin.read().splitlines()
 
-    env = os.environ.copy()
+    separator_idx = paths.index('--')
+    controller_paths = paths[:separator_idx]
+    target_paths = paths[separator_idx + 1 :]
 
-    controller_python_versions = env['ANSIBLE_TEST_CONTROLLER_PYTHON_VERSIONS'].split(',')
+    controller_python_versions = os.environ['ANSIBLE_TEST_CONTROLLER_PYTHON_VERSIONS'].split(',')
-    fix_mode = bool(int(env['ANSIBLE_TEST_FIX_MODE']))
+    remote_only_python_versions = os.environ['ANSIBLE_TEST_REMOTE_ONLY_PYTHON_VERSIONS'].split(',')
+    fix_mode = bool(int(os.environ['ANSIBLE_TEST_FIX_MODE']))
 
-    version_options = [('-t', f'py{version.replace(".", "")}') for version in controller_python_versions]
+    target_python_versions = remote_only_python_versions + controller_python_versions
+
+    black(controller_paths, controller_python_versions, fix_mode)
+    black(target_paths, target_python_versions, fix_mode)
+
+
+def black(paths: list[str], python_versions: list[str], fix_mode: bool) -> None:
+    """Run black on the specified paths."""
+    if not paths:
+        return
+
+    version_options = [('-t', f'py{version.replace(".", "")}') for version in python_versions]
 
     options = {
         '-m': 'black',
@@ -40,7 +54,7 @@ def main() -> None:
     cmd.extend(paths)
 
     try:
-        completed_process = subprocess.run(cmd, env=env, capture_output=True, check=True, text=True)
+        completed_process = subprocess.run(cmd, capture_output=True, check=True, text=True)
         stdout, stderr = completed_process.stdout, completed_process.stderr
 
         if stdout: