Fixing up notification testing (#575)
This commit is contained in:
@ -1,9 +0,0 @@
|
||||
import unittest
|
||||
|
||||
|
||||
class LemurTestCase(unittest.TestCase):
|
||||
pass
|
||||
|
||||
|
||||
class LemurPluginTestCase(LemurTestCase):
|
||||
pass
|
||||
|
@ -1,16 +1,12 @@
|
||||
import os
|
||||
import pytest
|
||||
|
||||
from flask import current_app
|
||||
|
||||
from flask_principal import identity_changed, Identity
|
||||
|
||||
from lemur import create_app
|
||||
from lemur.database import db as _db
|
||||
from lemur.auth.service import create_token
|
||||
|
||||
from .factories import AuthorityFactory, NotificationFactory, DestinationFactory, \
|
||||
CertificateFactory, UserFactory, RoleFactory, SourceFactory
|
||||
CertificateFactory, UserFactory, RoleFactory, SourceFactory, EndpointFactory
|
||||
|
||||
|
||||
def pytest_runtest_setup(item):
|
||||
@ -114,6 +110,14 @@ def certificate(session):
|
||||
return c
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def endpoint(session):
|
||||
s = SourceFactory()
|
||||
e = EndpointFactory(source=s)
|
||||
session.commit()
|
||||
return e
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def role(session):
|
||||
r = RoleFactory()
|
||||
@ -171,17 +175,3 @@ def source_plugin():
|
||||
from .plugins.source_plugin import TestSourcePlugin
|
||||
register(TestSourcePlugin)
|
||||
return TestSourcePlugin
|
||||
|
||||
|
||||
@pytest.yield_fixture(scope="function")
|
||||
def logged_in_user(session, app):
|
||||
with app.test_request_context():
|
||||
identity_changed.send(current_app._get_current_object(), identity=Identity(1))
|
||||
yield
|
||||
|
||||
|
||||
@pytest.yield_fixture(scope="function")
|
||||
def logged_in_admin(session, app):
|
||||
with app.test_request_context():
|
||||
identity_changed.send(current_app._get_current_object(), identity=Identity(2))
|
||||
yield
|
||||
|
@ -240,6 +240,7 @@ class EndpointFactory(BaseFactory):
|
||||
type = FuzzyChoice(['elb'])
|
||||
active = True
|
||||
port = FuzzyInteger(0, high=65535)
|
||||
dnsname = 'endpoint.example.com'
|
||||
policy = SubFactory(PolicyFactory)
|
||||
certificate = SubFactory(CertificateFactory)
|
||||
source = SubFactory(SourceFactory)
|
||||
|
@ -13,5 +13,5 @@ class TestNotificationPlugin(NotificationPlugin):
|
||||
super(TestNotificationPlugin, self).__init__(*args, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
def send(event_type, message, targets, options, **kwargs):
|
||||
def send(notification_type, message, targets, options, **kwargs):
|
||||
return
|
||||
|
@ -354,7 +354,7 @@ def test_get_account_number(client):
|
||||
assert get_account_number(arn) == '11111111'
|
||||
|
||||
|
||||
def test_mint_certificate(issuer_plugin, authority, logged_in_admin):
|
||||
def test_mint_certificate(issuer_plugin, authority):
|
||||
from lemur.certificates.service import mint
|
||||
cert_body, private_key, chain = mint(authority=authority, csr=CSR_STR)
|
||||
assert cert_body == INTERNAL_VALID_LONG_STR, INTERNAL_VALID_SAN_STR
|
||||
@ -372,7 +372,7 @@ def test_create_certificate(issuer_plugin, authority, user):
|
||||
assert cert.name == 'ACustomName1'
|
||||
|
||||
|
||||
def test_reissue_certificate(issuer_plugin, authority, certificate, logged_in_admin):
|
||||
def test_reissue_certificate(issuer_plugin, authority, certificate):
|
||||
from lemur.certificates.service import reissue_certificate
|
||||
new_cert = reissue_certificate(certificate)
|
||||
assert new_cert
|
||||
|
@ -3,6 +3,8 @@ from freezegun import freeze_time
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
from moto import mock_ses
|
||||
|
||||
|
||||
def test_needs_notification(app, certificate, notification):
|
||||
from lemur.notifications.messaging import needs_notification
|
||||
@ -21,11 +23,24 @@ def test_needs_notification(app, certificate, notification):
|
||||
assert needs_notification(certificate)
|
||||
|
||||
|
||||
@pytest.skip
|
||||
def test_send_expiration_notification():
|
||||
assert False
|
||||
@mock_ses
|
||||
def test_send_expiration_notification(certificate, notification, notification_plugin):
|
||||
from lemur.notifications.messaging import send_expiration_notifications
|
||||
notification.options = [{'name': 'interval', 'value': 10}, {'name': 'unit', 'value': 'days'}]
|
||||
certificate.notifications.append(notification)
|
||||
delta = certificate.not_after - timedelta(days=10)
|
||||
|
||||
with freeze_time(delta.datetime):
|
||||
sent = send_expiration_notifications()
|
||||
assert sent == 1
|
||||
|
||||
certificate.notify = False
|
||||
|
||||
sent = send_expiration_notifications()
|
||||
assert sent == 0
|
||||
|
||||
|
||||
@pytest.skip
|
||||
def test_send_rotation_notification():
|
||||
assert False
|
||||
@mock_ses
|
||||
def test_send_rotation_notification(notification_plugin, certificate):
|
||||
from lemur.notifications.messaging import send_rotation_notification
|
||||
send_rotation_notification(certificate, notification_plugin=notification_plugin)
|
||||
|
Reference in New Issue
Block a user