Infra
/
lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge branch 'master' into ses-arn-override

This commit is contained in:
Hossein Shafagh 2020-10-29 13:52:51 -07:00 committed by GitHub
parent 28686fcf5d aa2e0aa2f9
commit 28c6f8583a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 11 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
lemur/plugins/lemur_aws/sns.py
View File

@ -15,16 +15,18 @@ from flask import current_app
def publish(topic_arn, certificates, notification_type, **kwargs): def publish(topic_arn, certificates, notification_type, **kwargs):
sns_client = boto3.client("sns", **kwargs) sns_client = boto3.client("sns", **kwargs)
message_ids = {} message_ids = {}
subject = "Lemur: {0} Notification".format(notification_type.capitalize())
for certificate in certificates: for certificate in certificates:
message_ids[certificate["name"]] = publish_single(sns_client, topic_arn, certificate, notification_type) message_ids[certificate["name"]] = publish_single(sns_client, topic_arn, certificate, notification_type, subject)
return message_ids return message_ids
def publish_single(sns_client, topic_arn, certificate, notification_type): def publish_single(sns_client, topic_arn, certificate, notification_type, subject):
response = sns_client.publish( response = sns_client.publish(
TopicArn=topic_arn, TopicArn=topic_arn,
Message=format_message(certificate, notification_type), Message=format_message(certificate, notification_type),
Subject=subject,
) )
response_code = response["ResponseMetadata"]["HTTPStatusCode"] response_code = response["ResponseMetadata"]["HTTPStatusCode"]
@ -48,8 +50,9 @@ def format_message(certificate, notification_type):
json_message = { json_message = {
"notification_type": notification_type, "notification_type": notification_type,
"certificate_name": certificate["name"], "certificate_name": certificate["name"],
"expires": arrow.get(certificate["validityEnd"]).format("YYYY-MM-ddTHH:mm:ss"), # 2047-12-T22:00:00 "expires": arrow.get(certificate["validityEnd"]).format("YYYY-MM-DDTHH:mm:ss"), # 2047-12-31T22:00:00
"endpoints_detected": len(certificate["endpoints"]), "endpoints_detected": len(certificate["endpoints"]),
"owner": certificate["owner"],
"details": create_certificate_url(certificate["name"]) "details": create_certificate_url(certificate["name"])
} }
return json.dumps(json_message) return json.dumps(json_message)

7
lemur/plugins/lemur_aws/tests/test_sns.py
View File

@ -20,8 +20,9 @@ def test_format(certificate, endpoint):
expected_message = { expected_message = {
"notification_type": "expiration", "notification_type": "expiration",
"certificate_name": certificate["name"], "certificate_name": certificate["name"],
"expires": arrow.get(certificate["validityEnd"]).format("YYYY-MM-ddTHH:mm:ss"), "expires": arrow.get(certificate["validityEnd"]).format("YYYY-MM-DDTHH:mm:ss"),
"endpoints_detected": 0, "endpoints_detected": 0,
"owner": certificate["owner"],
"details": "https://lemur.example.com/#/certificates/{name}".format(name=certificate["name"]) "details": "https://lemur.example.com/#/certificates/{name}".format(name=certificate["name"])
} }
assert expected_message == json.loads(format_message(certificate, "expiration")) assert expected_message == json.loads(format_message(certificate, "expiration"))
@ -57,7 +58,9 @@ def test_publish(certificate, endpoint):
expected_message_id = message_ids[certificate["name"]] expected_message_id = message_ids[certificate["name"]]
actual_message = next( actual_message = next(
(m for m in received_messages if json.loads(m["Body"])["MessageId"] == expected_message_id), None) (m for m in received_messages if json.loads(m["Body"])["MessageId"] == expected_message_id), None)
assert json.loads(actual_message["Body"])["Message"] == format_message(certificate, "expiration") actual_json = json.loads(actual_message["Body"])
assert actual_json["Message"] == format_message(certificate, "expiration")
assert actual_json["Subject"] == "Lemur: Expiration Notification"
def get_options(): def get_options():