From a32d32354af13ef937ae6b7223a26752049fa788 Mon Sep 17 00:00:00 2001 From: Vichym Date: Tue, 16 Dec 2025 16:35:25 -0800 Subject: [PATCH 1/7] add integration test for LMI functions --- .../test_function_with_capacity_provider.py | 65 +++++++++++ integration/config/service_names.py | 1 + .../combination/function_lmi_custom.json | 18 +++ .../combination/function_lmi_default.json | 18 +++ .../combination/function_lmi_custom.yaml | 87 ++++++++++++++ .../combination/function_lmi_default.yaml | 31 +++++ integration/setup/companion-stack.yaml | 109 ++++++++++++++++++ 7 files changed, 329 insertions(+) create mode 100644 integration/combination/test_function_with_capacity_provider.py create mode 100644 integration/resources/expected/combination/function_lmi_custom.json create mode 100644 integration/resources/expected/combination/function_lmi_default.json create mode 100644 integration/resources/templates/combination/function_lmi_custom.yaml create mode 100644 integration/resources/templates/combination/function_lmi_default.yaml diff --git a/integration/combination/test_function_with_capacity_provider.py b/integration/combination/test_function_with_capacity_provider.py new file mode 100644 index 0000000000..928286ec88 --- /dev/null +++ b/integration/combination/test_function_with_capacity_provider.py @@ -0,0 +1,65 @@ +from unittest.case import skipIf + +import pytest + +from integration.config.service_names import LAMBDA_MANAGED_INSTANCES +from integration.helpers.base_test import BaseTest +from integration.helpers.resource import current_region_does_not_support + + +@skipIf( + current_region_does_not_support([LAMBDA_MANAGED_INSTANCES]), + "LambdaManagedInstance is not supported in this testing region", +) +class TestFunctionWithCapacityProvider(BaseTest): + @pytest.fixture(autouse=True) + def companion_stack_outputs(self, get_companion_stack_outputs): + self.companion_stack_outputs = get_companion_stack_outputs + + def test_function_with_capacity_provider_custom_role(self): + """Test Lambda function with CapacityProviderConfig using custom operator role.""" + # Phase 1: Prepare parameters from companion stack + parameters = [ + self.generate_parameter("SubnetId", self.companion_stack_outputs["LMISubnetId"]), + self.generate_parameter("SecurityGroup", self.companion_stack_outputs["LMISecurityGroupId"]), + self.generate_parameter("KMSKeyArn", self.companion_stack_outputs["LMIKMSKeyArn"]), + ] + + # Phase 2: Deploy and verify against expected JSON + self.create_and_verify_stack("combination/function_lmi_custom", parameters) + + # Phase 3: Verify resource counts + lambda_resources = self.get_stack_resources("AWS::Lambda::Function") + self.assertEqual(len(lambda_resources), 1, "Should create exactly one Lambda function") + + capacity_provider_resources = self.get_stack_resources("AWS::Lambda::CapacityProvider") + self.assertEqual(len(capacity_provider_resources), 1, "Should create exactly one CapacityProvider") + + iam_role_resources = self.get_stack_resources("AWS::IAM::Role") + self.assertEqual(len(iam_role_resources), 2, "Should create exactly two IAM roles") + + def test_function_with_capacity_provider_default_role(self): + """Test Lambda function with CapacityProviderConfig using default operator role. + + Note: This test is skipped until 12/01/2024 because the managed policy required for + the default operator role is not available until then. + """ + # Phase 1: Prepare parameters from companion stack + parameters = [ + self.generate_parameter("SubnetId", self.companion_stack_outputs["LMISubnetId"]), + self.generate_parameter("SecurityGroup", self.companion_stack_outputs["LMISecurityGroupId"]), + self.generate_parameter("KMSKeyArn", self.companion_stack_outputs["LMIKMSKeyArn"]), + ] + + # Phase 2: Deploy and verify against expected JSON + self.create_and_verify_stack("combination/function_lmi_default", parameters) + + # Phase 3: Verify resource counts + lambda_resources = self.get_stack_resources("AWS::Lambda::Function") + self.assertEqual(len(lambda_resources), 1, "Should create exactly one Lambda function") + + capacity_provider_resources = self.get_stack_resources("AWS::Lambda::CapacityProvider") + self.assertEqual(len(capacity_provider_resources), 1, "Should create exactly one CapacityProvider") + + iam_role_resources = self.get_stack_resources("AWS::IAM::Role") + self.assertEqual(len(iam_role_resources), 2, "Should create exactly two IAM roles") diff --git a/integration/config/service_names.py b/integration/config/service_names.py index b4a2c26261..4de385e0ea 100644 --- a/integration/config/service_names.py +++ b/integration/config/service_names.py @@ -30,6 +30,7 @@ STATE_MACHINE_CWE_CWS = "StateMachineCweCws" STATE_MACHINE_WITH_APIS = "StateMachineWithApis" LAMBDA_URL = "LambdaUrl" +LAMBDA_MANAGED_INSTANCES = "LambdaManagedInstances" LAMBDA_ENV_VARS = "LambdaEnvVars" EVENT_INVOKE_CONFIG = "EventInvokeConfig" API_KEY = "ApiKey" diff --git a/integration/resources/expected/combination/function_lmi_custom.json b/integration/resources/expected/combination/function_lmi_custom.json new file mode 100644 index 0000000000..52c2c0979f --- /dev/null +++ b/integration/resources/expected/combination/function_lmi_custom.json @@ -0,0 +1,18 @@ +[ + { + "LogicalResourceId": "MyCapacityProvider", + "ResourceType": "AWS::Lambda::CapacityProvider" + }, + { + "LogicalResourceId": "MyCapacityProviderCustomRole", + "ResourceType": "AWS::IAM::Role" + }, + { + "LogicalResourceId": "MyFunction", + "ResourceType": "AWS::Lambda::Function" + }, + { + "LogicalResourceId": "MyFunctionRole", + "ResourceType": "AWS::IAM::Role" + } +] diff --git a/integration/resources/expected/combination/function_lmi_default.json b/integration/resources/expected/combination/function_lmi_default.json new file mode 100644 index 0000000000..c71af2dc91 --- /dev/null +++ b/integration/resources/expected/combination/function_lmi_default.json @@ -0,0 +1,18 @@ +[ + { + "LogicalResourceId": "MyCapacityProvider", + "ResourceType": "AWS::Lambda::CapacityProvider" + }, + { + "LogicalResourceId": "MyCapacityProviderOperatorRole", + "ResourceType": "AWS::IAM::Role" + }, + { + "LogicalResourceId": "MyFunction", + "ResourceType": "AWS::Lambda::Function" + }, + { + "LogicalResourceId": "MyFunctionRole", + "ResourceType": "AWS::IAM::Role" + } +] diff --git a/integration/resources/templates/combination/function_lmi_custom.yaml b/integration/resources/templates/combination/function_lmi_custom.yaml new file mode 100644 index 0000000000..dbbf668dc3 --- /dev/null +++ b/integration/resources/templates/combination/function_lmi_custom.yaml @@ -0,0 +1,87 @@ +Parameters: + SubnetId: + Type: String + SecurityGroup: + Type: String + KMSKeyArn: + Type: String + +Resources: + MyCapacityProviderCustomRole: + Type: AWS::IAM::Role + Properties: + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Principal: + Service: lambda.amazonaws.com + Action: sts:AssumeRole + Policies: + - PolicyName: CapacityProviderOperatorRolePolicy + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: + - ec2:TerminateInstances + - ec2:AttachNetworkInterface + - ec2:RunInstances + - ec2:CreateTags + Resource: + - !Sub arn:${AWS::Partition}:ec2:*:*:instance/* + - !Sub arn:${AWS::Partition}:ec2:*:*:network-interface/* + - !Sub arn:${AWS::Partition}:ec2:*:*:volume/* + Condition: + StringEquals: + ec2:ManagedResourceOperator: scaler.lambda.amazonaws.com + - Effect: Allow + Action: + - ec2:DescribeAvailabilityZones + - ec2:DescribeCapacityReservations + - ec2:DescribeInstanceTypes + - ec2:DescribeInstanceTypeOfferings + - ec2:DescribeSecurityGroups + - ec2:DescribeSubnets + - ec2:DescribeInstances + - ec2:DescribeInstanceStatus + Resource: '*' + - Effect: Allow + Action: + - ec2:RunInstances + - ec2:CreateNetworkInterface + Resource: + - !Sub arn:${AWS::Partition}:ec2:*:*:subnet/* + - !Sub arn:${AWS::Partition}:ec2:*:*:security-group/* + - Effect: Allow + Action: + - ec2:RunInstances + Resource: + - !Sub arn:${AWS::Partition}:ec2:*:*:image/* + Condition: + Bool: + ec2:Public: 'true' + + MyCapacityProvider: + Type: AWS::Serverless::CapacityProvider + Properties: + CapacityProviderName: !Sub "${AWS::StackName}-cp" + VpcConfig: + SubnetIds: + - !Ref SubnetId + SecurityGroupIds: + - !Ref SecurityGroup + OperatorRole: !GetAtt MyCapacityProviderCustomRole.Arn + KmsKeyArn: !Ref KMSKeyArn + + MyFunction: + Type: AWS::Serverless::Function + Properties: + Runtime: nodejs22.x + Handler: index.handler + CodeUri: ${codeuri} + CapacityProviderConfig: + Arn: !GetAtt MyCapacityProvider.Arn + +Metadata: + SamTransformTest: true diff --git a/integration/resources/templates/combination/function_lmi_default.yaml b/integration/resources/templates/combination/function_lmi_default.yaml new file mode 100644 index 0000000000..b2a64fdc31 --- /dev/null +++ b/integration/resources/templates/combination/function_lmi_default.yaml @@ -0,0 +1,31 @@ +Parameters: + SubnetId: + Type: String + SecurityGroup: + Type: String + KMSKeyArn: + Type: String + +Resources: + MyCapacityProvider: + Type: AWS::Serverless::CapacityProvider + Properties: + CapacityProviderName: !Sub "${AWS::StackName}-cp" + VpcConfig: + SubnetIds: + - !Ref SubnetId + SecurityGroupIds: + - !Ref SecurityGroup + KmsKeyArn: !Ref KMSKeyArn + + MyFunction: + Type: AWS::Serverless::Function + Properties: + Runtime: nodejs22.x + Handler: index.handler + CodeUri: ${codeuri} + CapacityProviderConfig: + Arn: !GetAtt MyCapacityProvider.Arn + +Metadata: + SamTransformTest: true diff --git a/integration/setup/companion-stack.yaml b/integration/setup/companion-stack.yaml index 72a1232673..9f67bd3ffd 100644 --- a/integration/setup/companion-stack.yaml +++ b/integration/setup/companion-stack.yaml @@ -41,6 +41,106 @@ Resources: Type: AWS::S3::Bucket DeletionPolicy: Delete + LMIKMSKey: + Type: AWS::KMS::Key + Properties: + Description: KMS Key for Lambda Capacity Provider Resource + KeyPolicy: + Version: '2012-10-17' + Statement: + - Sid: Enable IAM User Permissions + Effect: Allow + Principal: + AWS: !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:root" + Action: kms:* + Resource: '*' + - Sid: Allow Lambda service + Effect: Allow + Principal: + Service: lambda.amazonaws.com + Action: + - kms:Decrypt + - kms:DescribeKey + Resource: '*' # Lambda Managed Instances (LMI) VPC Resources + + LMIVpc: + Type: AWS::EC2::VPC + Properties: + CidrBlock: 10.0.0.0/16 + EnableDnsHostnames: true + EnableDnsSupport: true + Tags: + - Key: Name + Value: !Sub "${AWS::StackName}-VPC" + + LMIPrivateSubnet: + Type: AWS::EC2::Subnet + Properties: + VpcId: !Ref LMIVpc + CidrBlock: 10.0.1.0/24 + AvailabilityZone: !Select [0, !GetAZs ''] + Tags: + - Key: Name + Value: !Sub "${AWS::StackName}-PrivateSubnet" + + LMIPrivateRouteTable: + Type: AWS::EC2::RouteTable + Properties: + VpcId: !Ref LMIVpc + Tags: + - Key: Name + Value: !Sub "${AWS::StackName}-PrivateRT" + + LMIPrivateSubnetRouteTableAssociation: + Type: AWS::EC2::SubnetRouteTableAssociation + Properties: + SubnetId: !Ref LMIPrivateSubnet + RouteTableId: !Ref LMIPrivateRouteTable + + LMISecurityGroup: + Type: AWS::EC2::SecurityGroup + Properties: + GroupDescription: Security group for capacity provider Lambda functions + VpcId: !Ref LMIVpc + SecurityGroupEgress: + - IpProtocol: tcp + FromPort: 443 + ToPort: 443 + CidrIp: 10.0.0.0/16 + Description: Allow HTTPS within VPC + Tags: + - Key: Name + Value: !Sub "${AWS::StackName}-LambdaSG" + + VPCEndpointSecurityGroup: + Type: AWS::EC2::SecurityGroup + Properties: + GroupDescription: Security group for VPC endpoints + VpcId: !Ref LMIVpc + SecurityGroupIngress: + - IpProtocol: tcp + FromPort: 443 + ToPort: 443 + CidrIp: 10.0.0.0/16 + Description: Allow HTTPS from within VPC + Tags: + - Key: Name + Value: !Sub "${AWS::StackName}-VPCEndpointSG" + + CloudWatchLogsEndpoint: + Type: AWS::EC2::VPCEndpoint + Properties: + VpcId: !Ref LMIVpc + ServiceName: !Sub 'com.amazonaws.${AWS::Region}.logs' + VpcEndpointType: Interface + PrivateDnsEnabled: true + SubnetIds: + - !Ref LMIPrivateSubnet + SecurityGroupIds: + - !Ref VPCEndpointSecurityGroup + Tags: + - Key: Name + Value: !Sub "${AWS::StackName}-CloudWatchLogsEndpoint" Outputs: PreCreatedVpc: Description: Pre-created VPC that can be used inside other tests @@ -66,5 +166,14 @@ Outputs: Description: Pre-created S3 Bucket that can be used inside other tests Value: Ref: PreCreatedS3Bucket + LMISubnetId: + Description: Private subnet ID for Lambda functions + Value: !Ref LMIPrivateSubnet + LMISecurityGroupId: + Description: Security group ID for Lambda functions + Value: !Ref LMISecurityGroup + LMIKMSKeyArn: + Description: ARN of the KMS key for Capacity Provider + Value: !GetAtt LMIKMSKey.Arn Metadata: SamTransformTest: true From d23f209231d2c833826036f23ab8654a1b0224f0 Mon Sep 17 00:00:00 2001 From: Vichym Date: Mon, 5 Jan 2026 17:17:14 -0800 Subject: [PATCH 2/7] address comment --- .../test_function_with_capacity_provider.py | 7 ++--- .../combination/function_lmi_default.json | 12 ++++++-- .../combination/function_lmi_custom.yaml | 10 +++---- .../combination/function_lmi_default.yaml | 29 ++++++++++++++++--- 4 files changed, 41 insertions(+), 17 deletions(-) diff --git a/integration/combination/test_function_with_capacity_provider.py b/integration/combination/test_function_with_capacity_provider.py index 928286ec88..d51e576d5a 100644 --- a/integration/combination/test_function_with_capacity_provider.py +++ b/integration/combination/test_function_with_capacity_provider.py @@ -40,9 +40,6 @@ def test_function_with_capacity_provider_custom_role(self): def test_function_with_capacity_provider_default_role(self): """Test Lambda function with CapacityProviderConfig using default operator role. - - Note: This test is skipped until 12/01/2024 because the managed policy required for - the default operator role is not available until then. """ # Phase 1: Prepare parameters from companion stack parameters = [ @@ -59,7 +56,7 @@ def test_function_with_capacity_provider_default_role(self): self.assertEqual(len(lambda_resources), 1, "Should create exactly one Lambda function") capacity_provider_resources = self.get_stack_resources("AWS::Lambda::CapacityProvider") - self.assertEqual(len(capacity_provider_resources), 1, "Should create exactly one CapacityProvider") + self.assertEqual(len(capacity_provider_resources), 2, "Should create exactly two CapacityProviders") iam_role_resources = self.get_stack_resources("AWS::IAM::Role") - self.assertEqual(len(iam_role_resources), 2, "Should create exactly two IAM roles") + self.assertEqual(len(iam_role_resources), 3, "Should create exactly three IAM roles") diff --git a/integration/resources/expected/combination/function_lmi_default.json b/integration/resources/expected/combination/function_lmi_default.json index c71af2dc91..b39c5e9422 100644 --- a/integration/resources/expected/combination/function_lmi_default.json +++ b/integration/resources/expected/combination/function_lmi_default.json @@ -1,10 +1,18 @@ [ { - "LogicalResourceId": "MyCapacityProvider", + "LogicalResourceId": "SimpleCapacityProvider", "ResourceType": "AWS::Lambda::CapacityProvider" }, { - "LogicalResourceId": "MyCapacityProviderOperatorRole", + "LogicalResourceId": "SimpleCapacityProviderOperatorRole", + "ResourceType": "AWS::IAM::Role" + }, + { + "LogicalResourceId": "AdvancedCapacityProvider", + "ResourceType": "AWS::Lambda::CapacityProvider" + }, + { + "LogicalResourceId": "AdvancedCapacityProviderOperatorRole", "ResourceType": "AWS::IAM::Role" }, { diff --git a/integration/resources/templates/combination/function_lmi_custom.yaml b/integration/resources/templates/combination/function_lmi_custom.yaml index dbbf668dc3..1f72cacfc4 100644 --- a/integration/resources/templates/combination/function_lmi_custom.yaml +++ b/integration/resources/templates/combination/function_lmi_custom.yaml @@ -24,10 +24,9 @@ Resources: Statement: - Effect: Allow Action: - - ec2:TerminateInstances - ec2:AttachNetworkInterface - - ec2:RunInstances - ec2:CreateTags + - ec2:RunInstances Resource: - !Sub arn:${AWS::Partition}:ec2:*:*:instance/* - !Sub arn:${AWS::Partition}:ec2:*:*:network-interface/* @@ -39,12 +38,12 @@ Resources: Action: - ec2:DescribeAvailabilityZones - ec2:DescribeCapacityReservations - - ec2:DescribeInstanceTypes + - ec2:DescribeInstances + - ec2:DescribeInstanceStatus - ec2:DescribeInstanceTypeOfferings + - ec2:DescribeInstanceTypes - ec2:DescribeSecurityGroups - ec2:DescribeSubnets - - ec2:DescribeInstances - - ec2:DescribeInstanceStatus Resource: '*' - Effect: Allow Action: @@ -72,7 +71,6 @@ Resources: SecurityGroupIds: - !Ref SecurityGroup OperatorRole: !GetAtt MyCapacityProviderCustomRole.Arn - KmsKeyArn: !Ref KMSKeyArn MyFunction: Type: AWS::Serverless::Function diff --git a/integration/resources/templates/combination/function_lmi_default.yaml b/integration/resources/templates/combination/function_lmi_default.yaml index b2a64fdc31..674cd89c4b 100644 --- a/integration/resources/templates/combination/function_lmi_default.yaml +++ b/integration/resources/templates/combination/function_lmi_default.yaml @@ -7,16 +7,14 @@ Parameters: Type: String Resources: - MyCapacityProvider: + SimpleCapacityProvider: Type: AWS::Serverless::CapacityProvider Properties: - CapacityProviderName: !Sub "${AWS::StackName}-cp" VpcConfig: SubnetIds: - !Ref SubnetId SecurityGroupIds: - !Ref SecurityGroup - KmsKeyArn: !Ref KMSKeyArn MyFunction: Type: AWS::Serverless::Function @@ -25,7 +23,30 @@ Resources: Handler: index.handler CodeUri: ${codeuri} CapacityProviderConfig: - Arn: !GetAtt MyCapacityProvider.Arn + Arn: !GetAtt SimpleCapacityProvider.Arn + + AdvancedCapacityProvider: + Type: AWS::Serverless::CapacityProvider + Properties: + CapacityProviderName: !Sub "${AWS::StackName}-cp" + VpcConfig: + SubnetIds: + - !Ref SubnetId + SecurityGroupIds: + - !Ref SecurityGroup + InstanceRequirements: + Architectures: + - x86_64 + AllowedTypes: + - m5.large + - m5.xlarge + - m5.2xlarge + ScalingConfig: + MaxVCpuCount: 64 + AverageCPUUtilization: 70 + KmsKeyArn: !Ref KMSKeyArn + Tags: + Environment: Test Metadata: SamTransformTest: true From 882502d60a3f115ba316e8e2c425499fbc88825d Mon Sep 17 00:00:00 2001 From: Vichym Date: Tue, 6 Jan 2026 11:57:30 -0800 Subject: [PATCH 3/7] fix format --- .../test_function_with_capacity_provider.py | 3 +-- .../templates/combination/function_lmi_default.yaml | 10 +++++----- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/integration/combination/test_function_with_capacity_provider.py b/integration/combination/test_function_with_capacity_provider.py index d51e576d5a..a559b5c9d0 100644 --- a/integration/combination/test_function_with_capacity_provider.py +++ b/integration/combination/test_function_with_capacity_provider.py @@ -39,8 +39,7 @@ def test_function_with_capacity_provider_custom_role(self): self.assertEqual(len(iam_role_resources), 2, "Should create exactly two IAM roles") def test_function_with_capacity_provider_default_role(self): - """Test Lambda function with CapacityProviderConfig using default operator role. - """ + """Test Lambda function with CapacityProviderConfig using default operator role.""" # Phase 1: Prepare parameters from companion stack parameters = [ self.generate_parameter("SubnetId", self.companion_stack_outputs["LMISubnetId"]), diff --git a/integration/resources/templates/combination/function_lmi_default.yaml b/integration/resources/templates/combination/function_lmi_default.yaml index 674cd89c4b..8137a51b21 100644 --- a/integration/resources/templates/combination/function_lmi_default.yaml +++ b/integration/resources/templates/combination/function_lmi_default.yaml @@ -24,7 +24,7 @@ Resources: CodeUri: ${codeuri} CapacityProviderConfig: Arn: !GetAtt SimpleCapacityProvider.Arn - + AdvancedCapacityProvider: Type: AWS::Serverless::CapacityProvider Properties: @@ -36,11 +36,11 @@ Resources: - !Ref SecurityGroup InstanceRequirements: Architectures: - - x86_64 + - x86_64 AllowedTypes: - - m5.large - - m5.xlarge - - m5.2xlarge + - m5.large + - m5.xlarge + - m5.2xlarge ScalingConfig: MaxVCpuCount: 64 AverageCPUUtilization: 70 From f1577b49c124a1f7943190b9ee85cf362f093085 Mon Sep 17 00:00:00 2001 From: Vichym Date: Tue, 13 Jan 2026 16:45:55 -0800 Subject: [PATCH 4/7] update tests with additional validation --- .../test_function_with_capacity_provider.py | 218 ++++++++++++++++-- 1 file changed, 205 insertions(+), 13 deletions(-) diff --git a/integration/combination/test_function_with_capacity_provider.py b/integration/combination/test_function_with_capacity_provider.py index a559b5c9d0..5154a47270 100644 --- a/integration/combination/test_function_with_capacity_provider.py +++ b/integration/combination/test_function_with_capacity_provider.py @@ -16,15 +16,18 @@ class TestFunctionWithCapacityProvider(BaseTest): def companion_stack_outputs(self, get_companion_stack_outputs): self.companion_stack_outputs = get_companion_stack_outputs - def test_function_with_capacity_provider_custom_role(self): - """Test Lambda function with CapacityProviderConfig using custom operator role.""" - # Phase 1: Prepare parameters from companion stack - parameters = [ + def generate_lmi_parameters(self): + return [ self.generate_parameter("SubnetId", self.companion_stack_outputs["LMISubnetId"]), self.generate_parameter("SecurityGroup", self.companion_stack_outputs["LMISecurityGroupId"]), self.generate_parameter("KMSKeyArn", self.companion_stack_outputs["LMIKMSKeyArn"]), ] + def test_function_with_capacity_provider_custom_role(self): + """Test Lambda function with CapacityProviderConfig using custom operator role.""" + # Phase 1: Prepare parameters from companion stack + parameters = self.generate_lmi_parameters() + # Phase 2: Deploy and verify against expected JSON self.create_and_verify_stack("combination/function_lmi_custom", parameters) @@ -35,17 +38,61 @@ def test_function_with_capacity_provider_custom_role(self): capacity_provider_resources = self.get_stack_resources("AWS::Lambda::CapacityProvider") self.assertEqual(len(capacity_provider_resources), 1, "Should create exactly one CapacityProvider") - iam_role_resources = self.get_stack_resources("AWS::IAM::Role") - self.assertEqual(len(iam_role_resources), 2, "Should create exactly two IAM roles") + # Phase 4: Validate function capacity provider configuration + lambda_function = lambda_resources[0] + capacity_provider = capacity_provider_resources[0] + + function_capacity_provider_config = self.get_function_capacity_provider_config( + lambda_function["PhysicalResourceId"] + ) + self.assertIsNotNone(function_capacity_provider_config, "Function should have capacity provider configuration") + self.assertIn( + "LambdaManagedInstancesCapacityProviderConfig", + function_capacity_provider_config, + "Function should have LambdaManagedInstancesCapacityProviderConfig", + ) + + lmi_config = function_capacity_provider_config["LambdaManagedInstancesCapacityProviderConfig"] + function_capacity_provider_arn = lmi_config.get("CapacityProviderArn") + self.assertIsNotNone(function_capacity_provider_arn, "Function should reference a capacity provider ARN") + + # Phase 5: Validate capacity provider details + capacity_provider_config = self.get_lambda_capacity_provider_config(capacity_provider) + self.assertIsNotNone(capacity_provider_config, "Capacity provider should have configuration") + self.assertEqual(capacity_provider_config["State"], "Active", "Capacity provider should be in Active state") + + # Verify the function uses the correct capacity provider ARN + actual_capacity_provider_arn = capacity_provider_config.get("CapacityProviderArn") + self.assertEqual( + function_capacity_provider_arn, + actual_capacity_provider_arn, + "Function should reference the correct capacity provider ARN", + ) + + # Phase 6: Verify capacity provider uses custom operator role + permissions_config = capacity_provider_config.get("PermissionsConfig") + self.assertIsNotNone(permissions_config, "Capacity provider should have permissions configuration") + + capacity_provider_operator_role_arn = permissions_config.get("CapacityProviderOperatorRoleArn") + self.assertIsNotNone( + capacity_provider_operator_role_arn, "Capacity provider should have custom operator role ARN" + ) + + # Get the physical ID of the custom role using its logical ID + custom_operator_role_physical_id = self.get_physical_id_by_logical_id("MyCapacityProviderCustomRole") + self.assertIsNotNone(custom_operator_role_physical_id, "MyCapacityProviderCustomRole should exist in the stack") + + # Verify the capacity provider uses the custom role by comparing physical IDs + self.assertIn( + custom_operator_role_physical_id, + capacity_provider_operator_role_arn, + f"Capacity provider should use the custom operator role with physical ID {custom_operator_role_physical_id}", + ) def test_function_with_capacity_provider_default_role(self): """Test Lambda function with CapacityProviderConfig using default operator role.""" # Phase 1: Prepare parameters from companion stack - parameters = [ - self.generate_parameter("SubnetId", self.companion_stack_outputs["LMISubnetId"]), - self.generate_parameter("SecurityGroup", self.companion_stack_outputs["LMISecurityGroupId"]), - self.generate_parameter("KMSKeyArn", self.companion_stack_outputs["LMIKMSKeyArn"]), - ] + parameters = self.generate_lmi_parameters() # Phase 2: Deploy and verify against expected JSON self.create_and_verify_stack("combination/function_lmi_default", parameters) @@ -57,5 +104,150 @@ def test_function_with_capacity_provider_default_role(self): capacity_provider_resources = self.get_stack_resources("AWS::Lambda::CapacityProvider") self.assertEqual(len(capacity_provider_resources), 2, "Should create exactly two CapacityProviders") - iam_role_resources = self.get_stack_resources("AWS::IAM::Role") - self.assertEqual(len(iam_role_resources), 3, "Should create exactly three IAM roles") + # Phase 4: Validate function capacity provider configuration + simple_capacity_provider = next( + r for r in capacity_provider_resources if r["LogicalResourceId"] == "SimpleCapacityProvider" + ) + advanced_capacity_provider = next( + r for r in capacity_provider_resources if r["LogicalResourceId"] == "AdvancedCapacityProvider" + ) + my_function = lambda_resources[0] + + function_config = self.get_function_capacity_provider_config(my_function["PhysicalResourceId"]) + self.assertIsNotNone(function_config, "Function should have capacity provider configuration") + self.assertIn( + "LambdaManagedInstancesCapacityProviderConfig", + function_config, + "Function should have LambdaManagedInstancesCapacityProviderConfig", + ) + + lmi_config = function_config["LambdaManagedInstancesCapacityProviderConfig"] + function_capacity_provider_arn = lmi_config.get("CapacityProviderArn") + self.assertIsNotNone(function_capacity_provider_arn, "Function should reference a capacity provider ARN") + + # Phase 5: Validate SimpleCapacityProvider configuration + simple_cp_config = self.get_lambda_capacity_provider_config(simple_capacity_provider) + self.assertIsNotNone(simple_cp_config, "SimpleCapacityProvider should have configuration") + self.assertEqual(simple_cp_config["State"], "Active", "SimpleCapacityProvider should be in Active state") + + # Verify the function uses SimpleCapacityProvider + simple_cp_arn = simple_cp_config.get("CapacityProviderArn") + self.assertEqual( + function_capacity_provider_arn, simple_cp_arn, "Function should reference SimpleCapacityProvider ARN" + ) + + # Verify SimpleCapacityProvider VPC configuration + simple_vpc_config = simple_cp_config.get("VpcConfig") + self.assertIsNotNone(simple_vpc_config, "SimpleCapacityProvider should have VPC configuration") + self.assertIn( + self.companion_stack_outputs["LMISubnetId"], + simple_vpc_config["SubnetIds"], + "SimpleCapacityProvider should use the correct subnet", + ) + self.assertIn( + self.companion_stack_outputs["LMISecurityGroupId"], + simple_vpc_config["SecurityGroupIds"], + "SimpleCapacityProvider should use the correct security group", + ) + + # Verify SimpleCapacityProvider uses SAM-generated default operator role + simple_permissions_config = simple_cp_config.get("PermissionsConfig") + self.assertIsNotNone(simple_permissions_config, "SimpleCapacityProvider should have permissions configuration") + simple_operator_arn = simple_permissions_config.get("CapacityProviderOperatorRoleArn") + self.assertIsNotNone(simple_operator_arn, "SimpleCapacityProvider should have operator role ARN") + + # Phase 6: Validate AdvancedCapacityProvider configuration + advanced_cp_config = self.get_lambda_capacity_provider_config(advanced_capacity_provider) + self.assertIsNotNone(advanced_cp_config, "AdvancedCapacityProvider should have configuration") + self.assertEqual(advanced_cp_config["State"], "Active", "AdvancedCapacityProvider should be in Active state") + + # Verify AdvancedCapacityProvider VPC configuration + advanced_vpc_config = advanced_cp_config.get("VpcConfig") + self.assertIsNotNone(advanced_vpc_config, "AdvancedCapacityProvider should have VPC configuration") + self.assertIn( + self.companion_stack_outputs["LMISubnetId"], + advanced_vpc_config["SubnetIds"], + "AdvancedCapacityProvider should use the correct subnet", + ) + self.assertIn( + self.companion_stack_outputs["LMISecurityGroupId"], + advanced_vpc_config["SecurityGroupIds"], + "AdvancedCapacityProvider should use the correct security group", + ) + + # Verify AdvancedCapacityProvider permissions configuration + advanced_permissions_config = advanced_cp_config.get("PermissionsConfig") + self.assertIsNotNone( + advanced_permissions_config, + "AdvancedCapacityProvider should have permissions configuration", + ) + + # Verify AdvancedCapacityProvider instance requirements + instance_requirements = advanced_cp_config.get("InstanceRequirements") + self.assertIsNotNone(instance_requirements, "AdvancedCapacityProvider should have instance requirements") + self.assertIn( + "x86_64", + instance_requirements.get("Architectures", []), + "AdvancedCapacityProvider should have x86_64 architecture", + ) + allowed_types = instance_requirements.get("AllowedInstanceTypes", []) + self.assertIn("m5.large", allowed_types, "AdvancedCapacityProvider should allow m5.large") + self.assertIn("m5.xlarge", allowed_types, "AdvancedCapacityProvider should allow m5.xlarge") + self.assertIn("m5.2xlarge", allowed_types, "AdvancedCapacityProvider should allow m5.2xlarge") + + # Verify AdvancedCapacityProvider scaling configuration + scaling_config = advanced_cp_config.get("CapacityProviderScalingConfig") + self.assertIsNotNone(scaling_config, "AdvancedCapacityProvider should have scaling configuration") + self.assertEqual( + scaling_config.get("MaxVCpuCount"), 64, "AdvancedCapacityProvider should have MaxVCpuCount of 64" + ) + scaling_policies = scaling_config.get("ScalingPolicies", []) + self.assertTrue(len(scaling_policies) > 0, "AdvancedCapacityProvider should have scaling policies") + cpu_policy = next( + ( + p + for p in scaling_policies + if p.get("PredefinedMetricType") == "LambdaCapacityProviderAverageCPUUtilization" + ), + None, + ) + self.assertIsNotNone(cpu_policy, "AdvancedCapacityProvider should have CPU utilization scaling policy") + self.assertEqual( + cpu_policy.get("TargetValue"), 70.0, "AdvancedCapacityProvider should have CPU utilization target of 70" + ) + + # Verify AdvancedCapacityProvider KMS key + self.assertEqual( + advanced_cp_config.get("KmsKeyArn"), + self.companion_stack_outputs["LMIKMSKeyArn"], + "AdvancedCapacityProvider should use the correct KMS key", + ) + + def get_function_capacity_provider_config(self, function_name, alias_name=None): + lambda_client = self.client_provider.lambda_client + + try: + # Build the function identifier - include alias if provided + function_identifier = f"{function_name}:{alias_name}" if alias_name else function_name + # Get the function configuration + response = lambda_client.get_function_configuration(FunctionName=function_identifier) + # Return the CapacityProviderConfig if it exists + return response.get("CapacityProviderConfig") + except Exception as e: + # Log the error and return None for graceful handling + print(f"Error getting function capacity provider config: {e}") + return None + + def get_lambda_capacity_provider_config(self, capacity_provider_resource): + lambda_client = self.client_provider.lambda_client + + try: + # Extract capacity provider name from resource dict + capacity_provider_name = capacity_provider_resource["PhysicalResourceId"] + # Get the capacity provider configuration + response = lambda_client.get_capacity_provider(CapacityProviderName=capacity_provider_name) + return response.get("CapacityProvider") + except Exception as e: + # Log the error and return None for graceful handling + print(f"Error getting capacity provider config for {capacity_provider_resource}: {e}") + return None From 28c45467df33dca3c43894dce0605790582add03 Mon Sep 17 00:00:00 2001 From: Vichym Date: Fri, 16 Jan 2026 10:49:16 -0800 Subject: [PATCH 5/7] refactor: address code review comments - Refactor get_lambda_capacity_provider_config to accept capacity provider name directly - Use get_physical_id_by_logical_id at call sites for consistency - Extract duplicate validation logic into reusable helper methods: - verify_capacity_provider_basic_config - verify_capacity_provider_vpc_config - verify_capacity_provider_permissions_config - Remove unused local variables (capacity_provider, simple_capacity_provider, advanced_capacity_provider) This addresses review comments from valerena on Jan 16, 2026 --- .../test_function_with_capacity_provider.py | 132 ++++++++---------- 1 file changed, 59 insertions(+), 73 deletions(-) diff --git a/integration/combination/test_function_with_capacity_provider.py b/integration/combination/test_function_with_capacity_provider.py index 5154a47270..8f5a69b256 100644 --- a/integration/combination/test_function_with_capacity_provider.py +++ b/integration/combination/test_function_with_capacity_provider.py @@ -23,6 +23,46 @@ def generate_lmi_parameters(self): self.generate_parameter("KMSKeyArn", self.companion_stack_outputs["LMIKMSKeyArn"]), ] + def verify_capacity_provider_basic_config(self, cp_config, cp_name): + """Verify basic capacity provider configuration (state, existence).""" + self.assertIsNotNone(cp_config, f"{cp_name} should have configuration") + self.assertEqual(cp_config["State"], "Active", f"{cp_name} should be in Active state") + + def verify_capacity_provider_vpc_config(self, vpc_config, cp_name): + """Verify capacity provider VPC configuration matches companion stack outputs.""" + self.assertIsNotNone(vpc_config, f"{cp_name} should have VPC configuration") + self.assertIn( + self.companion_stack_outputs["LMISubnetId"], + vpc_config["SubnetIds"], + f"{cp_name} should use the correct subnet", + ) + self.assertIn( + self.companion_stack_outputs["LMISecurityGroupId"], + vpc_config["SecurityGroupIds"], + f"{cp_name} should use the correct security group", + ) + + def verify_capacity_provider_permissions_config(self, permissions_config, cp_name): + """Verify capacity provider has permissions configuration with operator role.""" + self.assertIsNotNone(permissions_config, f"{cp_name} should have permissions configuration") + operator_role_arn = permissions_config.get("CapacityProviderOperatorRoleArn") + self.assertIsNotNone(operator_role_arn, f"{cp_name} should have operator role ARN") + return operator_role_arn + + def verify_function_capacity_provider_config(self, function_capacity_provider_config): + """Verify and extract capacity provider ARN from function configuration.""" + self.assertIsNotNone(function_capacity_provider_config, "Function should have capacity provider configuration") + self.assertIn( + "LambdaManagedInstancesCapacityProviderConfig", + function_capacity_provider_config, + "Function should have LambdaManagedInstancesCapacityProviderConfig", + ) + + lmi_config = function_capacity_provider_config["LambdaManagedInstancesCapacityProviderConfig"] + function_capacity_provider_arn = lmi_config.get("CapacityProviderArn") + self.assertIsNotNone(function_capacity_provider_arn, "Function should reference a capacity provider ARN") + return function_capacity_provider_arn + def test_function_with_capacity_provider_custom_role(self): """Test Lambda function with CapacityProviderConfig using custom operator role.""" # Phase 1: Prepare parameters from companion stack @@ -40,26 +80,17 @@ def test_function_with_capacity_provider_custom_role(self): # Phase 4: Validate function capacity provider configuration lambda_function = lambda_resources[0] - capacity_provider = capacity_provider_resources[0] function_capacity_provider_config = self.get_function_capacity_provider_config( lambda_function["PhysicalResourceId"] ) - self.assertIsNotNone(function_capacity_provider_config, "Function should have capacity provider configuration") - self.assertIn( - "LambdaManagedInstancesCapacityProviderConfig", - function_capacity_provider_config, - "Function should have LambdaManagedInstancesCapacityProviderConfig", + function_capacity_provider_arn = self.verify_function_capacity_provider_config( + function_capacity_provider_config ) - lmi_config = function_capacity_provider_config["LambdaManagedInstancesCapacityProviderConfig"] - function_capacity_provider_arn = lmi_config.get("CapacityProviderArn") - self.assertIsNotNone(function_capacity_provider_arn, "Function should reference a capacity provider ARN") - # Phase 5: Validate capacity provider details - capacity_provider_config = self.get_lambda_capacity_provider_config(capacity_provider) - self.assertIsNotNone(capacity_provider_config, "Capacity provider should have configuration") - self.assertEqual(capacity_provider_config["State"], "Active", "Capacity provider should be in Active state") + capacity_provider_config = self.get_lambda_capacity_provider_config("MyCapacityProvider") + self.verify_capacity_provider_basic_config(capacity_provider_config, "MyCapacityProvider") # Verify the function uses the correct capacity provider ARN actual_capacity_provider_arn = capacity_provider_config.get("CapacityProviderArn") @@ -71,11 +102,8 @@ def test_function_with_capacity_provider_custom_role(self): # Phase 6: Verify capacity provider uses custom operator role permissions_config = capacity_provider_config.get("PermissionsConfig") - self.assertIsNotNone(permissions_config, "Capacity provider should have permissions configuration") - - capacity_provider_operator_role_arn = permissions_config.get("CapacityProviderOperatorRoleArn") - self.assertIsNotNone( - capacity_provider_operator_role_arn, "Capacity provider should have custom operator role ARN" + capacity_provider_operator_role_arn = self.verify_capacity_provider_permissions_config( + permissions_config, "MyCapacityProvider" ) # Get the physical ID of the custom role using its logical ID @@ -105,30 +133,14 @@ def test_function_with_capacity_provider_default_role(self): self.assertEqual(len(capacity_provider_resources), 2, "Should create exactly two CapacityProviders") # Phase 4: Validate function capacity provider configuration - simple_capacity_provider = next( - r for r in capacity_provider_resources if r["LogicalResourceId"] == "SimpleCapacityProvider" - ) - advanced_capacity_provider = next( - r for r in capacity_provider_resources if r["LogicalResourceId"] == "AdvancedCapacityProvider" - ) my_function = lambda_resources[0] function_config = self.get_function_capacity_provider_config(my_function["PhysicalResourceId"]) - self.assertIsNotNone(function_config, "Function should have capacity provider configuration") - self.assertIn( - "LambdaManagedInstancesCapacityProviderConfig", - function_config, - "Function should have LambdaManagedInstancesCapacityProviderConfig", - ) - - lmi_config = function_config["LambdaManagedInstancesCapacityProviderConfig"] - function_capacity_provider_arn = lmi_config.get("CapacityProviderArn") - self.assertIsNotNone(function_capacity_provider_arn, "Function should reference a capacity provider ARN") + function_capacity_provider_arn = self.verify_function_capacity_provider_config(function_config) # Phase 5: Validate SimpleCapacityProvider configuration - simple_cp_config = self.get_lambda_capacity_provider_config(simple_capacity_provider) - self.assertIsNotNone(simple_cp_config, "SimpleCapacityProvider should have configuration") - self.assertEqual(simple_cp_config["State"], "Active", "SimpleCapacityProvider should be in Active state") + simple_cp_config = self.get_lambda_capacity_provider_config("SimpleCapacityProvider") + self.verify_capacity_provider_basic_config(simple_cp_config, "SimpleCapacityProvider") # Verify the function uses SimpleCapacityProvider simple_cp_arn = simple_cp_config.get("CapacityProviderArn") @@ -138,49 +150,23 @@ def test_function_with_capacity_provider_default_role(self): # Verify SimpleCapacityProvider VPC configuration simple_vpc_config = simple_cp_config.get("VpcConfig") - self.assertIsNotNone(simple_vpc_config, "SimpleCapacityProvider should have VPC configuration") - self.assertIn( - self.companion_stack_outputs["LMISubnetId"], - simple_vpc_config["SubnetIds"], - "SimpleCapacityProvider should use the correct subnet", - ) - self.assertIn( - self.companion_stack_outputs["LMISecurityGroupId"], - simple_vpc_config["SecurityGroupIds"], - "SimpleCapacityProvider should use the correct security group", - ) + self.verify_capacity_provider_vpc_config(simple_vpc_config, "SimpleCapacityProvider") # Verify SimpleCapacityProvider uses SAM-generated default operator role simple_permissions_config = simple_cp_config.get("PermissionsConfig") - self.assertIsNotNone(simple_permissions_config, "SimpleCapacityProvider should have permissions configuration") - simple_operator_arn = simple_permissions_config.get("CapacityProviderOperatorRoleArn") - self.assertIsNotNone(simple_operator_arn, "SimpleCapacityProvider should have operator role ARN") + self.verify_capacity_provider_permissions_config(simple_permissions_config, "SimpleCapacityProvider") # Phase 6: Validate AdvancedCapacityProvider configuration - advanced_cp_config = self.get_lambda_capacity_provider_config(advanced_capacity_provider) - self.assertIsNotNone(advanced_cp_config, "AdvancedCapacityProvider should have configuration") - self.assertEqual(advanced_cp_config["State"], "Active", "AdvancedCapacityProvider should be in Active state") + advanced_cp_config = self.get_lambda_capacity_provider_config("AdvancedCapacityProvider") + self.verify_capacity_provider_basic_config(advanced_cp_config, "AdvancedCapacityProvider") # Verify AdvancedCapacityProvider VPC configuration advanced_vpc_config = advanced_cp_config.get("VpcConfig") - self.assertIsNotNone(advanced_vpc_config, "AdvancedCapacityProvider should have VPC configuration") - self.assertIn( - self.companion_stack_outputs["LMISubnetId"], - advanced_vpc_config["SubnetIds"], - "AdvancedCapacityProvider should use the correct subnet", - ) - self.assertIn( - self.companion_stack_outputs["LMISecurityGroupId"], - advanced_vpc_config["SecurityGroupIds"], - "AdvancedCapacityProvider should use the correct security group", - ) + self.verify_capacity_provider_vpc_config(advanced_vpc_config, "AdvancedCapacityProvider") # Verify AdvancedCapacityProvider permissions configuration advanced_permissions_config = advanced_cp_config.get("PermissionsConfig") - self.assertIsNotNone( - advanced_permissions_config, - "AdvancedCapacityProvider should have permissions configuration", - ) + self.verify_capacity_provider_permissions_config(advanced_permissions_config, "AdvancedCapacityProvider") # Verify AdvancedCapacityProvider instance requirements instance_requirements = advanced_cp_config.get("InstanceRequirements") @@ -238,16 +224,16 @@ def get_function_capacity_provider_config(self, function_name, alias_name=None): print(f"Error getting function capacity provider config: {e}") return None - def get_lambda_capacity_provider_config(self, capacity_provider_resource): + def get_lambda_capacity_provider_config(self, capacity_provider_logical_id): lambda_client = self.client_provider.lambda_client try: - # Extract capacity provider name from resource dict - capacity_provider_name = capacity_provider_resource["PhysicalResourceId"] + # Get the physical ID from the logical ID + capacity_provider_name = self.get_physical_id_by_logical_id(capacity_provider_logical_id) # Get the capacity provider configuration response = lambda_client.get_capacity_provider(CapacityProviderName=capacity_provider_name) return response.get("CapacityProvider") except Exception as e: # Log the error and return None for graceful handling - print(f"Error getting capacity provider config for {capacity_provider_resource}: {e}") + print(f"Error getting capacity provider config for {capacity_provider_logical_id}: {e}") return None From a167affced8515655365221f9d33d3c1c66c7ada Mon Sep 17 00:00:00 2001 From: Vichym Date: Fri, 16 Jan 2026 11:40:48 -0800 Subject: [PATCH 6/7] additional refactoring --- .../test_function_with_capacity_provider.py | 37 +++++++++++-------- 1 file changed, 22 insertions(+), 15 deletions(-) diff --git a/integration/combination/test_function_with_capacity_provider.py b/integration/combination/test_function_with_capacity_provider.py index 8f5a69b256..a12bb98f54 100644 --- a/integration/combination/test_function_with_capacity_provider.py +++ b/integration/combination/test_function_with_capacity_provider.py @@ -24,9 +24,12 @@ def generate_lmi_parameters(self): ] def verify_capacity_provider_basic_config(self, cp_config, cp_name): - """Verify basic capacity provider configuration (state, existence).""" + """Verify basic capacity provider configuration (state, existence) and return ARN.""" self.assertIsNotNone(cp_config, f"{cp_name} should have configuration") self.assertEqual(cp_config["State"], "Active", f"{cp_name} should be in Active state") + capacity_provider_arn = cp_config.get("CapacityProviderArn") + self.assertIsNotNone(capacity_provider_arn, f"{cp_name} should have a capacity provider ARN") + return capacity_provider_arn def verify_capacity_provider_vpc_config(self, vpc_config, cp_name): """Verify capacity provider VPC configuration matches companion stack outputs.""" @@ -55,14 +58,24 @@ def verify_function_capacity_provider_config(self, function_capacity_provider_co self.assertIn( "LambdaManagedInstancesCapacityProviderConfig", function_capacity_provider_config, - "Function should have LambdaManagedInstancesCapacityProviderConfig", + "Function LambdaManagedInstancesCapacityProviderConfig should have LambdaManagedInstancesCapacityProviderConfig", ) lmi_config = function_capacity_provider_config["LambdaManagedInstancesCapacityProviderConfig"] function_capacity_provider_arn = lmi_config.get("CapacityProviderArn") - self.assertIsNotNone(function_capacity_provider_arn, "Function should reference a capacity provider ARN") + self.assertIsNotNone( + function_capacity_provider_arn, "Function capacity provider config should have a capacity provider ARN" + ) return function_capacity_provider_arn + def verify_capacity_provider_arn_match(self, function_capacity_provider_arn, capacity_provider_arn): + """Verify that the function references the correct capacity provider ARN.""" + self.assertEqual( + function_capacity_provider_arn, + capacity_provider_arn, + "Function should reference the correct capacity provider ARN", + ) + def test_function_with_capacity_provider_custom_role(self): """Test Lambda function with CapacityProviderConfig using custom operator role.""" # Phase 1: Prepare parameters from companion stack @@ -90,15 +103,12 @@ def test_function_with_capacity_provider_custom_role(self): # Phase 5: Validate capacity provider details capacity_provider_config = self.get_lambda_capacity_provider_config("MyCapacityProvider") - self.verify_capacity_provider_basic_config(capacity_provider_config, "MyCapacityProvider") + actual_capacity_provider_arn = self.verify_capacity_provider_basic_config( + capacity_provider_config, "MyCapacityProvider" + ) # Verify the function uses the correct capacity provider ARN - actual_capacity_provider_arn = capacity_provider_config.get("CapacityProviderArn") - self.assertEqual( - function_capacity_provider_arn, - actual_capacity_provider_arn, - "Function should reference the correct capacity provider ARN", - ) + self.verify_capacity_provider_arn_match(function_capacity_provider_arn, actual_capacity_provider_arn) # Phase 6: Verify capacity provider uses custom operator role permissions_config = capacity_provider_config.get("PermissionsConfig") @@ -140,13 +150,10 @@ def test_function_with_capacity_provider_default_role(self): # Phase 5: Validate SimpleCapacityProvider configuration simple_cp_config = self.get_lambda_capacity_provider_config("SimpleCapacityProvider") - self.verify_capacity_provider_basic_config(simple_cp_config, "SimpleCapacityProvider") + simple_cp_arn = self.verify_capacity_provider_basic_config(simple_cp_config, "SimpleCapacityProvider") # Verify the function uses SimpleCapacityProvider - simple_cp_arn = simple_cp_config.get("CapacityProviderArn") - self.assertEqual( - function_capacity_provider_arn, simple_cp_arn, "Function should reference SimpleCapacityProvider ARN" - ) + self.verify_capacity_provider_arn_match(function_capacity_provider_arn, simple_cp_arn) # Verify SimpleCapacityProvider VPC configuration simple_vpc_config = simple_cp_config.get("VpcConfig") From 311dd700afa1c89cd113804b31fe272d875d2dd8 Mon Sep 17 00:00:00 2001 From: Vichym Date: Wed, 21 Jan 2026 10:46:35 -0800 Subject: [PATCH 7/7] refactor: clean up test code by moving .get() calls into verification methods - Move .get('VpcConfig') and .get('PermissionsConfig') inside verification methods - Remove redundant inline comments that duplicate what the code clearly shows - Simplify test methods by passing cp_config directly to verification methods --- .../test_function_with_capacity_provider.py | 53 ++++--------------- 1 file changed, 9 insertions(+), 44 deletions(-) diff --git a/integration/combination/test_function_with_capacity_provider.py b/integration/combination/test_function_with_capacity_provider.py index a12bb98f54..203bbac2a0 100644 --- a/integration/combination/test_function_with_capacity_provider.py +++ b/integration/combination/test_function_with_capacity_provider.py @@ -31,8 +31,9 @@ def verify_capacity_provider_basic_config(self, cp_config, cp_name): self.assertIsNotNone(capacity_provider_arn, f"{cp_name} should have a capacity provider ARN") return capacity_provider_arn - def verify_capacity_provider_vpc_config(self, vpc_config, cp_name): + def verify_capacity_provider_vpc_config(self, cp_config, cp_name): """Verify capacity provider VPC configuration matches companion stack outputs.""" + vpc_config = cp_config.get("VpcConfig") self.assertIsNotNone(vpc_config, f"{cp_name} should have VPC configuration") self.assertIn( self.companion_stack_outputs["LMISubnetId"], @@ -45,8 +46,9 @@ def verify_capacity_provider_vpc_config(self, vpc_config, cp_name): f"{cp_name} should use the correct security group", ) - def verify_capacity_provider_permissions_config(self, permissions_config, cp_name): + def verify_capacity_provider_permissions_config(self, cp_config, cp_name): """Verify capacity provider has permissions configuration with operator role.""" + permissions_config = cp_config.get("PermissionsConfig") self.assertIsNotNone(permissions_config, f"{cp_name} should have permissions configuration") operator_role_arn = permissions_config.get("CapacityProviderOperatorRoleArn") self.assertIsNotNone(operator_role_arn, f"{cp_name} should have operator role ARN") @@ -78,22 +80,16 @@ def verify_capacity_provider_arn_match(self, function_capacity_provider_arn, cap def test_function_with_capacity_provider_custom_role(self): """Test Lambda function with CapacityProviderConfig using custom operator role.""" - # Phase 1: Prepare parameters from companion stack parameters = self.generate_lmi_parameters() - - # Phase 2: Deploy and verify against expected JSON self.create_and_verify_stack("combination/function_lmi_custom", parameters) - # Phase 3: Verify resource counts lambda_resources = self.get_stack_resources("AWS::Lambda::Function") self.assertEqual(len(lambda_resources), 1, "Should create exactly one Lambda function") capacity_provider_resources = self.get_stack_resources("AWS::Lambda::CapacityProvider") self.assertEqual(len(capacity_provider_resources), 1, "Should create exactly one CapacityProvider") - # Phase 4: Validate function capacity provider configuration lambda_function = lambda_resources[0] - function_capacity_provider_config = self.get_function_capacity_provider_config( lambda_function["PhysicalResourceId"] ) @@ -101,26 +97,19 @@ def test_function_with_capacity_provider_custom_role(self): function_capacity_provider_config ) - # Phase 5: Validate capacity provider details capacity_provider_config = self.get_lambda_capacity_provider_config("MyCapacityProvider") actual_capacity_provider_arn = self.verify_capacity_provider_basic_config( capacity_provider_config, "MyCapacityProvider" ) - - # Verify the function uses the correct capacity provider ARN self.verify_capacity_provider_arn_match(function_capacity_provider_arn, actual_capacity_provider_arn) - # Phase 6: Verify capacity provider uses custom operator role - permissions_config = capacity_provider_config.get("PermissionsConfig") capacity_provider_operator_role_arn = self.verify_capacity_provider_permissions_config( - permissions_config, "MyCapacityProvider" + capacity_provider_config, "MyCapacityProvider" ) - # Get the physical ID of the custom role using its logical ID custom_operator_role_physical_id = self.get_physical_id_by_logical_id("MyCapacityProviderCustomRole") self.assertIsNotNone(custom_operator_role_physical_id, "MyCapacityProviderCustomRole should exist in the stack") - # Verify the capacity provider uses the custom role by comparing physical IDs self.assertIn( custom_operator_role_physical_id, capacity_provider_operator_role_arn, @@ -129,53 +118,31 @@ def test_function_with_capacity_provider_custom_role(self): def test_function_with_capacity_provider_default_role(self): """Test Lambda function with CapacityProviderConfig using default operator role.""" - # Phase 1: Prepare parameters from companion stack parameters = self.generate_lmi_parameters() - - # Phase 2: Deploy and verify against expected JSON self.create_and_verify_stack("combination/function_lmi_default", parameters) - # Phase 3: Verify resource counts lambda_resources = self.get_stack_resources("AWS::Lambda::Function") self.assertEqual(len(lambda_resources), 1, "Should create exactly one Lambda function") capacity_provider_resources = self.get_stack_resources("AWS::Lambda::CapacityProvider") self.assertEqual(len(capacity_provider_resources), 2, "Should create exactly two CapacityProviders") - # Phase 4: Validate function capacity provider configuration my_function = lambda_resources[0] - function_config = self.get_function_capacity_provider_config(my_function["PhysicalResourceId"]) function_capacity_provider_arn = self.verify_function_capacity_provider_config(function_config) - # Phase 5: Validate SimpleCapacityProvider configuration simple_cp_config = self.get_lambda_capacity_provider_config("SimpleCapacityProvider") simple_cp_arn = self.verify_capacity_provider_basic_config(simple_cp_config, "SimpleCapacityProvider") - - # Verify the function uses SimpleCapacityProvider self.verify_capacity_provider_arn_match(function_capacity_provider_arn, simple_cp_arn) - # Verify SimpleCapacityProvider VPC configuration - simple_vpc_config = simple_cp_config.get("VpcConfig") - self.verify_capacity_provider_vpc_config(simple_vpc_config, "SimpleCapacityProvider") - - # Verify SimpleCapacityProvider uses SAM-generated default operator role - simple_permissions_config = simple_cp_config.get("PermissionsConfig") - self.verify_capacity_provider_permissions_config(simple_permissions_config, "SimpleCapacityProvider") + self.verify_capacity_provider_vpc_config(simple_cp_config, "SimpleCapacityProvider") + self.verify_capacity_provider_permissions_config(simple_cp_config, "SimpleCapacityProvider") - # Phase 6: Validate AdvancedCapacityProvider configuration advanced_cp_config = self.get_lambda_capacity_provider_config("AdvancedCapacityProvider") self.verify_capacity_provider_basic_config(advanced_cp_config, "AdvancedCapacityProvider") + self.verify_capacity_provider_vpc_config(advanced_cp_config, "AdvancedCapacityProvider") + self.verify_capacity_provider_permissions_config(advanced_cp_config, "AdvancedCapacityProvider") - # Verify AdvancedCapacityProvider VPC configuration - advanced_vpc_config = advanced_cp_config.get("VpcConfig") - self.verify_capacity_provider_vpc_config(advanced_vpc_config, "AdvancedCapacityProvider") - - # Verify AdvancedCapacityProvider permissions configuration - advanced_permissions_config = advanced_cp_config.get("PermissionsConfig") - self.verify_capacity_provider_permissions_config(advanced_permissions_config, "AdvancedCapacityProvider") - - # Verify AdvancedCapacityProvider instance requirements instance_requirements = advanced_cp_config.get("InstanceRequirements") self.assertIsNotNone(instance_requirements, "AdvancedCapacityProvider should have instance requirements") self.assertIn( @@ -188,7 +155,6 @@ def test_function_with_capacity_provider_default_role(self): self.assertIn("m5.xlarge", allowed_types, "AdvancedCapacityProvider should allow m5.xlarge") self.assertIn("m5.2xlarge", allowed_types, "AdvancedCapacityProvider should allow m5.2xlarge") - # Verify AdvancedCapacityProvider scaling configuration scaling_config = advanced_cp_config.get("CapacityProviderScalingConfig") self.assertIsNotNone(scaling_config, "AdvancedCapacityProvider should have scaling configuration") self.assertEqual( @@ -209,7 +175,6 @@ def test_function_with_capacity_provider_default_role(self): cpu_policy.get("TargetValue"), 70.0, "AdvancedCapacityProvider should have CPU utilization target of 70" ) - # Verify AdvancedCapacityProvider KMS key self.assertEqual( advanced_cp_config.get("KmsKeyArn"), self.companion_stack_outputs["LMIKMSKeyArn"],