feat(cli): add --device-tag argument for per-device filtering

nac_test/cli/main.py

@@ -177,6 +177,16 @@ def version_callback(value: bool) -> None:
 ]
 
 
+DeviceTag = Annotated[
+    str | None,
+    typer.Option(
+        "--device-tag",
+        help="Only test devices whose 'tags' list in the data model contains this value.",
+        envvar="NAC_TEST_DEVICE_TAG",
+    ),
+]
+
+
 RenderOnly = Annotated[
     bool,
     typer.Option(
@@ -302,6 +312,7 @@ def main(
     tests: Tests = None,
     include: Include = None,
     exclude: Exclude = None,
+    device_tag: DeviceTag = None,
     render_only: RenderOnly = False,
     dry_run: DryRun = False,
     processes: Processes = None,
@@ -411,6 +422,7 @@ def main(
         tests_path=tests,
         include_tags=include,
         exclude_tags=exclude,
+        device_tag=device_tag,
         render_only=render_only,
         dry_run=dry_run,
         processes=processes,

nac_test/combined_orchestrator.py

@@ -80,6 +80,7 @@ def __init__(
         tests_path: Path | None = None,
         include_tags: list[str] | None = None,
         exclude_tags: list[str] | None = None,
+        device_tag: str | None = None,
         render_only: bool = False,
         dry_run: bool = False,
         max_parallel_devices: int | None = None,
@@ -127,6 +128,7 @@ def __init__(
         self.tests_path = tests_path
         self.include_tags = include_tags or []
         self.exclude_tags = exclude_tags or []
+        self.device_tag = device_tag
         self.render_only = render_only
         self.dry_run = dry_run
         self.processes = processes
@@ -224,6 +226,7 @@ def run_tests(self) -> CombinedResults:
                 loglevel=self.loglevel,
                 include_tags=self.include_tags,
                 exclude_tags=self.exclude_tags,
+                device_tag=self.device_tag,
             )
             if self.max_parallel_devices is not None:
                 pyats_orchestrator.max_parallel_devices = self.max_parallel_devices

nac_test/pyats_core/execution/device/device_executor.py

@@ -32,6 +32,7 @@ def __init__(
         base_output_dir: Path,
         merged_data_path: Path,
         custom_testbed_path: Path | None = None,
+        device_tag: str | None = None,
     ):
         """Initialize device executor.
 
@@ -43,6 +44,7 @@ def __init__(
             base_output_dir: Base output directory for test results
             merged_data_path: Path to the merged data model YAML file
             custom_testbed_path: Optional path to custom PyATS testbed YAML
+            device_tag: Only test devices whose 'tags' list contains this value
         """
         self.job_generator = job_generator
         self.subprocess_runner = subprocess_runner
@@ -51,6 +53,7 @@ def __init__(
         self.base_output_dir = base_output_dir
         self.merged_data_path = merged_data_path
         self.custom_testbed_path = custom_testbed_path
+        self.device_tag = device_tag
 
     async def run_device_job_with_semaphore(
         self,
@@ -108,17 +111,18 @@ async def run_device_job_with_semaphore(
                 # Set up environment for this device
                 # Always start with a copy of os.environ to preserve PATH and other variables
                 env = os.environ.copy()
-                env.update(
-                    {
-                        "HOSTNAME": hostname,
-                        "DEVICE_INFO": json.dumps(device),
-                        "MERGED_DATA_MODEL_TEST_VARIABLES_FILEPATH": str(
-                            self.merged_data_path
-                        ),
-                        "NAC_TEST_TYPE": "d2d",
-                        ENV_TEST_DIR: str(self.test_dir),
-                    }
-                )
+                env_updates: dict[str, str] = {
+                    "HOSTNAME": hostname,
+                    "DEVICE_INFO": json.dumps(device),
+                    "MERGED_DATA_MODEL_TEST_VARIABLES_FILEPATH": str(
+                        self.merged_data_path
+                    ),
+                    "NAC_TEST_TYPE": "d2d",
+                    ENV_TEST_DIR: str(self.test_dir),
+                }
+                if self.device_tag:
+                    env_updates["NAC_TEST_DEVICE_TAG"] = self.device_tag
+                env.update(env_updates)
 
                 # Track test status for this device.
                 #

nac_test/pyats_core/orchestrator.py

@@ -75,6 +75,7 @@ def __init__(
         loglevel: LogLevel = DEFAULT_LOGLEVEL,
         include_tags: list[str] | None = None,
         exclude_tags: list[str] | None = None,
+        device_tag: str | None = None,
     ):
         """Initialize the PyATS orchestrator.
 
@@ -91,6 +92,7 @@ def __init__(
             loglevel: Log level for PyATS output filtering
             include_tags: Tag patterns to include (Robot Framework syntax)
             exclude_tags: Tag patterns to exclude (Robot Framework syntax)
+            device_tag: Only test devices whose 'tags' list contains this value
         """
         self.data_paths = data_paths
         # Use absolute() rather than resolve() to preserve symlinks — resolve() would
@@ -112,6 +114,7 @@ def __init__(
         self.loglevel = loglevel
         self.include_tags = include_tags
         self.exclude_tags = exclude_tags
+        self.device_tag = device_tag
 
         # Track test status by type for combined summary
         self.api_test_status: dict[str, dict[str, Any]] = {}
@@ -278,6 +281,9 @@ async def _execute_api_tests_standard(self, test_files: list[Path]) -> Path | No
             env["NAC_TEST_TYPE"] = "api"
             # Pass test_dir so plugin can compute relative test names
             env[ENV_TEST_DIR] = str(self.test_dir)
+            # Pass device tag filter to subprocess for per-device scoping
+            if self.device_tag:
+                env["NAC_TEST_DEVICE_TAG"] = self.device_tag
 
             # Execute and return the archive path
             assert self.subprocess_runner is not None  # Should be initialized by now
@@ -393,6 +399,7 @@ async def _execute_device_tests_with_broker(
                 self.base_output_dir,
                 self.merged_data_path,
                 self.custom_testbed_path,
+                device_tag=self.device_tag,
             )
 
         # Use a local narrowed variable to satisfy mypy
@@ -659,6 +666,12 @@ async def _run_tests_async(self) -> PyATSResults:
             tasks.append(self._execute_api_tests_standard(api_tests))
 
         if d2d_tests:
+            # Set device tag in environment so the resolver can filter during discovery
+            if self.device_tag:
+                os.environ["NAC_TEST_DEVICE_TAG"] = self.device_tag
+            elif "NAC_TEST_DEVICE_TAG" in os.environ:
+                del os.environ["NAC_TEST_DEVICE_TAG"]
+
             # Get device inventory for D2D tests
             devices = self.device_inventory_discovery.get_device_inventory(d2d_tests)
 

tests/unit/test_combined_orchestrator_controller.py

@@ -239,6 +239,7 @@ def test_combined_orchestrator_passes_controller_to_pyats(
                     loglevel=DEFAULT_LOGLEVEL,
                     include_tags=[],
                     exclude_tags=[],
+                    device_tag=None,
                 )
 
                 # Verify run_tests was called on the instance
@@ -404,6 +405,7 @@ def test_combined_orchestrator_production_mode_passes_controller(
                     loglevel=DEFAULT_LOGLEVEL,
                     include_tags=[],
                     exclude_tags=[],
+                    device_tag=None,
                 )
 
                 # Verify run_tests was called on the instance