alerts: rework and significantly expand scope of tests

gracedb/alerts/tests/conftest.py

+import copy
+import pytest
+import re
+
+from django.contrib.auth import get_user_model
+
+from alerts.models import Contact, Notification
+from events.models import Label, Group, Pipeline, Search, Event
+from superevents.models import Superevent
+from .constants import (
+    DEFAULT_FAR_T, DEFAULT_LABELS, DEFAULT_LABEL_QUERY, LABEL_QUERY2,
+    RANDOM_LABEL, LABEL_QUERY_PARSER, DEFAULT_GROUP, DEFAULT_PIPELINE,
+    DEFAULT_SEARCH
+)
+
+UserModel = get_user_model()
+
+
+###############################################################################
+# UTILITY FUNCTIONS ###########################################################
+###############################################################################
+def create_notification(
+    user,
+    notification_category,
+    contact_description='test',
+    phone=None,
+    email=None,
+    phone_method=Contact.CONTACT_PHONE_BOTH,
+    notification_description='test',
+    far_threshold=None,
+    ns_candidate=None,
+    label_names=None,
+    label_query=None,
+    groups=None,
+    pipelines=None,
+    searches=None,
+):
+    # Create contact
+    contact_dict = {}
+    if phone is not None and email is not None:
+        raise ValueError("Specify only one of label_names or label_query")
+    elif phone:
+        contact_dict['phone'] = phone
+        contact_dict['phone_method'] = phone_method
+    elif email:
+        contact_dict['email'] = email
+    c = Contact.objects.create(
+        user=user,
+        description=contact_description,
+        verified=True,
+        **contact_dict
+    )
+
+    # Create notification
+    notification_dict = {}
+    if far_threshold:
+        notification_dict['far_threshold'] = far_threshold
+    if ns_candidate:
+        notification_dict['ns_candidate'] = ns_candidate
+    if label_query:
+        notification_dict['label_query'] = label_query
+        
+    n = Notification.objects.create(
+        user=user,
+        description=notification_description,
+        category=notification_category,
+        **notification_dict
+    )
+    # Add m2m relations
+    n.contacts.add(c)
+    if label_names and label_query:
+        raise ValueError('')
+    elif label_query:
+        label_names = LABEL_QUERY_PARSER.findall(label_query)
+    if label_names:
+        for l in label_names:
+            label, _ = Label.objects.get_or_create(name=l)
+            n.labels.add(label)
+    if notification_category == Notification.NOTIFICATION_CATEGORY_EVENT:
+        if groups:
+            for g in groups:
+                group, _ = Group.objects.get_or_create(name=g)
+                n.groups.add(group)
+        if pipelines:
+            for p in pipelines:
+                pipeline, _ = Pipeline.objects.get_or_create(name=p)
+                n.pipelines.add(pipeline)
+        if searches:
+            for s in searches:
+                search, _ = Search.objects.get_or_create(name=s)
+                n.searches.add(search)
+    return n
+
+
+###############################################################################
+# FIXTURES ####################################################################
+###############################################################################
+@pytest.mark.django_db
+@pytest.fixture
+def event():
+    group, _ = Group.objects.get_or_create(name='event_group')
+    pipeline, _ = Pipeline.objects.get_or_create(name='event_pipeline')
+    search, _ = Search.objects.get_or_create(name='event_search')
+    user = UserModel.objects.create(username='event.creator')
+    event = Event.objects.create(group=group, pipeline=pipeline, search=search,
+                                 far=1, submitter=user)
+    return event
+
+
+@pytest.mark.django_db
+@pytest.fixture
+def superevent(event):
+    user = UserModel.objects.create(username='superevent.creator')
+    superevent = Superevent.objects.create(submitter=user, t_start=0, t_0=1,
+                                           t_end=2, preferred_event=event)
+    return superevent
+
+
+SUPEREVENT_NOTIFICATION_DATA = [
+    dict(desc='all'),
+    dict(desc='far_t_only', far_threshold=DEFAULT_FAR_T),
+    dict(desc='nscand_only', ns_candidate=True),
+    dict(desc='labels_only', label_names=DEFAULT_LABELS),
+    dict(desc='labelq_only', label_query=DEFAULT_LABEL_QUERY['query']),
+    dict(desc='far_t_and_nscand', far_threshold=DEFAULT_FAR_T,
+         ns_candidate=True),
+    dict(desc='far_t_and_labels', far_threshold=DEFAULT_FAR_T,
+         label_names=DEFAULT_LABELS),
+    dict(desc='far_t_and_labelq', far_threshold=DEFAULT_FAR_T,
+         label_query=DEFAULT_LABEL_QUERY['query']),
+    dict(desc='nscand_and_labels', ns_candidate=True,
+         label_names=DEFAULT_LABELS),
+    dict(desc='nscand_and_labelq', ns_candidate=True,
+         label_query=DEFAULT_LABEL_QUERY['query']),
+    dict(desc='far_t_and_nscand_and_labels', far_threshold=DEFAULT_FAR_T,
+         ns_candidate=True, label_names=DEFAULT_LABELS),
+    dict(desc='far_t_and_nscand_and_labelq', far_threshold=DEFAULT_FAR_T,
+         ns_candidate=True, label_query=DEFAULT_LABEL_QUERY['query']),
+    dict(desc='labelq2_only', label_query=LABEL_QUERY2),
+    dict(desc='far_t_and_labelq2', far_threshold=DEFAULT_FAR_T,
+         label_query=LABEL_QUERY2),
+    dict(desc='nscand_and_labelq2', ns_candidate=True,
+         label_query=LABEL_QUERY2),
+    dict(desc='far_t_and_nscand_and_labelq2', far_threshold=DEFAULT_FAR_T,
+         ns_candidate=True, label_query=LABEL_QUERY2),
+]
+@pytest.mark.django_db
+@pytest.fixture
+def superevent_notifications(request):
+    # Get user fixture
+    user = request.getfixturevalue('internal_user')
+
+    # Create notifications
+    notification_pks = []
+    notification_data = copy.deepcopy(SUPEREVENT_NOTIFICATION_DATA) 
+    for notification_dict in notification_data:
+        desc = notification_dict.pop('desc')
+        n = create_notification(
+            user,
+            Notification.NOTIFICATION_CATEGORY_SUPEREVENT,
+            phone='12345678901',
+            notification_description=desc,
+            **notification_dict
+        )
+        notification_pks.append(n.pk)
+
+    return Notification.objects.filter(pk__in=notification_pks)
+
+
+EVENT_NOTIFICATION_DATA = copy.deepcopy(SUPEREVENT_NOTIFICATION_DATA)
+EVENT_NOTIFICATION_DATA += [
+    dict(desc='gps_only', groups=[DEFAULT_GROUP], pipelines=[DEFAULT_PIPELINE],
+         searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_gps', far_threshold=DEFAULT_FAR_T,
+         groups=[DEFAULT_GROUP], pipelines=[DEFAULT_PIPELINE],
+         searches=[DEFAULT_SEARCH]),
+    dict(desc='nscand_and_gps', ns_candidate=True,
+         groups=[DEFAULT_GROUP], pipelines=[DEFAULT_PIPELINE],
+         searches=[DEFAULT_SEARCH]),
+    dict(desc='labels_and_gps', label_names=DEFAULT_LABELS,
+         groups=[DEFAULT_GROUP], pipelines=[DEFAULT_PIPELINE],
+         searches=[DEFAULT_SEARCH]),
+    dict(desc='labelq_and_gps', label_query=DEFAULT_LABEL_QUERY['query'],
+         groups=[DEFAULT_GROUP], pipelines=[DEFAULT_PIPELINE],
+         searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_nscand_and_gps', far_threshold=DEFAULT_FAR_T,
+         ns_candidate=True, groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_labels_and_gps', far_threshold=DEFAULT_FAR_T,
+         label_names=DEFAULT_LABELS, groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_labelq_and_gps', far_threshold=DEFAULT_FAR_T,
+         label_query=DEFAULT_LABEL_QUERY['query'], groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='nscand_and_labels_and_gps', ns_candidate=True,
+         label_names=DEFAULT_LABELS, groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='nscand_and_labelq_and_gps', ns_candidate=True,
+         label_query=DEFAULT_LABEL_QUERY['query'], groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_nscand_and_labels_and_gps',
+         far_threshold=DEFAULT_FAR_T, ns_candidate=True,
+         label_names=DEFAULT_LABELS, groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_nscand_and_labelq_and_gps',
+         far_threshold=DEFAULT_FAR_T, ns_candidate=True,
+         label_query=DEFAULT_LABEL_QUERY['query'], groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='labelq2_and_gps', label_query=LABEL_QUERY2,
+         groups=[DEFAULT_GROUP], pipelines=[DEFAULT_PIPELINE],
+         searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_labelq2_and_gps', far_threshold=DEFAULT_FAR_T,
+         label_query=LABEL_QUERY2, groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='nscand_and_labelq2_and_gps', ns_candidate=True,
+         label_query=LABEL_QUERY2, groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+    dict(desc='far_t_and_nscand_and_labelq2_and_gps',
+         far_threshold=DEFAULT_FAR_T, ns_candidate=True,
+         label_query=LABEL_QUERY2, groups=[DEFAULT_GROUP],
+         pipelines=[DEFAULT_PIPELINE], searches=[DEFAULT_SEARCH]),
+]
+@pytest.mark.django_db
+@pytest.fixture
+def event_notifications(request):
+    # Get user fixture
+    user = request.getfixturevalue('internal_user')
+
+    # Create notifications
+    notification_pks = []
+    notification_data = copy.deepcopy(EVENT_NOTIFICATION_DATA) 
+    for notification_dict in notification_data:
+        desc = notification_dict.pop('desc')
+        n = create_notification(
+            user,
+            Notification.NOTIFICATION_CATEGORY_EVENT,
+            phone='12345678901',
+            notification_description=desc,
+            **notification_dict
+        )
+        notification_pks.append(n.pk)
+
+    return Notification.objects.filter(pk__in=notification_pks)

gracedb/alerts/tests/constants.py

+import re
+
+
+# Constants for use in various tests
+DEFAULT_FAR_T = 1  # Hz
+DEFAULT_LABELS = ['L1', 'L2']
+LABEL_QUERY_PARSER = re.compile(r'([a-zA-Z0-9]+)')
+DEFAULT_LABEL_QUERY = {'query': 'L3 & L4'}
+DEFAULT_LABEL_QUERY['labels'] = LABEL_QUERY_PARSER.findall(
+    DEFAULT_LABEL_QUERY['query']
+)
+LABEL_QUERY2 = 'L5 & L6 & ~L7'
+RANDOM_LABEL = 'L12345'
+DEFAULT_GROUP = 'Group1'
+DEFAULT_PIPELINE = 'Pipeline1'
+DEFAULT_SEARCH = 'Search1'

gracedb/alerts/tests/test_alerts.py

+import itertools
+try:
+    from unittest import mock
+except ImportError:  # python < 3
+    import mock
+import types
+
+import pytest
+
+from alerts.main import issue_alerts
+from events.models import Event
+from superevents.models import Superevent
+
+
+###############################################################################
+# FIXTURES ####################################################################
+###############################################################################
+# Mock recipient getter dict
+@pytest.fixture
+def mock_rg_dict():
+    def _inner(alert_type='new', recips_exist=True):
+        # Set up mock recipient getter stuff
+        mock_recipients = mock.MagicMock()
+        mock_recipients.exists.return_value = recips_exist
+        mock_rg = mock.MagicMock()
+        mock_rg.get_recipients.return_value = \
+            (mock_recipients, mock_recipients)
+        mock_rg_class = mock.MagicMock()
+        mock_rg_class.return_value = mock_rg
+        mock_rg_dict = {alert_type: mock_rg_class}
+        return mock_rg_dict
+    return _inner
+
+
+###############################################################################
+# TESTS #######################################################################
+###############################################################################
+@pytest.mark.parametrize(
+    "xmpp,email,phone",
+    list(itertools.product((True, False), repeat=3))
+)
+def test_alert_settings(settings, mock_rg_dict, xmpp, email, phone):
+    # Set up settings
+    settings.SEND_XMPP_ALERTS = xmpp
+    settings.SEND_EMAIL_ALERTS = email
+    settings.SEND_PHONE_ALERTS = phone
+
+    # Set up mock superevent object
+    superevent = mock.MagicMock()
+    superevent.is_test.return_value = False
+    superevent.is_mdc.return_value = False
+    preferred_event = mock.MagicMock()
+    type(preferred_event).offline = mock.PropertyMock(return_value=False)
+    superevent.preferred_event = preferred_event
+
+    # Call issue_alerts
+    with mock.patch('alerts.main.issue_xmpp_alerts') as mock_xmpp, \
+         mock.patch('alerts.main.issue_email_alerts') as mock_email, \
+         mock.patch('alerts.main.issue_phone_alerts') as mock_phone, \
+         mock.patch('alerts.main.is_event') as mock_is_event, \
+         mock.patch.dict('alerts.main.ALERT_TYPE_RECIPIENT_GETTERS',
+                         mock_rg_dict(), clear=True):
+
+        mock_is_event.return_value = False
+        issue_alerts(superevent, 'new', None)
+
+    # Check results
+    if xmpp:
+        assert mock_xmpp.call_count == 1
+    if phone:
+        assert mock_phone.call_count == 1
+    if email:
+        assert mock_email.call_count == 1
+    if not (phone or email):
+        assert mock_is_event.call_count == 0 
+
+
+@pytest.mark.parametrize(
+    "is_event,is_mdc,is_test,is_offline",
+    list(itertools.product((True, False), repeat=4))
+)
+def test_no_alerts_for_test_mdc_offline_events_and_superevents(
+    settings, mock_rg_dict, is_event, is_mdc, is_test, is_offline
+):
+    # Set up settings
+    settings.SEND_XMPP_ALERTS = False
+    settings.SEND_EMAIL_ALERTS = True
+    settings.SEND_PHONE_ALERTS = True
+
+    # Set up mock event/superevent object
+    # First, set up event
+    event = mock.MagicMock()
+    type(event).offline = mock.PropertyMock(return_value=is_offline)
+    # If we're doing a superevent, then set the event as its preferred_event
+    if not is_event:
+        es = mock.MagicMock()
+        es.preferred_event = event
+    else:
+        es = event
+    # is_mdc and is_test are handled the same way for both events and
+    # superevents, so we can set it up at this point
+    es.is_mdc.return_value = is_mdc
+    es.is_test.return_value = is_test
+
+    # Instantiate mock_rg_dict
+    mock_rg_dict = mock_rg_dict()
+
+    # Call issue_alerts
+    with mock.patch('alerts.main.issue_xmpp_alerts') as mock_xmpp, \
+         mock.patch('alerts.main.issue_email_alerts') as mock_email, \
+         mock.patch('alerts.main.issue_phone_alerts') as mock_phone, \
+         mock.patch('alerts.main.is_event') as mock_is_event, \
+         mock.patch.dict('alerts.main.ALERT_TYPE_RECIPIENT_GETTERS',
+                         mock_rg_dict, clear=True):
+
+        mock_is_event.return_value = is_event
+        issue_alerts(es, 'new', None)
+
+    # Check results
+    assert es.is_mdc.call_count == 1
+    assert es.is_test.call_count == int(not es.is_mdc())
+    # Whether the recipient_getter class is called or not depends
+    # finally on whether the event is offline or not
+    if (not (es.is_mdc() or es.is_test())):
+        assert mock_rg_dict['new'].call_count == int(not event.offline)
+    else:
+        assert mock_rg_dict['new'].call_count == 0

gracedb/alerts/tests/test_contacts.py

+import pytest
+
+from django.contrib.auth import get_user_model
+
+from alerts.models import Contact, Notification
+from alerts.recipients import CreationRecipientGetter
+
+UserModel = get_user_model()
+
+
+@pytest.mark.django_db
+def test_multiple_contacts(superevent, internal_user):
+    # Set up contacts and notifications
+    c1 = Contact.objects.create(
+        user=internal_user, description='c1', verified=True,
+        phone='12345678901', phone_method=Contact.CONTACT_PHONE_TEXT,
+    )
+    c2 = Contact.objects.create(
+        user=internal_user, description='c2', verified=True,
+        email='test@test.com'
+    )
+    n = Notification.objects.create(
+        user=internal_user, description='test',
+        category=Notification.NOTIFICATION_CATEGORY_SUPEREVENT
+    )
+    n.contacts.add(*(c1, c2))
+
+    # Get recipients 
+    recipient_getter = CreationRecipientGetter(superevent)
+    email_contacts, phone_contacts = recipient_getter.get_recipients()
+
+    # Check results
+    assert email_contacts.count() == 1
+    assert email_contacts.first().pk == c2.pk
+    assert phone_contacts.count() == 1
+    assert phone_contacts.first().pk == c1.pk
+
+
+@pytest.mark.django_db
+def test_duplicate_contacts(superevent, internal_user):
+    # Set up contacts and notifications
+    c = Contact.objects.create(
+        user=internal_user, description='test', verified=True,
+        phone='12345678901', phone_method=Contact.CONTACT_PHONE_TEXT,
+    )
+    n1 = Notification.objects.create(
+        user=internal_user, description='n1',
+        category=Notification.NOTIFICATION_CATEGORY_SUPEREVENT
+    )
+    n2 = Notification.objects.create(
+        user=internal_user, description='n2',
+        category=Notification.NOTIFICATION_CATEGORY_SUPEREVENT
+    )
+    n1.contacts.add(c)
+    n2.contacts.add(c)
+
+    # Get notifications and check results
+    recipient_getter = CreationRecipientGetter(superevent)
+    notifications = recipient_getter.get_notifications()
+    assert notifications.count() == 2
+    for pk in (n1.pk, n2.pk):
+        assert notifications.filter(pk=pk).exists()
+   
+    # Get recipients and check results 
+    email_contacts, phone_contacts = recipient_getter.get_recipients()
+
+    # Check results
+    assert email_contacts.count() == 0
+    assert phone_contacts.count() == 1
+    assert phone_contacts.first().pk == c.pk
+
+
+@pytest.mark.django_db
+def test_contacts_non_internal_user(superevent, internal_user):
+    # NOTE: this test handles the case where a user with contacts/notifications
+    # already set up leaves the collboration.
+
+    # Create a non-internal user
+    external_user = UserModel.objects.create(username='external.user')
+
+    # Set up contacts and notifications
+    c_internal = Contact.objects.create(
+        user=internal_user, description='test', verified=True,
+        phone='12345678901', phone_method=Contact.CONTACT_PHONE_TEXT,
+    )
+    c_external = Contact.objects.create(
+        user=external_user, description='test', verified=True,
+        phone='12345678901', phone_method=Contact.CONTACT_PHONE_TEXT,
+    )
+    n_internal = Notification.objects.create(
+        user=internal_user, description='internal',
+        category=Notification.NOTIFICATION_CATEGORY_SUPEREVENT
+    )
+    n_internal.contacts.add(c_internal)
+    n_external = Notification.objects.create(
+        user=external_user, description='external',
+        category=Notification.NOTIFICATION_CATEGORY_SUPEREVENT
+    )
+    n_external.contacts.add(c_external)
+
+    # Get notifications and check results
+    recipient_getter = CreationRecipientGetter(superevent)
+    notifications = recipient_getter.get_notifications()
+    assert notifications.count() == 2
+    for pk in (n_internal.pk, n_external.pk):
+        assert notifications.filter(pk=pk).exists()
+   
+    # Get recipients and check results 
+    email_contacts, phone_contacts = recipient_getter.get_recipients()
+
+    # Check results
+    assert email_contacts.count() == 0
+    assert phone_contacts.count() == 1
+    assert phone_contacts.first().pk == c_internal.pk
+
+
+@pytest.mark.django_db
+def test_unverified_contact(superevent, internal_user):
+    # NOTE: this test handles the case where a user with contacts/notifications
+    # already set up leaves the collboration.
+
+    # Set up contacts and notifications
+    c_verified = Contact.objects.create(
+        user=internal_user, description='test', verified=True,
+        phone='12345678901', phone_method=Contact.CONTACT_PHONE_TEXT,
+    )
+    c_unverified = Contact.objects.create(
+        user=internal_user, description='test', verified=False,
+        phone='12345678901', phone_method=Contact.CONTACT_PHONE_TEXT,
+    )
+    n = Notification.objects.create(
+        user=internal_user, description='internal',
+        category=Notification.NOTIFICATION_CATEGORY_SUPEREVENT
+    )
+    n.contacts.add(*(c_verified, c_unverified))
+
+    # Get notifications and check results
+    recipient_getter = CreationRecipientGetter(superevent)
+    notifications = recipient_getter.get_notifications()
+    assert notifications.count() == 1
+    assert notifications.first().pk == n.pk
+   
+    # Get recipients and check results 
+    email_contacts, phone_contacts = recipient_getter.get_recipients()
+
+    # Check results
+    assert email_contacts.count() == 0
+    assert phone_contacts.count() == 1
+    assert phone_contacts.first().pk == c_verified.pk