diff --git a/.github/workflows/dependency_test.yml b/.github/workflows/dependency_test.yml index e6c411bd5..98d818d88 100644 --- a/.github/workflows/dependency_test.yml +++ b/.github/workflows/dependency_test.yml @@ -20,3 +20,10 @@ jobs: - name: Run test run: | scripts/dependency_test.sh + - name: Archive logs + if: always() + uses: actions/upload-artifact@v2 + with: + name: buildfolder-${{ matrix.python-version }} + path: | + test_results_*.log diff --git a/setup.py b/setup.py index 93b62c527..e5551e48c 100755 --- a/setup.py +++ b/setup.py @@ -49,6 +49,7 @@ _dep_python_jose = "python-jose[cryptography]>=3.1.0,<4.0.0" _dep_python_jose_ecdsa_pin = ( "ecdsa<0.15" # https://github.com/spulec/moto/pull/3263#discussion_r477404984 ) +_dep_dataclasses = "dataclasses; python_version < '3.7'" _dep_docker = "docker>=2.5.1" _dep_jsondiff = "jsondiff>=1.1.2" _dep_aws_xray_sdk = "aws-xray-sdk!=0.96,>=0.93" @@ -87,7 +88,7 @@ extras_per_service.update( "ses": [], "sns": [], "sqs": [], - "ssm": [_dep_PyYAML], + "ssm": [_dep_PyYAML, _dep_dataclasses], # XRay module uses pkg_resources, but doesn't have an explicit dependency listed # This should be fixed in the next version: https://github.com/aws/aws-xray-sdk-python/issues/305 "xray": [_dep_aws_xray_sdk, _setuptools], diff --git a/tests/test_config/test_config_rules.py b/tests/test_config/test_config_rules.py index 1f9e19940..1601acb81 100644 --- a/tests/test_config/test_config_rules.py +++ b/tests/test_config/test_config_rules.py @@ -5,10 +5,8 @@ describe_config_rule delete_config_rule """ -from io import BytesIO import json from string import ascii_lowercase -from zipfile import ZipFile, ZIP_DEFLATED import boto3 from botocore.exceptions import ClientError @@ -17,8 +15,7 @@ import pytest from moto.config import mock_config from moto.config.models import random_string from moto.config.models import ConfigRule, CONFIG_RULE_PAGE_SIZE -from moto.core import ACCOUNT_ID -from moto import settings, mock_iam, mock_lambda +from moto import settings TEST_REGION = "us-east-1" if settings.TEST_SERVER_MODE else "us-west-2" @@ -37,71 +34,6 @@ def managed_config_rule(): } -def custom_config_rule(): - """Return a valid custom AWS Config Rule.""" - return { - "ConfigRuleName": f"custom_rule_{random_string()}", - "Description": "Custom S3 Public Read Prohibited Bucket Rule", - "Scope": {"ComplianceResourceTypes": ["AWS::S3::Bucket", "AWS::IAM::Group"]}, - "Source": { - "Owner": "CUSTOM_LAMBDA", - "SourceIdentifier": f"arn:aws:lambda:{TEST_REGION}:{ACCOUNT_ID}:function:test_config_rule", - "SourceDetails": [ - { - "EventSource": "aws.config", - "MessageType": "ScheduledNotification", - "MaximumExecutionFrequency": "Three_Hours", - }, - ], - }, - "MaximumExecutionFrequency": "One_Hour", - } - - -def zipped_lambda_function(): - """Return a simple test lambda function, zipped.""" - func_str = """ -def lambda_handler(event, context): - print("testing") - return event -""" - zip_output = BytesIO() - with ZipFile(zip_output, "w", ZIP_DEFLATED) as zip_file: - zip_file.writestr("lambda_function.py", func_str) - zip_file.close() - zip_output.seek(0) - return zip_output.read() - - -@mock_lambda -def create_lambda_for_config_rule(): - """Return the ARN of a lambda that can be used by a custom rule.""" - role_name = "test-role" - lambda_role = None - with mock_iam(): - iam_client = boto3.client("iam", region_name=TEST_REGION) - try: - lambda_role = iam_client.get_role(RoleName=role_name)["Role"]["Arn"] - except ClientError: - lambda_role = iam_client.create_role( - RoleName=role_name, AssumeRolePolicyDocument="test policy", Path="/", - )["Role"]["Arn"] - - # Create the lambda function and identify its location. - lambda_client = boto3.client("lambda", region_name=TEST_REGION) - lambda_client.create_function( - FunctionName="test_config_rule", - Runtime="python3.8", - Role=lambda_role, - Handler="lambda_function.lambda_handler", - Code={"ZipFile": zipped_lambda_function()}, - Description="Lambda test function for config rule", - Timeout=3, - MemorySize=128, - Publish=True, - ) - - @mock_config def test_put_config_rule_errors(): """Test various error conditions in put_config_rule API call.""" @@ -377,156 +309,6 @@ def test_config_rules_scope_errors(): # pylint: disable=too-many-statements assert "Scope cannot be applied to both resource and tag" in err["Message"] -@mock_config -def test_config_rules_source_errors(): # pylint: disable=too-many-statements - """Test various error conditions in ConfigRule.Source instantiation.""" - client = boto3.client("config", region_name=TEST_REGION) - - # Missing fields (ParamValidationError) caught by botocore and not - # tested here: ConfigRule.Source.SourceIdentifier - - managed_rule = managed_config_rule() - managed_rule["Source"]["Owner"] = "test" - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=managed_rule) - err = exc.value.response["Error"] - assert err["Code"] == "ValidationException" - assert "Member must satisfy enum value set: {AWS, CUSTOM_LAMBDA}" in err["Message"] - - managed_rule = managed_config_rule() - managed_rule["Source"]["SourceIdentifier"] = "test" - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=managed_rule) - err = exc.value.response["Error"] - assert err["Code"] == "InvalidParameterValueException" - assert ( - "The sourceIdentifier test is invalid. Please refer to the " - "documentation for a list of valid sourceIdentifiers that can be used " - "when AWS is the Owner" in err["Message"] - ) - - managed_rule = managed_config_rule() - managed_rule["Source"]["SourceDetails"] = [ - {"EventSource": "aws.config", "MessageType": "ScheduledNotification"} - ] - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=managed_rule) - err = exc.value.response["Error"] - assert err["Code"] == "InvalidParameterValueException" - assert ( - "SourceDetails should be null/empty if the owner is AWS. " - "SourceDetails should be provided if the owner is CUSTOM_LAMBDA" - in err["Message"] - ) - - custom_rule = custom_config_rule() - custom_rule["Source"] = { - "Owner": "CUSTOM_LAMBDA", - "SourceIdentifier": "test", - } - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "InvalidParameterValueException" - assert ( - "SourceDetails should be null/empty if the owner is AWS. " - "SourceDetails should be provided if the owner is CUSTOM_LAMBDA" - in err["Message"] - ) - - custom_rule = custom_config_rule() - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "InsufficientPermissionsException" - assert ( - f'The AWS Lambda function {custom_rule["Source"]["SourceIdentifier"]} ' - f"cannot be invoked. Check the specified function ARN, and check the " - f"function's permissions" in err["Message"] - ) - - -@mock_config -def test_config_rules_source_details_errors(): - """Test error conditions with ConfigRule.Source_Details instantiation.""" - client = boto3.client("config", region_name=TEST_REGION) - - create_lambda_for_config_rule() - - custom_rule = custom_config_rule() - custom_rule["Source"]["SourceDetails"][0] = {"MessageType": "ScheduledNotification"} - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "ParamValidationError" - assert ( - "Missing required parameter in ConfigRule.SourceDetails: 'EventSource'" - in err["Message"] - ) - - custom_rule = custom_config_rule() - custom_rule["Source"]["SourceDetails"][0]["EventSource"] = "foo" - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "ValidationException" - assert "Member must satisfy enum value set: {aws.config}" in err["Message"] - - custom_rule = custom_config_rule() - custom_rule["Source"]["SourceDetails"][0] = {"EventSource": "aws.config"} - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "ParamValidationError" - assert ( - "Missing required parameter in ConfigRule.SourceDetails: 'MessageType'" - in err["Message"] - ) - - custom_rule = custom_config_rule() - custom_rule["Source"]["SourceDetails"][0] = { - "MessageType": "foo", - "EventSource": "aws.config", - } - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "ValidationException" - assert ( - "Member must satisfy enum value set: " - "{ConfigurationItemChangeNotification, " - "ConfigurationSnapshotDeliveryCompleted, " - "OversizedConfigurationItemChangeNotification, ScheduledNotification}" - in err["Message"] - ) - - custom_rule = custom_config_rule() - custom_rule["Source"]["SourceDetails"][0]["MaximumExecutionFrequency"] = "foo" - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "ValidationException" - assert ( - "Member must satisfy enum value set: " - "{One_Hour, Six_Hours, Three_Hours, Twelve_Hours, TwentyFour_Hours}" - in err["Message"] - ) - - custom_rule = custom_config_rule() - custom_rule["Source"]["SourceDetails"][0][ - "MessageType" - ] = "ConfigurationItemChangeNotification" - with pytest.raises(ClientError) as exc: - client.put_config_rule(ConfigRule=custom_rule) - err = exc.value.response["Error"] - assert err["Code"] == "InvalidParameterValueException" - assert ( - "A maximum execution frequency is not allowed if MessageType " - "is ConfigurationItemChangeNotification or " - "OversizedConfigurationItemChangeNotification" in err["Message"] - ) - - @mock_config def test_valid_put_config_managed_rule(): """Test valid put_config_rule API calls for managed rules.""" @@ -565,74 +347,6 @@ def test_valid_put_config_managed_rule(): assert managed_rule_json == rsp_json -@mock_config -def test_valid_put_config_custom_rule(): - """Test valid put_config_rule API calls for custom rules.""" - client = boto3.client("config", region_name=TEST_REGION) - # Create custom rule and compare input against describe_config_rule - # output. - create_lambda_for_config_rule() - custom_rule = custom_config_rule() - custom_rule["Scope"]["ComplianceResourceTypes"] = ["AWS::IAM::Group"] - custom_rule["Scope"]["ComplianceResourceId"] = "basic_custom_test" - custom_rule["InputParameters"] = '{"TestName":"true"}' - custom_rule["ConfigRuleState"] = "ACTIVE" - client.put_config_rule(ConfigRule=custom_rule) - - rsp = client.describe_config_rules(ConfigRuleNames=[custom_rule["ConfigRuleName"]]) - custom_rule_json = json.dumps(custom_rule, sort_keys=True) - new_config_rule = rsp["ConfigRules"][0] - rule_arn = new_config_rule.pop("ConfigRuleArn") - rule_id = new_config_rule.pop("ConfigRuleId") - rsp_json = json.dumps(new_config_rule, sort_keys=True) - assert custom_rule_json == rsp_json - - # Update custom rule and compare again. - custom_rule["ConfigRuleArn"] = rule_arn - custom_rule["ConfigRuleId"] = rule_id - custom_rule["Description"] = "Updated Managed S3 Public Read Rule" - custom_rule["Scope"]["ComplianceResourceTypes"] = ["AWS::S3::Bucket"] - custom_rule["Scope"]["ComplianceResourceId"] = "S3-BUCKET_VERSIONING_ENABLED" - custom_rule["Source"]["SourceDetails"][0] = { - "EventSource": "aws.config", - "MessageType": "ConfigurationItemChangeNotification", - } - custom_rule["InputParameters"] = "{}" - client.put_config_rule(ConfigRule=custom_rule) - - rsp = client.describe_config_rules(ConfigRuleNames=[custom_rule["ConfigRuleName"]]) - custom_rule_json = json.dumps(custom_rule, sort_keys=True) - rsp_json = json.dumps(rsp["ConfigRules"][0], sort_keys=True) - assert custom_rule_json == rsp_json - - # Update a custom rule specifying just the rule Id. Test the default - # value for MaximumExecutionFrequency while we're at it. - del custom_rule["ConfigRuleArn"] - rule_name = custom_rule.pop("ConfigRuleName") - custom_rule["Source"]["SourceDetails"][0] = { - "EventSource": "aws.config", - "MessageType": "ConfigurationSnapshotDeliveryCompleted", - } - client.put_config_rule(ConfigRule=custom_rule) - rsp = client.describe_config_rules(ConfigRuleNames=[rule_name]) - updated_rule = rsp["ConfigRules"][0] - assert updated_rule["ConfigRuleName"] == rule_name - assert ( - updated_rule["Source"]["SourceDetails"][0]["MaximumExecutionFrequency"] - == "TwentyFour_Hours" - ) - - # Update a custom rule specifying just the rule ARN. - custom_rule["ConfigRuleArn"] = rule_arn - del custom_rule["ConfigRuleId"] - custom_rule["MaximumExecutionFrequency"] = "Six_Hours" - client.put_config_rule(ConfigRule=custom_rule) - rsp = client.describe_config_rules(ConfigRuleNames=[rule_name]) - updated_rule = rsp["ConfigRules"][0] - assert updated_rule["ConfigRuleName"] == rule_name - assert updated_rule["MaximumExecutionFrequency"] == "Six_Hours" - - @mock_config def test_describe_config_rules(): """Test the describe_config_rules API.""" diff --git a/tests/test_config/test_config_rules_integration.py b/tests/test_config/test_config_rules_integration.py new file mode 100644 index 000000000..5fc6effbb --- /dev/null +++ b/tests/test_config/test_config_rules_integration.py @@ -0,0 +1,294 @@ +from .test_config_rules import managed_config_rule, TEST_REGION +from botocore.exceptions import ClientError +from moto import mock_config, mock_iam, mock_lambda +from moto.core import ACCOUNT_ID +from io import BytesIO +from uuid import uuid4 +from zipfile import ZipFile, ZIP_DEFLATED + +import boto3 +import json +import pytest + + +def custom_config_rule(func_name="test_config_rule"): + """Return a valid custom AWS Config Rule.""" + return { + "ConfigRuleName": f"custom_rule_{uuid4()}[:6]", + "Description": "Custom S3 Public Read Prohibited Bucket Rule", + "Scope": {"ComplianceResourceTypes": ["AWS::S3::Bucket", "AWS::IAM::Group"]}, + "Source": { + "Owner": "CUSTOM_LAMBDA", + "SourceIdentifier": f"arn:aws:lambda:{TEST_REGION}:{ACCOUNT_ID}:function:{func_name}", + "SourceDetails": [ + { + "EventSource": "aws.config", + "MessageType": "ScheduledNotification", + "MaximumExecutionFrequency": "Three_Hours", + }, + ], + }, + "MaximumExecutionFrequency": "One_Hour", + } + + +def zipped_lambda_function(): + """Return a simple test lambda function, zipped.""" + func_str = """ +def lambda_handler(event, context): + print("testing") + return event +""" + zip_output = BytesIO() + with ZipFile(zip_output, "w", ZIP_DEFLATED) as zip_file: + zip_file.writestr("lambda_function.py", func_str) + zip_file.close() + zip_output.seek(0) + return zip_output.read() + + +@mock_lambda +def create_lambda_for_config_rule(): + """Return the ARN of a lambda that can be used by a custom rule.""" + role_name = "test-role" + lambda_role = None + with mock_iam(): + iam_client = boto3.client("iam", region_name=TEST_REGION) + try: + lambda_role = iam_client.get_role(RoleName=role_name)["Role"]["Arn"] + except ClientError: + lambda_role = iam_client.create_role( + RoleName=role_name, AssumeRolePolicyDocument="test policy", Path="/", + )["Role"]["Arn"] + + # Create the lambda function and identify its location. + lambda_client = boto3.client("lambda", region_name=TEST_REGION) + lambda_client.create_function( + FunctionName="test_config_rule", + Runtime="python3.8", + Role=lambda_role, + Handler="lambda_function.lambda_handler", + Code={"ZipFile": zipped_lambda_function()}, + Description="Lambda test function for config rule", + Timeout=3, + MemorySize=128, + Publish=True, + ) + + +@mock_config +def test_config_rules_source_details_errors(): + """Test error conditions with ConfigRule.Source_Details instantiation.""" + client = boto3.client("config", region_name=TEST_REGION) + + create_lambda_for_config_rule() + + custom_rule = custom_config_rule() + custom_rule["Source"]["SourceDetails"][0] = {"MessageType": "ScheduledNotification"} + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "ParamValidationError" + assert ( + "Missing required parameter in ConfigRule.SourceDetails: 'EventSource'" + in err["Message"] + ) + + custom_rule = custom_config_rule() + custom_rule["Source"]["SourceDetails"][0]["EventSource"] = "foo" + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "ValidationException" + assert "Member must satisfy enum value set: {aws.config}" in err["Message"] + + custom_rule = custom_config_rule() + custom_rule["Source"]["SourceDetails"][0] = {"EventSource": "aws.config"} + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "ParamValidationError" + assert ( + "Missing required parameter in ConfigRule.SourceDetails: 'MessageType'" + in err["Message"] + ) + + custom_rule = custom_config_rule() + custom_rule["Source"]["SourceDetails"][0] = { + "MessageType": "foo", + "EventSource": "aws.config", + } + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "ValidationException" + assert ( + "Member must satisfy enum value set: " + "{ConfigurationItemChangeNotification, " + "ConfigurationSnapshotDeliveryCompleted, " + "OversizedConfigurationItemChangeNotification, ScheduledNotification}" + in err["Message"] + ) + + custom_rule = custom_config_rule() + custom_rule["Source"]["SourceDetails"][0]["MaximumExecutionFrequency"] = "foo" + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "ValidationException" + assert ( + "Member must satisfy enum value set: " + "{One_Hour, Six_Hours, Three_Hours, Twelve_Hours, TwentyFour_Hours}" + in err["Message"] + ) + + custom_rule = custom_config_rule() + custom_rule["Source"]["SourceDetails"][0][ + "MessageType" + ] = "ConfigurationItemChangeNotification" + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidParameterValueException" + assert ( + "A maximum execution frequency is not allowed if MessageType " + "is ConfigurationItemChangeNotification or " + "OversizedConfigurationItemChangeNotification" in err["Message"] + ) + + +@mock_config +def test_valid_put_config_custom_rule(): + """Test valid put_config_rule API calls for custom rules.""" + client = boto3.client("config", region_name=TEST_REGION) + # Create custom rule and compare input against describe_config_rule + # output. + create_lambda_for_config_rule() + custom_rule = custom_config_rule() + custom_rule["Scope"]["ComplianceResourceTypes"] = ["AWS::IAM::Group"] + custom_rule["Scope"]["ComplianceResourceId"] = "basic_custom_test" + custom_rule["InputParameters"] = '{"TestName":"true"}' + custom_rule["ConfigRuleState"] = "ACTIVE" + client.put_config_rule(ConfigRule=custom_rule) + + rsp = client.describe_config_rules(ConfigRuleNames=[custom_rule["ConfigRuleName"]]) + custom_rule_json = json.dumps(custom_rule, sort_keys=True) + new_config_rule = rsp["ConfigRules"][0] + rule_arn = new_config_rule.pop("ConfigRuleArn") + rule_id = new_config_rule.pop("ConfigRuleId") + rsp_json = json.dumps(new_config_rule, sort_keys=True) + assert custom_rule_json == rsp_json + + # Update custom rule and compare again. + custom_rule["ConfigRuleArn"] = rule_arn + custom_rule["ConfigRuleId"] = rule_id + custom_rule["Description"] = "Updated Managed S3 Public Read Rule" + custom_rule["Scope"]["ComplianceResourceTypes"] = ["AWS::S3::Bucket"] + custom_rule["Scope"]["ComplianceResourceId"] = "S3-BUCKET_VERSIONING_ENABLED" + custom_rule["Source"]["SourceDetails"][0] = { + "EventSource": "aws.config", + "MessageType": "ConfigurationItemChangeNotification", + } + custom_rule["InputParameters"] = "{}" + client.put_config_rule(ConfigRule=custom_rule) + + rsp = client.describe_config_rules(ConfigRuleNames=[custom_rule["ConfigRuleName"]]) + custom_rule_json = json.dumps(custom_rule, sort_keys=True) + rsp_json = json.dumps(rsp["ConfigRules"][0], sort_keys=True) + assert custom_rule_json == rsp_json + + # Update a custom rule specifying just the rule Id. Test the default + # value for MaximumExecutionFrequency while we're at it. + del custom_rule["ConfigRuleArn"] + rule_name = custom_rule.pop("ConfigRuleName") + custom_rule["Source"]["SourceDetails"][0] = { + "EventSource": "aws.config", + "MessageType": "ConfigurationSnapshotDeliveryCompleted", + } + client.put_config_rule(ConfigRule=custom_rule) + rsp = client.describe_config_rules(ConfigRuleNames=[rule_name]) + updated_rule = rsp["ConfigRules"][0] + assert updated_rule["ConfigRuleName"] == rule_name + assert ( + updated_rule["Source"]["SourceDetails"][0]["MaximumExecutionFrequency"] + == "TwentyFour_Hours" + ) + + # Update a custom rule specifying just the rule ARN. + custom_rule["ConfigRuleArn"] = rule_arn + del custom_rule["ConfigRuleId"] + custom_rule["MaximumExecutionFrequency"] = "Six_Hours" + client.put_config_rule(ConfigRule=custom_rule) + rsp = client.describe_config_rules(ConfigRuleNames=[rule_name]) + updated_rule = rsp["ConfigRules"][0] + assert updated_rule["ConfigRuleName"] == rule_name + assert updated_rule["MaximumExecutionFrequency"] == "Six_Hours" + + +@mock_config +def test_config_rules_source_errors(): # pylint: disable=too-many-statements + """Test various error conditions in ConfigRule.Source instantiation.""" + client = boto3.client("config", region_name=TEST_REGION) + + # Missing fields (ParamValidationError) caught by botocore and not + # tested here: ConfigRule.Source.SourceIdentifier + + managed_rule = managed_config_rule() + managed_rule["Source"]["Owner"] = "test" + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=managed_rule) + err = exc.value.response["Error"] + assert err["Code"] == "ValidationException" + assert "Member must satisfy enum value set: {AWS, CUSTOM_LAMBDA}" in err["Message"] + + managed_rule = managed_config_rule() + managed_rule["Source"]["SourceIdentifier"] = "test" + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=managed_rule) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidParameterValueException" + assert ( + "The sourceIdentifier test is invalid. Please refer to the " + "documentation for a list of valid sourceIdentifiers that can be used " + "when AWS is the Owner" in err["Message"] + ) + + managed_rule = managed_config_rule() + managed_rule["Source"]["SourceDetails"] = [ + {"EventSource": "aws.config", "MessageType": "ScheduledNotification"} + ] + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=managed_rule) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidParameterValueException" + assert ( + "SourceDetails should be null/empty if the owner is AWS. " + "SourceDetails should be provided if the owner is CUSTOM_LAMBDA" + in err["Message"] + ) + + custom_rule = custom_config_rule() + custom_rule["Source"] = { + "Owner": "CUSTOM_LAMBDA", + "SourceIdentifier": "test", + } + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidParameterValueException" + assert ( + "SourceDetails should be null/empty if the owner is AWS. " + "SourceDetails should be provided if the owner is CUSTOM_LAMBDA" + in err["Message"] + ) + + custom_rule = custom_config_rule(func_name="unknown_func") + with pytest.raises(ClientError) as exc: + client.put_config_rule(ConfigRule=custom_rule) + err = exc.value.response["Error"] + assert err["Code"] == "InsufficientPermissionsException" + assert ( + f'The AWS Lambda function {custom_rule["Source"]["SourceIdentifier"]} ' + f"cannot be invoked. Check the specified function ARN, and check the " + f"function's permissions" in err["Message"] + )