'tag_cluster' policy applying kubernetes.io/cluster/<name>=owned to Hypershift SG’s (#976)

* Initial code fix

* Move constant to environment_variables.py

* Reuse existing env variable

* Review comments

* Review comments

* Add Unittests

Author: pragya811

cloud_governance/policy/policy_operations/aws/tag_cluster/tag_cluster_resouces.py

@@ -2,6 +2,7 @@
 
 from cloud_governance.common.clouds.aws.utils.common_methods import get_tag_value_from_tags
 from cloud_governance.common.logger.init_logger import logger
+from cloud_governance.main.environment_variables import environment_variables
 from cloud_governance.policy.policy_operations.aws.tag_cluster.tag_cluster_operations import TagClusterOperations
 
 from cloud_governance.policy.policy_operations.aws.tag_non_cluster.tag_non_cluster_resources import \
@@ -108,9 +109,15 @@ def __get_cluster_tags_by_instance_cluster(self, cluster_name: str):
                         for tag in item.get('Tags'):
                             if any(prefix in tag.get('Key', '') for prefix in self.cluster_prefix):
                                 if cluster_name in tag.get('Key'):
+                                    # Propagation-blocked prefixes derived from CLUSTER_PREFIX (e.g. kubernetes.io/, sigs.k8s.io/)
+                                    cluster_prefix = environment_variables.environment_variables_dict['CLUSTER_PREFIX']
+                                    no_propagate_prefixes = tuple(p.split('/', 1)[0] + '/' for p in cluster_prefix)
                                     i_tags = [instance_tag for instance_tag in item.get('Tags') if
-                                              instance_tag.get('Key') != 'Name']
-                                    return [i_tag for i_tag in i_tags if i_tag.get('Key') != cluster_name]
+                                              instance_tag.get('Key') != 'Name' and
+                                              not any((instance_tag.get('Key') or '').startswith(prefix)
+                                                      for prefix in self.cluster_prefix) and
+                                              not (instance_tag.get('Key') or '').startswith(no_propagate_prefixes)]
+                                    return i_tags
         return []
 
     def get_date_from_date(self, date_time: datetime):

tests/unittest/cloud_governance/aws/tag_cluster/test_tag_cluster_resources.py

@@ -1,6 +1,7 @@
 # TEST DRY RUN: mandatory_tags = None
 import json
 import os
+from unittest.mock import patch
 
 import pytest
 from moto import mock_ec2, mock_cloudtrail, mock_iam, mock_s3, mock_elb, mock_elbv2
@@ -236,3 +237,108 @@ def test_cluster_ec2():
     tag_resources = TagClusterResources(cluster_prefix=cluster_prefix, cluster_name=cluster_name,
                                         input_tags=mandatory_tags, region='us-east-2')
     assert len(tag_resources.cluster_instance()) == 3
