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

Merge branch 'master' into cname_01

This commit is contained in:
Chad S 2020-10-29 14:01:14 -07:00 committed by GitHub
parent a1f99c29c0 aa2e0aa2f9
commit 14348a1f95
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):
sns_client = boto3.client("sns", **kwargs)
message_ids = {}
subject = "Lemur: {0} Notification".format(notification_type.capitalize())
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
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(
TopicArn=topic_arn,
Message=format_message(certificate, notification_type),
Subject=subject,
)
response_code = response["ResponseMetadata"]["HTTPStatusCode"]
@ -48,8 +50,9 @@ def format_message(certificate, notification_type):
json_message = {
"notification_type": notification_type,
"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"]),
"owner": certificate["owner"],
"details": create_certificate_url(certificate["name"])
}
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 = {
"notification_type": "expiration",
"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,
"owner": certificate["owner"],
"details": "https://lemur.example.com/#/certificates/{name}".format(name=certificate["name"])
}
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"]]
actual_message = next(
(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():