[cli] Include queue and compute-resource tags in runInstance dry run

cli/src/pcluster/config/common.py

@@ -311,6 +311,20 @@ def __init__(self, key: str = None, value: str = None):
         self.value = Resource.init_param(value)
 
 
+def merge_tags(*tag_lists):
+    """
+    Merge lists of Tag objects, deduplicating by key.
+
+    Later lists take precedence over earlier ones on duplicate keys, mirroring the tag precedence
+    applied at launch time (cluster < queue < compute resource). ``None`` lists are skipped.
+    """
+    merged = {}
+    for tags in tag_lists:
+        for tag in tags or []:
+            merged[tag.key] = tag
+    return list(merged.values())
+
+
 class AdditionalIamPolicy(Resource):
     """Represent the Additional IAM Policy configuration."""
 

cli/src/pcluster/templates/queues_stack.py

@@ -8,7 +8,7 @@
 
 from pcluster.aws.aws_api import AWSApi
 from pcluster.config.cluster_config import SlurmClusterConfig, SlurmComputeResource, SlurmQueue
-from pcluster.config.common import DefaultUserHomeType, SharedStorageType
+from pcluster.config.common import DefaultUserHomeType, SharedStorageType, merge_tags
 from pcluster.constants import (
     DEFAULT_EPHEMERAL_DIR,
     NODE_BOOTSTRAP_TIMEOUT,
@@ -24,7 +24,6 @@
     create_hash_suffix,
     dict_to_cfn_tags,
     get_common_user_data_env,
-    get_custom_tags,
     get_default_instance_tags,
     get_default_volume_tags,
     get_queue_security_groups_full,
@@ -160,10 +159,8 @@ def _add_launch_templates(self):
 
     def _get_custom_compute_resource_tags(self, queue_config, compute_resource_config):
         """Compute resource tags and Queue Tags value on Cluster level tags if there are duplicated keys."""
-        tags = get_custom_tags(self._config, raw_dict=True)
-        queue_tags = get_custom_tags(queue_config, raw_dict=True)
-        compute_resource_tags = get_custom_tags(compute_resource_config, raw_dict=True)
-        return dict_to_cfn_tags({**tags, **queue_tags, **compute_resource_tags})
+        merged = merge_tags(self._config.get_tags(), queue_config.get_tags(), compute_resource_config.get_tags())
+        return dict_to_cfn_tags({tag.key: tag.value for tag in merged})
 
     def _add_compute_resource_launch_template(
         self,

cli/src/pcluster/validators/cluster_validators.py

@@ -20,7 +20,7 @@
 from pcluster.aws.aws_resources import InstanceTypeInfo
 from pcluster.aws.common import AWSClientError
 from pcluster.cli.commands.dcv_util import get_supported_dcv_os
-from pcluster.config.common import CapacityType, SharedStorageType
+from pcluster.config.common import CapacityType, SharedStorageType, merge_tags
 from pcluster.constants import (
     CIDR_ALL_IPS,
     DELETE_POLICY,
@@ -1354,6 +1354,9 @@ def _validate(self, queue, root_volume_device_name, ami_id, tags, imds_support):
             )
 
             placement_group_name = compute_resource_placement_group.assignment
+            # Merge cluster, queue, and compute resource tags to mirror the tags applied at launch time,
+            # so the dry run reflects the tags an actual RunInstances would carry.
+            merged_tags = merge_tags(tags, queue.get_tags(), dry_run_compute_resource.get_tags())
             # For SlurmFlexibleComputeResource test only the first InstanceType through a RunInstances
             self._test_compute_resource(
                 queue=queue,
@@ -1364,7 +1367,7 @@ def _validate(self, queue, root_volume_device_name, ami_id, tags, imds_support):
                 subnet_id=queue_subnet_id,
                 security_groups_ids=queue_security_groups,
                 placement_group={"GroupName": placement_group_name} if placement_group_name else {},
-                tags=tags,
+                tags=merged_tags,
                 imds_support=imds_support,
             )
         except Exception as e:

cli/tests/pcluster/validators/test_cluster_validators.py

@@ -31,7 +31,7 @@
     SlurmSettings,
     Tag,
 )
-from pcluster.config.common import CapacityType
+from pcluster.config.common import CapacityType, merge_tags
 from pcluster.constants import PCLUSTER_NAME_MAX_LENGTH, PCLUSTER_NAME_MAX_LENGTH_SLURM_ACCOUNTING
 from pcluster.validators.cluster_validators import (
     FSX_MESSAGES,
@@ -1938,6 +1938,38 @@ def test_generate_tag_specifications(input_tags):
     assert_that(_LaunchTemplateValidator._generate_tag_specifications(input_tags)).is_equal_to(expected_output_tags)
 
 
+@pytest.mark.parametrize(
+    "tag_lists, expected",
+    [
+        # No tags at all
+        ([], []),
+        ([None, None], []),
+        # Single source
+        ([[{"key": "A", "value": "1"}]], [("A", "1")]),
+        # Distinct keys across sources are all kept
+        (
+            [[{"key": "A", "value": "1"}], [{"key": "B", "value": "2"}], [{"key": "C", "value": "3"}]],
+            [("A", "1"), ("B", "2"), ("C", "3")],
+        ),
+        # Duplicate keys: later sources (queue, compute resource) override earlier (cluster)
+        (
+            [[{"key": "A", "value": "cluster"}], [{"key": "A", "value": "queue"}], [{"key": "A", "value": "cr"}]],
+            [("A", "cr")],
+        ),
+        # None sources are skipped without affecting precedence
+        (
+            [[{"key": "A", "value": "cluster"}], None, [{"key": "A", "value": "cr"}]],
+            [("A", "cr")],
+        ),
+    ],
+)
+def test_merge_tags(tag_lists, expected):
+    """Verify tag merging deduplicates by key with later sources taking precedence."""
+    tag_lists = [[Tag(tag["key"], tag["value"]) for tag in tags] if tags is not None else None for tags in tag_lists]
+    merged = merge_tags(*tag_lists)
+    assert_that([(tag.key, tag.value) for tag in merged]).is_equal_to(expected)
+
+
 def get_network_card_list(index_list):
     return [NetworkCardInfo({"NetworkCardIndex": index}) for index in index_list]