Merge branch 'master' into log_update
This commit is contained in:
Hossein Shafagh
GitHub
commit
adf8f37718
|
|
@ -340,6 +340,8 @@ class CertificateOutputSchema(LemurOutputSchema):
|
|||
|
||||
@post_dump
|
||||
def handle_subject_details(self, data):
|
||||
subject_details = ["country", "state", "location", "organization", "organizational_unit"]
|
||||
|
||||
# Remove subject details if authority is CA/Browser Forum compliant. The code will use default set of values in that case.
|
||||
# If CA/Browser Forum compliance of an authority is unknown (None), it is safe to fallback to default values. Thus below
|
||||
# condition checks for 'not False' ==> 'True or None'
|
||||
|
|
@ -347,11 +349,13 @@ class CertificateOutputSchema(LemurOutputSchema):
|
|||
is_cab_compliant = data.get("authority").get("isCabCompliant")
|
||||
|
||||
if is_cab_compliant is not False:
|
||||
data.pop("country", None)
|
||||
data.pop("state", None)
|
||||
data.pop("location", None)
|
||||
data.pop("organization", None)
|
||||
data.pop("organizational_unit", None)
|
||||
for field in subject_details:
|
||||
data.pop(field, None)
|
||||
|
||||
# Removing subject fields if None, else it complains in de-serialization
|
||||
for field in subject_details:
|
||||
if field in data and data[field] is None:
|
||||
data.pop(field)
|
||||
|
||||
|
||||
class CertificateShortOutputSchema(LemurOutputSchema):
|
||||
|
|
|
|||
|
|
@ -110,9 +110,11 @@ def organizational_unit(cert):
|
|||
:return:
|
||||
"""
|
||||
try:
|
||||
return cert.subject.get_attributes_for_oid(x509.OID_ORGANIZATIONAL_UNIT_NAME)[
|
||||
0
|
||||
].value.strip()
|
||||
ou = cert.subject.get_attributes_for_oid(x509.OID_ORGANIZATIONAL_UNIT_NAME)
|
||||
if not ou:
|
||||
return None
|
||||
|
||||
return ou[0].value.strip()
|
||||
except Exception as e:
|
||||
sentry.captureException()
|
||||
current_app.logger.error("Unable to get organizational unit! {0}".format(e))
|
||||
|
|
@ -155,9 +157,11 @@ def location(cert):
|
|||
:return:
|
||||
"""
|
||||
try:
|
||||
return cert.subject.get_attributes_for_oid(x509.OID_LOCALITY_NAME)[
|
||||
0
|
||||
].value.strip()
|
||||
loc = cert.subject.get_attributes_for_oid(x509.OID_LOCALITY_NAME)
|
||||
if not loc:
|
||||
return None
|
||||
|
||||
return loc[0].value.strip()
|
||||
except Exception as e:
|
||||
sentry.captureException()
|
||||
current_app.logger.error("Unable to get location! {0}".format(e))
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@
|
|||
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
|
||||
|
||||
"""
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from datetime import timedelta
|
||||
from itertools import groupby
|
||||
|
|
@ -96,11 +97,20 @@ def send_notification(event_type, data, targets, notification):
|
|||
:param notification:
|
||||
:return:
|
||||
"""
|
||||
function = f"{__name__}.{sys._getframe().f_code.co_name}"
|
||||
log_data = {
|
||||
"function": function,
|
||||
"message": f"Sending expiration notification for to targets {targets}",
|
||||
"notification_type": "expiration",
|
||||
"certificate_targets": targets,
|
||||
}
|
||||
status = FAILURE_METRIC_STATUS
|
||||
try:
|
||||
notification.plugin.send(event_type, data, targets, notification.options)
|
||||
status = SUCCESS_METRIC_STATUS
|
||||
except Exception as e:
|
||||
log_data["message"] = f"Unable to send expiration notification to targets {targets}"
|
||||
current_app.logger.error(log_data, exc_info=True)
|
||||
sentry.captureException()
|
||||
|
||||
metrics.send(
|
||||
|
|
@ -187,21 +197,29 @@ def send_rotation_notification(certificate, notification_plugin=None):
|
|||
:param notification_plugin:
|
||||
:return:
|
||||
"""
|
||||
function = f"{__name__}.{sys._getframe().f_code.co_name}"
|
||||
log_data = {
|
||||
"function": function,
|
||||
"message": f"Sending rotation notification for certificate {certificate.name}",
|
||||
"notification_type": "rotation",
|
||||
"certificate_name": certificate.name,
|
||||
"certificate_owner": certificate.owner,
|
||||
}
|
||||
status = FAILURE_METRIC_STATUS
|
||||
if not notification_plugin:
|
||||
notification_plugin = plugins.get(
|
||||
current_app.config.get("LEMUR_DEFAULT_NOTIFICATION_PLUGIN")
|
||||
current_app.config.get("LEMUR_DEFAULT_NOTIFICATION_PLUGIN", "email-notification")
|
||||
)
|
||||
|
||||
data = certificate_notification_output_schema.dump(certificate).data
|
||||
|
||||
try:
|
||||
notification_plugin.send("rotation", data, [data["owner"]])
|
||||
notification_plugin.send("rotation", data, [data["owner"]], [])
|
||||
status = SUCCESS_METRIC_STATUS
|
||||
except Exception as e:
|
||||
current_app.logger.error(
|
||||
"Unable to send notification to {}.".format(data["owner"]), exc_info=True
|
||||
)
|
||||
log_data["message"] = f"Unable to send rotation notification for certificate {certificate.name} " \
|
||||
f"to owner {data['owner']}"
|
||||
current_app.logger.error(log_data, exc_info=True)
|
||||
sentry.captureException()
|
||||
|
||||
metrics.send(
|
||||
|
|
@ -225,6 +243,14 @@ def send_pending_failure_notification(
|
|||
:param notification_plugin:
|
||||
:return:
|
||||
"""
|
||||
function = f"{__name__}.{sys._getframe().f_code.co_name}"
|
||||
log_data = {
|
||||
"function": function,
|
||||
"message": f"Sending pending failure notification for pending certificate {pending_cert}",
|
||||
"notification_type": "failed",
|
||||
"certificate_name": pending_cert.name,
|
||||
"certificate_owner": pending_cert.owner,
|
||||
}
|
||||
status = FAILURE_METRIC_STATUS
|
||||
|
||||
if not notification_plugin:
|
||||
|
|
@ -242,12 +268,10 @@ def send_pending_failure_notification(
|
|||
notification_plugin.send("failed", data, [data["owner"]], pending_cert)
|
||||
status = SUCCESS_METRIC_STATUS
|
||||
except Exception as e:
|
||||
current_app.logger.error(
|
||||
"Unable to send pending failure notification to {}.".format(
|
||||
data["owner"]
|
||||
),
|
||||
exc_info=True,
|
||||
)
|
||||
log_data["recipient"] = data["owner"]
|
||||
log_data["message"] = f"Unable to send pending failure notification for certificate {pending_cert.name} " \
|
||||
f"to owner {pending_cert.owner}"
|
||||
current_app.logger.error(log_data, exc_info=True)
|
||||
sentry.captureException()
|
||||
|
||||
if notify_security:
|
||||
|
|
@ -257,18 +281,17 @@ def send_pending_failure_notification(
|
|||
)
|
||||
status = SUCCESS_METRIC_STATUS
|
||||
except Exception as e:
|
||||
current_app.logger.error(
|
||||
"Unable to send pending failure notification to "
|
||||
"{}.".format(data["security_email"]),
|
||||
exc_info=True,
|
||||
)
|
||||
log_data["recipient"] = data["security_email"]
|
||||
log_data["message"] = f"Unable to send pending failure notification for certificate {pending_cert.name} " \
|
||||
f"to security email {pending_cert.owner}"
|
||||
current_app.logger.error(log_data, exc_info=True)
|
||||
sentry.captureException()
|
||||
|
||||
metrics.send(
|
||||
"notification",
|
||||
"counter",
|
||||
1,
|
||||
metric_tags={"status": status, "event_type": "rotation"},
|
||||
metric_tags={"status": status, "event_type": "failed"},
|
||||
)
|
||||
|
||||
if status == SUCCESS_METRIC_STATUS:
|
||||
|
|
@ -290,7 +313,7 @@ def needs_notification(certificate):
|
|||
|
||||
for notification in certificate.notifications:
|
||||
if not notification.active or not notification.options:
|
||||
return
|
||||
continue
|
||||
|
||||
interval = get_plugin_option("interval", notification.options)
|
||||
unit = get_plugin_option("unit", notification.options)
|
||||
|
|
@ -306,7 +329,7 @@ def needs_notification(certificate):
|
|||
|
||||
else:
|
||||
raise Exception(
|
||||
"Invalid base unit for expiration interval: {0}".format(unit)
|
||||
f"Invalid base unit for expiration interval: {unit}"
|
||||
)
|
||||
|
||||
if days == interval:
|
||||
|
|
|
|||
|
|
@ -19,14 +19,16 @@ from lemur.plugins import lemur_email as email
|
|||
from lemur.plugins.lemur_email.templates.config import env
|
||||
|
||||
|
||||
def render_html(template_name, message):
|
||||
def render_html(template_name, options, certificates):
|
||||
"""
|
||||
Renders the html for our email notification.
|
||||
|
||||
:param template_name:
|
||||
:param message:
|
||||
:param options:
|
||||
:param certificates:
|
||||
:return:
|
||||
"""
|
||||
message = {"options": options, "certificates": certificates}
|
||||
template = env.get_template("{}.html".format(template_name))
|
||||
return template.render(
|
||||
dict(message=message, hostname=current_app.config.get("LEMUR_HOSTNAME"))
|
||||
|
|
@ -100,8 +102,7 @@ class EmailNotificationPlugin(ExpirationNotificationPlugin):
|
|||
|
||||
subject = "Lemur: {0} Notification".format(notification_type.capitalize())
|
||||
|
||||
data = {"options": options, "certificates": message}
|
||||
body = render_html(notification_type, data)
|
||||
body = render_html(notification_type, options, message)
|
||||
|
||||
s_type = current_app.config.get("LEMUR_EMAIL_SENDER", "ses").lower()
|
||||
|
||||
|
|
|
|||
|
|
@ -83,12 +83,12 @@
|
|||
<td width="32px"></td>
|
||||
<td width="16px"></td>
|
||||
<td style="line-height:1.2">
|
||||
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ certificate.name }}</span>
|
||||
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ message.certificates.name }}</span>
|
||||
<br>
|
||||
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:13px;color:#727272">
|
||||
<br>{{ certificate.owner }}
|
||||
<br>{{ certificate.validityEnd | time }}
|
||||
<a href="https://{{ hostname }}/#/certificates/{{ certificate.name }}" target="_blank">Details</a>
|
||||
<br>{{ message.certificates.owner }}
|
||||
<br>{{ message.certificates.validityEnd | time }}
|
||||
<a href="https://{{ hostname }}/#/certificates/{{ message.certificates.name }}" target="_blank">Details</a>
|
||||
</span>
|
||||
</td>
|
||||
</tr>
|
||||
|
|
@ -110,12 +110,12 @@
|
|||
<td width="32px"></td>
|
||||
<td width="16px"></td>
|
||||
<td style="line-height:1.2">
|
||||
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ certificate.replacedBy[0].name }}</span>
|
||||
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:20px;color:#202020">{{ message.certificates.name }}</span>
|
||||
<br>
|
||||
<span style="font-family:Roboto-Regular,Helvetica,Arial,sans-serif;font-size:13px;color:#727272">
|
||||
<br>{{ certificate.replacedBy[0].owner }}
|
||||
<br>{{ certificate.replacedBy[0].validityEnd | time }}
|
||||
<a href="https://{{ hostname }}/#/certificates/{{ certificate.replacedBy[0].name }}" target="_blank">Details</a>
|
||||
<br>{{ message.certificates.owner }}
|
||||
<br>{{ message.certificates.validityEnd | time }}
|
||||
<a href="https://{{ hostname }}/#/certificates/{{ message.certificates.name }}" target="_blank">Details</a>
|
||||
</span>
|
||||
</td>
|
||||
</tr>
|
||||
|
|
@ -133,7 +133,7 @@
|
|||
<table border="0" cellspacing="0" cellpadding="0"
|
||||
style="margin-top:48px;margin-bottom:48px">
|
||||
<tbody>
|
||||
{% for endpoint in certificate.endpoints %}
|
||||
{% for endpoint in message.certificates.endpoints %}
|
||||
<tr valign="middle">
|
||||
<td width="32px"></td>
|
||||
<td width="16px"></td>
|
||||
|
|
|
|||
|
|
@ -1,36 +1,83 @@
|
|||
import os
|
||||
from lemur.plugins.lemur_email.templates.config import env
|
||||
from datetime import timedelta
|
||||
|
||||
import arrow
|
||||
import boto3
|
||||
from moto import mock_ses
|
||||
|
||||
from lemur.certificates.schemas import certificate_notification_output_schema
|
||||
from lemur.plugins.lemur_email.plugin import render_html
|
||||
from lemur.tests.factories import CertificateFactory
|
||||
|
||||
dir_path = os.path.dirname(os.path.realpath(__file__))
|
||||
|
||||
|
||||
def test_render(certificate, endpoint):
|
||||
from lemur.certificates.schemas import certificate_notification_output_schema
|
||||
@mock_ses
|
||||
def verify_sender_email():
|
||||
ses_client = boto3.client("ses", region_name="us-east-1")
|
||||
ses_client.verify_email_identity(EmailAddress="lemur@example.com")
|
||||
|
||||
|
||||
def get_options():
|
||||
return [
|
||||
{"name": "interval", "value": 10},
|
||||
{"name": "unit", "value": "days"},
|
||||
]
|
||||
|
||||
|
||||
def test_render_expiration(certificate, endpoint):
|
||||
|
||||
new_cert = CertificateFactory()
|
||||
new_cert.replaces.append(certificate)
|
||||
|
||||
data = {
|
||||
"certificates": [certificate_notification_output_schema.dump(certificate).data],
|
||||
"options": [
|
||||
{"name": "interval", "value": 10},
|
||||
{"name": "unit", "value": "days"},
|
||||
],
|
||||
}
|
||||
assert render_html("expiration", get_options(), [certificate_notification_output_schema.dump(certificate).data])
|
||||
|
||||
template = env.get_template("{}.html".format("expiration"))
|
||||
|
||||
body = template.render(dict(message=data, hostname="lemur.test.example.com"))
|
||||
|
||||
template = env.get_template("{}.html".format("rotation"))
|
||||
|
||||
def test_render_rotation(certificate, endpoint):
|
||||
certificate.endpoints.append(endpoint)
|
||||
|
||||
body = template.render(
|
||||
dict(
|
||||
certificate=certificate_notification_output_schema.dump(certificate).data,
|
||||
hostname="lemur.test.example.com",
|
||||
)
|
||||
)
|
||||
assert render_html("rotation", get_options(), certificate_notification_output_schema.dump(certificate).data)
|
||||
|
||||
|
||||
def test_render_rotation_failure(pending_certificate):
|
||||
assert render_html("failed", get_options(), certificate_notification_output_schema.dump(pending_certificate).data)
|
||||
|
||||
|
||||
@mock_ses
|
||||
def test_send_expiration_notification():
|
||||
from lemur.notifications.messaging import send_expiration_notifications
|
||||
from lemur.tests.factories import CertificateFactory
|
||||
from lemur.tests.factories import NotificationFactory
|
||||
|
||||
now = arrow.utcnow()
|
||||
in_ten_days = now + timedelta(days=10, hours=1) # a bit more than 10 days since we'll check in the future
|
||||
certificate = CertificateFactory()
|
||||
notification = NotificationFactory(plugin_name="email-notification")
|
||||
|
||||
certificate.not_after = in_ten_days
|
||||
certificate.notifications.append(notification)
|
||||
certificate.notifications[0].options = get_options()
|
||||
|
||||
verify_sender_email()
|
||||
assert send_expiration_notifications([]) == (2, 0)
|
||||
|
||||
|
||||
@mock_ses
|
||||
def test_send_rotation_notification(endpoint, source_plugin):
|
||||
from lemur.notifications.messaging import send_rotation_notification
|
||||
from lemur.deployment.service import rotate_certificate
|
||||
|
||||
new_certificate = CertificateFactory()
|
||||
rotate_certificate(endpoint, new_certificate)
|
||||
assert endpoint.certificate == new_certificate
|
||||
|
||||
verify_sender_email()
|
||||
assert send_rotation_notification(new_certificate)
|
||||
|
||||
|
||||
@mock_ses
|
||||
def test_send_pending_failure_notification(user, pending_certificate, async_issuer_plugin):
|
||||
from lemur.notifications.messaging import send_pending_failure_notification
|
||||
|
||||
verify_sender_email()
|
||||
assert send_pending_failure_notification(pending_certificate)
|
||||
|
|
|
|||
|
|
@ -58,7 +58,6 @@ def create_rotation_attachments(certificate):
|
|||
"title": certificate["name"],
|
||||
"title_link": create_certificate_url(certificate["name"]),
|
||||
"fields": [
|
||||
{
|
||||
{"title": "Owner", "value": certificate["owner"], "short": True},
|
||||
{
|
||||
"title": "Expires",
|
||||
|
|
@ -67,17 +66,11 @@ def create_rotation_attachments(certificate):
|
|||
),
|
||||
"short": True,
|
||||
},
|
||||
{
|
||||
"title": "Replaced By",
|
||||
"value": len(certificate["replaced"][0]["name"]),
|
||||
"short": True,
|
||||
},
|
||||
{
|
||||
"title": "Endpoints Rotated",
|
||||
"value": len(certificate["endpoints"]),
|
||||
"short": True,
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,10 @@
|
|||
from datetime import timedelta
|
||||
|
||||
import arrow
|
||||
|
||||
from lemur.tests.factories import NotificationFactory, CertificateFactory
|
||||
|
||||
|
||||
def test_formatting(certificate):
|
||||
from lemur.plugins.lemur_slack.plugin import create_expiration_attachments
|
||||
from lemur.certificates.schemas import certificate_notification_output_schema
|
||||
|
|
@ -21,3 +28,49 @@ def test_formatting(certificate):
|
|||
}
|
||||
|
||||
assert attachment == create_expiration_attachments(data)[0]
|
||||
|
||||
|
||||
def get_options():
|
||||
return [
|
||||
{"name": "interval", "value": 10},
|
||||
{"name": "unit", "value": "days"},
|
||||
{"name": "webhook", "value": "https://slack.com/api/api.test"},
|
||||
]
|
||||
|
||||
|
||||
def test_send_expiration_notification():
|
||||
from lemur.notifications.messaging import send_expiration_notifications
|
||||
|
||||
notification = NotificationFactory(plugin_name="slack-notification")
|
||||
notification.options = get_options()
|
||||
|
||||
now = arrow.utcnow()
|
||||
in_ten_days = now + timedelta(days=10, hours=1) # a bit more than 10 days since we'll check in the future
|
||||
|
||||
certificate = CertificateFactory()
|
||||
certificate.not_after = in_ten_days
|
||||
certificate.notifications.append(notification)
|
||||
|
||||
assert send_expiration_notifications([]) == (2, 0)
|
||||
|
||||
|
||||
# Currently disabled as the Slack plugin doesn't support this type of notification
|
||||
# def test_send_rotation_notification(endpoint, source_plugin):
|
||||
# from lemur.notifications.messaging import send_rotation_notification
|
||||
# from lemur.deployment.service import rotate_certificate
|
||||
#
|
||||
# notification = NotificationFactory(plugin_name="slack-notification")
|
||||
# notification.options = get_options()
|
||||
#
|
||||
# new_certificate = CertificateFactory()
|
||||
# rotate_certificate(endpoint, new_certificate)
|
||||
# assert endpoint.certificate == new_certificate
|
||||
#
|
||||
# assert send_rotation_notification(new_certificate, notification_plugin=notification.plugin)
|
||||
|
||||
|
||||
# Currently disabled as the Slack plugin doesn't support this type of notification
|
||||
# def test_send_pending_failure_notification(user, pending_certificate, async_issuer_plugin):
|
||||
# from lemur.notifications.messaging import send_pending_failure_notification
|
||||
#
|
||||
# assert send_pending_failure_notification(pending_certificate, notification_plugin=plugins.get("slack-notification"))
|
||||
|
|
|
|||
|
|
@ -20,8 +20,16 @@
|
|||
Key Type
|
||||
</label>
|
||||
<div class="col-sm-10">
|
||||
<select class="form-control" ng-model="authority.keyType" ng-options="option for option in ['RSA2048', 'RSA4096', 'ECCPRIME192V1', 'ECCPRIME256V1', 'ECCSECP192R1', 'ECCSECP224R1', 'ECCSECP256R1', 'ECCSECP384R1', 'ECCSECP521R1', 'ECCSECP256K1',
|
||||
'ECCSECT163K1', 'ECCSECT233K1', 'ECCSECT283K1', 'ECCSECT409K1', 'ECCSECT571K1', 'ECCSECT163R2', 'ECCSECT233R1', 'ECCSECT283R1', 'ECCSECT409R1', 'ECCSECT571R2']" ng-init="authority.keyType = 'RSA2048'"></select>
|
||||
<select class="form-control" ng-model="authority.keyType"
|
||||
ng-options="option.value as option.name for option in [
|
||||
{'name': 'RSA-2048', 'value': 'RSA2048'},
|
||||
{'name': 'RSA-4096', 'value': 'RSA4096'},
|
||||
{'name': 'ECC-PRIME256V1', 'value': 'ECCPRIME256V1'},
|
||||
{'name': 'ECC-SECP384R1', 'value': 'ECCSECP384R1'},
|
||||
{'name': 'ECC-SECP521R1', 'value': 'ECCSECP521R1'}]"
|
||||
|
||||
ng-init="authority.keyType = 'RSA2048'">
|
||||
</select>
|
||||
</div>
|
||||
</div>
|
||||
<div ng-show="authority.sensitivity == 'high'" class="form-group">
|
||||
|
|
|
|||
|
|
@ -32,10 +32,12 @@
|
|||
</label>
|
||||
<div class="col-sm-10">
|
||||
<select class="form-control" ng-model="certificate.keyType"
|
||||
ng-options="option for option in ['RSA2048', 'RSA4096', 'ECCPRIME192V1', 'ECCPRIME256V1', 'ECCSECP192R1',
|
||||
'ECCSECP224R1', 'ECCSECP256R1', 'ECCSECP384R1', 'ECCSECP521R1', 'ECCSECP256K1',
|
||||
'ECCSECT163K1', 'ECCSECT233K1', 'ECCSECT283K1', 'ECCSECT409K1', 'ECCSECT571K1',
|
||||
'ECCSECT163R2', 'ECCSECT233R1', 'ECCSECT283R1', 'ECCSECT409R1', 'ECCSECT571R2']"
|
||||
ng-options="option.value as option.name for option in [
|
||||
{'name': 'RSA-2048', 'value': 'RSA2048'},
|
||||
{'name': 'RSA-4096', 'value': 'RSA4096'},
|
||||
{'name': 'ECC-PRIME256V1', 'value': 'ECCPRIME256V1'},
|
||||
{'name': 'ECC-SECP384R1', 'value': 'ECCSECP384R1'}]"
|
||||
|
||||
ng-init="certificate.keyType = 'RSA2048'"></select>
|
||||
</div>
|
||||
</div>
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ LEMUR_ALLOWED_DOMAINS = [
|
|||
|
||||
# Lemur currently only supports SES for sending email, this address
|
||||
# needs to be verified
|
||||
LEMUR_EMAIL = ""
|
||||
LEMUR_EMAIL = "lemur@example.com"
|
||||
LEMUR_SECURITY_TEAM_EMAIL = ["security@example.com"]
|
||||
|
||||
LEMUR_HOSTNAME = "lemur.example.com"
|
||||
|
|
|
|||
|
|
@ -1,8 +1,8 @@
|
|||
from datetime import timedelta
|
||||
|
||||
import arrow
|
||||
import pytest
|
||||
from freezegun import freeze_time
|
||||
|
||||
from datetime import timedelta
|
||||
import arrow
|
||||
from moto import mock_ses
|
||||
|
||||
|
||||
|
|
@ -105,4 +105,11 @@ def test_send_expiration_notification_with_no_notifications(
|
|||
def test_send_rotation_notification(notification_plugin, certificate):
|
||||
from lemur.notifications.messaging import send_rotation_notification
|
||||
|
||||
send_rotation_notification(certificate, notification_plugin=notification_plugin)
|
||||
assert send_rotation_notification(certificate, notification_plugin=notification_plugin)
|
||||
|
||||
|
||||
@mock_ses
|
||||
def test_send_pending_failure_notification(notification_plugin, async_issuer_plugin, pending_certificate):
|
||||
from lemur.notifications.messaging import send_pending_failure_notification
|
||||
|
||||
assert send_pending_failure_notification(pending_certificate, notification_plugin=notification_plugin)
|
||||
|
|
|
|||
Loading…
Reference in New Issue