Notification method tests
This commit is contained in:
Joseph Schorr
parent
5e5e191f87
commit
348b544f23
2 changed files with 165 additions and 5 deletions
|
|
@ -57,7 +57,7 @@ class NotificationMethod(object):
|
|||
"""
|
||||
Performs the notification method.
|
||||
|
||||
notification_obj: The noticication namedtuple.
|
||||
notification_obj: The notification namedtuple.
|
||||
event_handler: The NotificationEvent handler.
|
||||
notification_data: The dict of notification data placed in the queue.
|
||||
"""
|
||||
|
|
@ -83,7 +83,9 @@ class QuayNotificationMethod(NotificationMethod):
|
|||
raise CannotValidateNotificationMethodException(err_message)
|
||||
|
||||
def find_targets(self, repository, config_data):
|
||||
target_info = config_data['target']
|
||||
target_info = config_data.get('target', None)
|
||||
if not target_info or not target_info.get('kind'):
|
||||
return (True, 'Missing target', [])
|
||||
|
||||
if target_info['kind'] == 'user':
|
||||
target = model.user.get_nonrobot_user(target_info['name'])
|
||||
|
|
@ -93,9 +95,9 @@ class QuayNotificationMethod(NotificationMethod):
|
|||
|
||||
return (True, None, [target])
|
||||
elif target_info['kind'] == 'org':
|
||||
target = model.organization.get_organization(target_info['name'])
|
||||
if not target:
|
||||
# Just to be safe.
|
||||
try:
|
||||
target = model.organization.get_organization(target_info['name'])
|
||||
except model.organization.InvalidOrganizationException:
|
||||
return (True, 'Unknown organization %s' % target_info['name'], None)
|
||||
|
||||
# Only repositories under the organization can cause notifications to that org.
|
||||
|
|
|
|||
158
notifications/test/test_notificationmethod.py
Normal file
158
notifications/test/test_notificationmethod.py
Normal file
|
|
@ -0,0 +1,158 @@
|
|||
import pytest
|
||||
|
||||
from mock import patch, Mock
|
||||
from httmock import urlmatch, HTTMock
|
||||
|
||||
from data import model
|
||||
from notifications.notificationmethod import (QuayNotificationMethod, EmailMethod, WebhookMethod,
|
||||
FlowdockMethod, HipchatMethod, SlackMethod,
|
||||
CannotValidateNotificationMethodException)
|
||||
from notifications.notificationevent import NotificationEvent
|
||||
from notifications.models_interface import Repository, Notification
|
||||
|
||||
from test.fixtures import *
|
||||
|
||||
def assert_validated(method, method_config, error_message, namespace_name, repo_name):
|
||||
if error_message is None:
|
||||
method.validate(Repository(namespace_name, repo_name), method_config)
|
||||
else:
|
||||
with pytest.raises(CannotValidateNotificationMethodException) as ipe:
|
||||
method.validate(Repository(namespace_name, repo_name), method_config)
|
||||
assert ipe.value.message == error_message
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method_config,error_message', [
|
||||
({}, 'Missing target'),
|
||||
({'target': {'name': 'invaliduser', 'kind': 'user'}}, 'Unknown user invaliduser'),
|
||||
({'target': {'name': 'invalidorg', 'kind': 'org'}}, 'Unknown organization invalidorg'),
|
||||
({'target': {'name': 'invalidteam', 'kind': 'team'}}, 'Unknown team invalidteam'),
|
||||
|
||||
({'target': {'name': 'devtable', 'kind': 'user'}}, None),
|
||||
({'target': {'name': 'buynlarge', 'kind': 'org'}}, None),
|
||||
({'target': {'name': 'owners', 'kind': 'team'}}, None),
|
||||
])
|
||||
def test_validate_quay_notification(method_config, error_message, initialized_db):
|
||||
method = QuayNotificationMethod()
|
||||
assert_validated(method, method_config, error_message, 'buynlarge', 'orgrepo')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method_config,error_message', [
|
||||
({}, 'Missing e-mail address'),
|
||||
({'email': 'a@b.com'}, 'The specified e-mail address is not authorized to receive '
|
||||
'notifications for this repository'),
|
||||
|
||||
({'email': 'jschorr@devtable.com'}, None),
|
||||
])
|
||||
def test_validate_email(method_config, error_message, initialized_db):
|
||||
method = EmailMethod()
|
||||
assert_validated(method, method_config, error_message, 'devtable', 'simple')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method_config,error_message', [
|
||||
({}, 'Missing webhook URL'),
|
||||
({'url': 'http://example.com'}, None),
|
||||
])
|
||||
def test_validate_webhook(method_config, error_message, initialized_db):
|
||||
method = WebhookMethod()
|
||||
assert_validated(method, method_config, error_message, 'devtable', 'simple')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method_config,error_message', [
|
||||
({}, 'Missing Flowdock API Token'),
|
||||
({'flow_api_token': 'sometoken'}, None),
|
||||
])
|
||||
def test_validate_flowdock(method_config, error_message, initialized_db):
|
||||
method = FlowdockMethod()
|
||||
assert_validated(method, method_config, error_message, 'devtable', 'simple')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method_config,error_message', [
|
||||
({}, 'Missing Hipchat Room Notification Token'),
|
||||
({'notification_token': 'sometoken'}, 'Missing Hipchat Room ID'),
|
||||
({'notification_token': 'sometoken', 'room_id': 'foo'}, None),
|
||||
])
|
||||
def test_validate_hipchat(method_config, error_message, initialized_db):
|
||||
method = HipchatMethod()
|
||||
assert_validated(method, method_config, error_message, 'devtable', 'simple')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method_config,error_message', [
|
||||
({}, 'Missing Slack Callback URL'),
|
||||
({'url': 'http://example.com'}, None),
|
||||
])
|
||||
def test_validate_slack(method_config, error_message, initialized_db):
|
||||
method = SlackMethod()
|
||||
assert_validated(method, method_config, error_message, 'devtable', 'simple')
|
||||
|
||||
|
||||
@pytest.mark.parametrize('target,expected_users', [
|
||||
({'name': 'devtable', 'kind': 'user'}, ['devtable']),
|
||||
({'name': 'buynlarge', 'kind': 'org'}, ['buynlarge']),
|
||||
({'name': 'creators', 'kind': 'team'}, ['creator']),
|
||||
])
|
||||
def test_perform_quay_notification(target, expected_users, initialized_db):
|
||||
repository = Repository('buynlarge', 'orgrepo')
|
||||
notification = Notification(uuid='fake', event_name='repo_push', method_name='quay',
|
||||
event_config_dict={}, method_config_dict={'target': target},
|
||||
repository=repository)
|
||||
|
||||
event_handler = NotificationEvent.get_event('repo_push')
|
||||
|
||||
sample_data = event_handler.get_sample_data(repository, {})
|
||||
|
||||
method = QuayNotificationMethod()
|
||||
method.perform(notification, event_handler, {'event_data': sample_data})
|
||||
|
||||
# Ensure that the notification was written for all the expected users.
|
||||
if target['kind'] != 'team':
|
||||
user = model.user.get_namespace_user(target['name'])
|
||||
assert len(model.notification.list_notifications(user, kind_name='repo_push')) > 0
|
||||
|
||||
|
||||
def test_perform_email(initialized_db):
|
||||
repository = Repository('buynlarge', 'orgrepo')
|
||||
notification = Notification(uuid='fake', event_name='repo_push', method_name='email',
|
||||
event_config_dict={}, method_config_dict={'email': 'test@example.com'},
|
||||
repository=repository)
|
||||
|
||||
event_handler = NotificationEvent.get_event('repo_push')
|
||||
|
||||
sample_data = event_handler.get_sample_data(repository, {})
|
||||
|
||||
mock = Mock()
|
||||
def get_mock(*args, **kwargs):
|
||||
return mock
|
||||
|
||||
with patch('notifications.notificationmethod.Message', get_mock):
|
||||
method = EmailMethod()
|
||||
method.perform(notification, event_handler, {'event_data': sample_data, 'performer_data': {}})
|
||||
|
||||
mock.send.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.parametrize('method, method_config, netloc', [
|
||||
(WebhookMethod, {'url': 'http://testurl'}, 'testurl'),
|
||||
(FlowdockMethod, {'flow_api_token': 'token'}, 'api.flowdock.com'),
|
||||
(HipchatMethod, {'notification_token': 'token', 'room_id': 'foo'}, 'api.hipchat.com'),
|
||||
(SlackMethod, {'url': 'http://example.com'}, 'example.com'),
|
||||
])
|
||||
def test_perform_http_call(method, method_config, netloc, initialized_db):
|
||||
repository = Repository('buynlarge', 'orgrepo')
|
||||
notification = Notification(uuid='fake', event_name='repo_push', method_name=method.method_name(),
|
||||
event_config_dict={}, method_config_dict=method_config,
|
||||
repository=repository)
|
||||
|
||||
event_handler = NotificationEvent.get_event('repo_push')
|
||||
|
||||
sample_data = event_handler.get_sample_data(repository, {})
|
||||
|
||||
url_hit = [False]
|
||||
@urlmatch(netloc=netloc)
|
||||
def url_handler(_, __):
|
||||
url_hit[0] = True
|
||||
return ''
|
||||
|
||||
with HTTMock(url_handler):
|
||||
method().perform(notification, event_handler, {'event_data': sample_data, 'performer_data': {}})
|
||||
|
||||
assert url_hit[0]
|
||||
Reference in a new issue