darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-18 03:31:42 -05:00
Code Issues Packages Projects Releases Wiki Activity

notify-api-360 stop worrying about aws regions

Browse Source
This commit is contained in:
Kenneth Kehl
2023-09-27 08:57:00 -07:00
parent 2bd4945847
commit f1a4516c9e
2 changed files with 18 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
app/clients/cloudwatch/aws_cloudwatch.py
View File

@@ -80,28 +80,16 @@ class AwsCloudwatchClient(Client):
break break
return all_log_events return all_log_events
def _extract_account_number(self, ses_domain_arn, region): def _extract_account_number(self, ses_domain_arn):
account_number = ses_domain_arn account_number = ses_domain_arn.split(":")
# handle cloud.gov case
if "aws-us-gov" in account_number:
account_number = account_number.replace(f"arn:aws-us-gov:ses:{region}:", "")
account_number = account_number.split(":")
account_number = account_number[0]
# handle staging case
else:
account_number = account_number.replace(f"arn:aws:ses:{region}:", "")
account_number = account_number.split(":")
account_number = account_number[0]
return account_number return account_number
def check_sms(self, message_id, notification_id, created_at): def check_sms(self, message_id, notification_id, created_at):
region = cloud_config.sns_region region = cloud_config.sns_region
# TODO this clumsy approach to getting the account number will be fixed as part of notify-api #258 # TODO this clumsy approach to getting the account number will be fixed as part of notify-api #258
account_number = self._extract_account_number( account_number = self._extract_account_number(cloud_config.ses_domain_arn)
cloud_config.ses_domain_arn, region
)
log_group_name = f"sns/{region}/{account_number}/DirectPublishToPhoneNumber" log_group_name = f"sns/{region}/{account_number[4]}/DirectPublishToPhoneNumber"
current_app.logger.info( current_app.logger.info(
f"Log group name: {log_group_name} message id: {message_id}" f"Log group name: {log_group_name} message id: {message_id}"
) )

25
tests/app/clients/test_aws_cloudwatch.py
View File

@@ -54,7 +54,10 @@ def test_check_sms_success(notify_api, mocker):
aws_cloudwatch_client.init_app(current_app) aws_cloudwatch_client.init_app(current_app)
boto_mock = mocker.patch.object(aws_cloudwatch_client, "_client", create=True) boto_mock = mocker.patch.object(aws_cloudwatch_client, "_client", create=True)
boto_mock.filter_log_events.side_effect = side_effect boto_mock.filter_log_events.side_effect = side_effect
mocker.patch.dict("os.environ", {"SES_DOMAIN_ARN": "1111:"}) mocker.patch.dict(
"os.environ",
{"SES_DOMAIN_ARN": "arn:aws:ses:us-west-2:12345:identity/ses-xxx.xxx.xxx.xxx"},
)
message_id = "succeed" message_id = "succeed"
notification_id = "ccc" notification_id = "ccc"
@@ -73,8 +76,10 @@ def test_check_sms_failure(notify_api, mocker):
aws_cloudwatch_client.init_app(current_app) aws_cloudwatch_client.init_app(current_app)
boto_mock = mocker.patch.object(aws_cloudwatch_client, "_client", create=True) boto_mock = mocker.patch.object(aws_cloudwatch_client, "_client", create=True)
boto_mock.filter_log_events.side_effect = side_effect boto_mock.filter_log_events.side_effect = side_effect
mocker.patch.dict("os.environ", {"SES_DOMAIN_ARN": "1111:"}) mocker.patch.dict(
"os.environ",
{"SES_DOMAIN_ARN": "arn:aws:ses:us-west-2:12345:identity/ses-xxx.xxx.xxx.xxx"},
)
message_id = "fail" message_id = "fail"
notification_id = "bbb" notification_id = "bbb"
with notify_api.app_context(): with notify_api.app_context():
@@ -90,17 +95,15 @@ def test_check_sms_failure(notify_api, mocker):
def test_extract_account_number_gov_cloud(): def test_extract_account_number_gov_cloud():
domain_arn = "arn:aws-us-gov:ses:us-gov-west-1:12345:identity/ses-abc.xxx.xxx.xxx" domain_arn = "arn:aws-us-gov:ses:us-gov-west-1:12345:identity/ses-abc.xxx.xxx.xxx"
actual_account_number = aws_cloudwatch_client._extract_account_number( actual_account_number = aws_cloudwatch_client._extract_account_number(domain_arn)
domain_arn, "us-gov-west-1" assert len(actual_account_number) == 6
)
expected_account_number = "12345" expected_account_number = "12345"
assert actual_account_number == expected_account_number assert actual_account_number[4] == expected_account_number
def test_extract_account_number_gov_staging(): def test_extract_account_number_gov_staging():
domain_arn = "arn:aws:ses:us-south-14:12345:identity/ses-abc.xxx.xxx.xxx" domain_arn = "arn:aws:ses:us-south-14:12345:identity/ses-abc.xxx.xxx.xxx"
actual_account_number = aws_cloudwatch_client._extract_account_number( actual_account_number = aws_cloudwatch_client._extract_account_number(domain_arn)
domain_arn, "us-south-14" assert len(actual_account_number) == 6
)
expected_account_number = "12345" expected_account_number = "12345"
assert actual_account_number == expected_account_number assert actual_account_number[4] == expected_account_number