Add unit tests for ROSA cluster tagging username resolution

Cover CloudTrail fallback lookups, automation-user exclusion, strategy
ordering, and ambiguous cluster role matching introduced in PR #1007.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: pragya811

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

@@ -0,0 +1,85 @@
+from datetime import datetime, timezone
+from unittest.mock import Mock, patch
+
+from cloud_governance.policy.policy_operations.aws.tag_cluster.tag_cluster_operations import TagClusterOperations
+
+
+def _make_tag_cluster_ops(**overrides):
+    """Build TagClusterOperations without calling __init__ (avoids real AWS calls)."""
+    ops = TagClusterOperations.__new__(TagClusterOperations)
+    ops.iam_users = overrides.get('iam_users', ['pragchau', 'cloud-governance-delete-user'])
+    ops._automation_user = overrides.get('_automation_user', 'cloud-governance-user')
+    ops.get_user_name_from_name_tag = Mock(return_value=overrides.get('name_tag_user'))
+    ops._get_username_from_instance_id_and_time = Mock(
+        return_value=overrides.get('instance_ct_user'))
+    ops.cloudtrail = Mock()
+    ops.cloudtrail.get_username_from_cluster_role = Mock(
+        return_value=overrides.get('cluster_role_user', ''))
+    ops._regional_cloudtrail = Mock()
+    ops._regional_cloudtrail.get_username_from_resource_events = Mock(
+        return_value=overrides.get('resource_events_user', ''))
+    return ops
+
+
+class TestTagClusterOperationsGetUsername:
+    LAUNCH_TIME = datetime(2026, 6, 30, 14, 6, 46, tzinfo=timezone.utc)
+    RESOURCE_ID = 'i-0123456789abcdef0'
+    CLUSTER_ID = 'z2r7p1f3o6m2h3y-t5bx8'
+
+    def test_returns_autoscaling_without_iam_validation(self):
+        ops = _make_tag_cluster_ops(instance_ct_user='AutoScaling')
+        result = ops.get_username(
+            start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID,
+            resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID)
+        assert result == 'AutoScaling'
+
+    def test_skips_automation_user_from_instance_cloudtrail(self):
+        ops = _make_tag_cluster_ops(
+            instance_ct_user='cloud-governance-user',
+            cluster_role_user='pragchau')
+        result = ops.get_username(
+            start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID,
+            resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID)
+        assert result == 'pragchau'
+        ops.cloudtrail.get_username_from_cluster_role.assert_called_once()
+
+    def test_cluster_role_resolved_before_resource_events_fallback(self):
+        ops = _make_tag_cluster_ops(
+            instance_ct_user='178000000000',
+            cluster_role_user='pragchau',
+            resource_events_user='cloud-governance-delete-user')
+        result = ops.get_username(
+            start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID,
+            resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID)
+        assert result == 'pragchau'
+        ops._regional_cloudtrail.get_username_from_resource_events.assert_not_called()
+
+    def test_resource_events_fallback_excludes_automation_user(self):
+        ops = _make_tag_cluster_ops(
+            instance_ct_user='178000000000',
+            cluster_role_user='',
+            resource_events_user='pragchau')
+        result = ops.get_username(
+            start_time=self.LAUNCH_TIME, resource_id=self.RESOURCE_ID,
+            resource_type='AWS::EC2::Instance', tags=[], cluster_id=self.CLUSTER_ID)
+        assert result == 'pragchau'
+        call_kwargs = ops._regional_cloudtrail.get_username_from_resource_events.call_args.kwargs
+        assert call_kwargs['exclude_users'] == {'cloud-governance-user'}
+
+
+class TestTagClusterOperationsClusterInstanceFallback:
+    def test_get_username_from_cluster_instances_returns_tagged_peer(self):
+        ops = TagClusterOperations.__new__(TagClusterOperations)
+        ops.cluster_prefix = ['kubernetes.io/cluster/']
+        ops.iam_users = ['pragchau']
+        ops.ec2_operations = Mock()
+        ops.ec2_operations.get_tag_value_from_tags = Mock(return_value='pragchau')
+        resources = [[{
+            'Tags': [
+                {'Key': 'kubernetes.io/cluster/z2r7p1f3o6m2h3y-t5bx8', 'Value': 'owned'},
+                {'Key': 'User', 'Value': 'pragchau'},
+            ]
+        }]]
+        result = ops.get_username_from_cluster_instances(
+            resources=resources, cluster_name='z2r7p1f3o6m2h3y-t5bx8')
+        assert result == 'pragchau'