+
+
+# --- Hypershift / tag_cluster propagation (PR #976): do not propagate kubernetes.io/ or sigs.k8s.io/ tags ---
+
+CLUSTER_STAMP_KEY = 'kubernetes.io/cluster/hyper2-unittest'
+
+
+def test_get_cluster_tags_for_propagation_excludes_cluster_and_k8s_sigs_tags():
+    """
+    Tags returned for propagation must not include kubernetes.io/cluster, other kubernetes.io/*,
+    sigs.k8s.io/*, or Name — only governance-style tags (e.g. User) should propagate.
+    """
+    with mock_ec2(), mock_iam(), mock_cloudtrail(), mock_s3(), mock_elb(), mock_elbv2():
+        tcr = TagClusterResources(
+            cluster_prefix=cluster_prefix,
+            cluster_name=cluster_name,
+            input_tags=mandatory_tags,
+            region='us-east-2',
+        )
+        fake_instance = [{
+            'InstanceId': 'i-hyper2test',
+            'Tags': [
+                {'Key': CLUSTER_STAMP_KEY, 'Value': 'owned'},
+                {
+                    'Key': 'sigs.k8s.io/cluster-api-provider-aws/cluster/hyper2-unittest',
+                    'Value': 'owned',
+                },
+                {'Key': 'kubernetes.io/role/worker', 'Value': 'true'},
+                {'Key': 'User', 'Value': 'alice'},
+                {'Key': 'Manager', 'Value': 'bob'},
+                {'Key': 'Name', 'Value': 'hypershift-node-1'},
+            ],
+        }]
+        with patch.object(tcr, '_get_instances_data', return_value=[fake_instance]):
+            result = tcr._TagClusterResources__get_cluster_tags_by_instance_cluster(CLUSTER_STAMP_KEY)
+
+    assert result == [
+        {'Key': 'User', 'Value': 'alice'},
+        {'Key': 'Manager', 'Value': 'bob'},
+    ]
+    assert not any(t['Key'].startswith('kubernetes.io/') for t in result)
+    assert not any(t['Key'].startswith('sigs.k8s.io/') for t in result)
+    assert not any(t['Key'] == 'Name' for t in result)
+
+
+def test_get_cluster_tags_for_propagation_empty_when_only_cluster_system_tags():
+    """If the instance has only cluster stamp and k8s/sigs system tags (plus Name), nothing should propagate."""
+    with mock_ec2(), mock_iam(), mock_cloudtrail(), mock_s3(), mock_elb(), mock_elbv2():
+        tcr = TagClusterResources(
+            cluster_prefix=cluster_prefix,
+            cluster_name=cluster_name,
+            input_tags=mandatory_tags,
+            region='us-east-2',
+        )
+        fake_instance = [{
+            'InstanceId': 'i-onlysys',
+            'Tags': [
+                {'Key': CLUSTER_STAMP_KEY, 'Value': 'owned'},
+                {
+                    'Key': 'sigs.k8s.io/cluster-api-provider-aws/cluster/hyper2-unittest',
+                    'Value': 'owned',
+                },
+                {'Key': 'kubernetes.io/role/master', 'Value': 'true'},
+                {'Key': 'Name', 'Value': 'cp-0'},
+            ],
+        }]
+        with patch.object(tcr, '_get_instances_data', return_value=[fake_instance]):
+            result = tcr._TagClusterResources__get_cluster_tags_by_instance_cluster(CLUSTER_STAMP_KEY)
+
+    assert result == []
+
+
+def test_get_cluster_tags_for_propagation_uses_cluster_prefix_from_environment_variables():
+    """
+    no_propagate_prefixes must follow CLUSTER_PREFIX in environment_variables (same derivation as production).
+    """
+    from cloud_governance.main.environment_variables import environment_variables
+
+    custom_prefixes = ['api.openshift.com/cluster', 'kubernetes.io/cluster']
+    original = environment_variables.environment_variables_dict['CLUSTER_PREFIX']
+    try:
+        environment_variables.environment_variables_dict['CLUSTER_PREFIX'] = custom_prefixes
+        with mock_ec2(), mock_iam(), mock_cloudtrail(), mock_s3(), mock_elb(), mock_elbv2():
+            tcr = TagClusterResources(
+                cluster_prefix=custom_prefixes,
+                cluster_name=cluster_name,
+                input_tags=mandatory_tags,
+                region='us-east-2',
+            )
+            stamp = 'api.openshift.com/cluster/hyper2-unittest'
+            fake_instance = [{
+                'InstanceId': 'i-custom',
+                'Tags': [
+                    {'Key': stamp, 'Value': 'owned'},
+                    {'Key': 'api.openshift.com/something-else', 'Value': 'x'},
+                    {'Key': 'User', 'Value': 'carol'},
+                ],
+            }]
+            with patch.object(tcr, '_get_instances_data', return_value=[fake_instance]):
+                result = tcr._TagClusterResources__get_cluster_tags_by_instance_cluster(stamp)
+        # api.openshift.com/ and kubernetes.io/ blocked; User kept
+        assert result == [{'Key': 'User', 'Value': 'carol'}]
+        assert not any(t['Key'].startswith('api.openshift.com/') for t in result)
+    finally:
+        environment_variables.environment_variables_dict['CLUSTER_PREFIX'] = original