tests/unittest/cloud_governance/aws/tag_non_cluster/test_non_cluster_operations.py

@@ -0,0 +1,47 @@
+from datetime import datetime, timezone
+from unittest.mock import Mock
+
+from cloud_governance.policy.policy_operations.aws.tag_non_cluster.non_cluster_operations import NonClusterOperations
+
+
+def _make_non_cluster_ops(**overrides):
+    ops = NonClusterOperations.__new__(NonClusterOperations)
+    ops.iam_users = overrides.get('iam_users', ['pragchau', 'cloud-governance-delete-user'])
+    ops._automation_user = overrides.get('_automation_user', 'cloud-governance-user')
+    ops.get_user_name_from_name_tag = Mock(return_value=overrides.get('name_tag_user'))
+    ops._get_username_from_cloudtrail = Mock(return_value=overrides.get('cloudtrail_user'))
+    ops.cloudtrail = Mock()
+    ops.cloudtrail.get_username_from_resource_events = Mock(
+        return_value=overrides.get('resource_events_user', ''))
+    return ops
+
+
+class TestNonClusterOperationsGetUsername:
+    START_TIME = datetime(2026, 6, 30, 14, 0, 0, tzinfo=timezone.utc)
+
+    def test_excludes_automation_user_from_primary_cloudtrail(self):
+        ops = _make_non_cluster_ops(cloudtrail_user='cloud-governance-user')
+        result = ops.get_username(
+            start_time=self.START_TIME, resource_id='vol-abc123',
+            resource_type='AWS::EC2::Volume', tags=[])
+        assert result == ''
+        ops.cloudtrail.get_username_from_resource_events.assert_called_once()
+
+    def test_returns_valid_user_from_primary_cloudtrail(self):
+        ops = _make_non_cluster_ops(cloudtrail_user='pragchau')
+        result = ops.get_username(
+            start_time=self.START_TIME, resource_id='vol-abc123',
+            resource_type='AWS::EC2::Volume', tags=[])
+        assert result == 'pragchau'
+        ops.cloudtrail.get_username_from_resource_events.assert_not_called()
+
+    def test_resource_events_fallback_excludes_automation_user(self):
+        ops = _make_non_cluster_ops(
+            cloudtrail_user='',
+            resource_events_user='pragchau')
+        result = ops.get_username(
+            start_time=self.START_TIME, resource_id='vol-abc123',
+            resource_type='AWS::EC2::Volume', tags=[])
+        assert result == 'pragchau'
+        call_kwargs = ops.cloudtrail.get_username_from_resource_events.call_args.kwargs
+        assert call_kwargs['exclude_users'] == {'cloud-governance-user'}

tests/unittest/cloud_governance/common/clouds/aws/cloudtrail/test_cloudtrail_operations.py

@@ -1,4 +1,5 @@
 import json
+from datetime import datetime, timezone
 from unittest.mock import Mock, patch
 
 import pytest
@@ -349,3 +350,126 @@ def test_arn_parsing_different_session_formats(self, mock_boto3):
         for arn, expected_username in test_cases:
             session_name = arn.split('/')[-1]
             assert session_name == expected_username, f"Failed to parse {arn}"
+
+
+class TestCloudTrailOperationsRosaTagging:
+    """Tests for ROSA cluster ownership resolution added in PR #1007."""
+
+    LAUNCH_TIME = datetime(2026, 6, 30, 14, 6, 46, tzinfo=timezone.utc)
+    IAM_USERS = ['pragchau', 'cloud-governance-delete-user']
+
+    @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3')
+    def test_get_username_from_resource_events_skips_excluded_users(self, mock_boto3):
+        cloudtrail_ops = CloudTrailOperations(region_name='us-east-1')
+        cloudtrail_ops.get_full_responses = Mock(return_value=[
+            {
+                'EventName': 'CreateTags',
+                'Username': 'cloud-governance-delete-user',
+                'CloudTrailEvent': '{}',
+            },
+            {
+                'EventName': 'CreateTags',
+                'Username': 'pragchau',
+                'CloudTrailEvent': '{}',
+            },
+        ])
+        result = cloudtrail_ops.get_username_from_resource_events(
+            resource_id='i-0123456789abcdef0',
+            iam_users=self.IAM_USERS,
+            start_time=self.LAUNCH_TIME,
+            exclude_users={'cloud-governance-delete-user'})
+        assert result == 'pragchau'
+
+    @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3')
+    def test_get_username_from_cluster_role_direct_match(self, mock_boto3):
+        mock_iam = Mock()
+        mock_paginator = Mock()
+        mock_paginator.paginate.return_value = [{
+            'Roles': [{
+                'RoleName': 'z2r7p1f3o6m2h3y-t5bx8-master-role',
+                'Arn': 'arn:aws:iam::123456789012:role/z2r7p1f3o6m2h3y-t5bx8-master-role',
+                'CreateDate': self.LAUNCH_TIME,
+            }]
+        }]
+        mock_iam.get_paginator.return_value = mock_paginator
+        mock_boto3.client.side_effect = lambda service, **kwargs: {
+            'cloudtrail': Mock(),
+            'iam': mock_iam,
+        }[service]
+
+        cloudtrail_ops = CloudTrailOperations(region_name='us-east-1')
+        cloudtrail_ops._CloudTrailOperations__get_username_from_role_cloudtrail = Mock(
+            return_value='pragchau')
+
+        result = cloudtrail_ops.get_username_from_cluster_role(
+            cluster_id='z2r7p1f3o6m2h3y-t5bx8',
+            iam_users=self.IAM_USERS,
+            launch_time=self.LAUNCH_TIME)
+        assert result == 'pragchau'
+
+    @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3')
+    def test_get_username_from_cluster_role_ambiguous_temporal_match_returns_empty(
+            self, mock_boto3):
+        role_create = datetime(2026, 6, 30, 13, 0, 0, tzinfo=timezone.utc)
+        mock_iam = Mock()
+        mock_paginator = Mock()
+        mock_paginator.paginate.return_value = [{
+            'Roles': [
+                {
+                    'RoleName': 'cluster-a-openshift-machine-api',
+                    'Arn': 'arn:aws:iam::123456789012:role/cluster-a-openshift-machine-api',
+                    'CreateDate': role_create,
+                },
+                {
+                    'RoleName': 'cluster-b-openshift-machine-api',
+                    'Arn': 'arn:aws:iam::123456789012:role/cluster-b-openshift-machine-api',
+                    'CreateDate': role_create,
+                },
+            ]
+        }]
+        mock_iam.get_paginator.return_value = mock_paginator
+        mock_boto3.client.side_effect = lambda service, **kwargs: {
+            'cloudtrail': Mock(),
+            'iam': mock_iam,
+        }[service]
+
+        cloudtrail_ops = CloudTrailOperations(region_name='us-east-1')
+        cloudtrail_ops._CloudTrailOperations__get_username_from_role_cloudtrail = Mock(
+            side_effect=['pragchau', 'otheruser'])
+        cloudtrail_ops._CloudTrailOperations__get_username_from_create_role_events = Mock(
+            return_value='')
+
+        result = cloudtrail_ops.get_username_from_cluster_role(
+            cluster_id='unknown-infra-id',
+            iam_users=self.IAM_USERS + ['otheruser'],
+            launch_time=self.LAUNCH_TIME)
+        assert result == ''
+        cloudtrail_ops._CloudTrailOperations__get_username_from_create_role_events.assert_called_once()
+
+    @patch('cloud_governance.common.clouds.aws.cloudtrail.cloudtrail_operations.boto3')
+    def test_get_username_from_create_role_events_paginates_global_cloudtrail(
+            self, mock_boto3):
+        mock_global_ct = Mock()
+        mock_global_ct.lookup_events.side_effect = [
+            {'Events': [], 'NextToken': 'page2'},
+            {'Events': [{
+                'EventName': 'CreateRole',
+                'Username': 'pragchau',
+                'Resources': [{
+                    'ResourceType': 'AWS::IAM::Role',
+                    'ResourceName': 'rosa-test-openshift-machine-api',
+                }],
+                'CloudTrailEvent': '{}',
+            }]},
+        ]
+        mock_boto3.client.return_value = Mock()
+
+        cloudtrail_ops = CloudTrailOperations(region_name='us-west-2')
+        cloudtrail_ops._CloudTrailOperations__global_cloudtrail = mock_global_ct
+
+        start = datetime(2026, 6, 30, 8, 0, 0, tzinfo=timezone.utc)
+        end = datetime(2026, 6, 30, 14, 0, 0, tzinfo=timezone.utc)
+        result = cloudtrail_ops._CloudTrailOperations__get_username_from_create_role_events(
+            start_time=start, end_time=end, iam_users=self.IAM_USERS)
+        assert result == 'pragchau'
+        assert mock_global_ct.lookup_events.call_count == 2