From 348b544f232c1f7873a81e5931e1445fbd837c47 Mon Sep 17 00:00:00 2001 From: Joseph Schorr Date: Fri, 14 Jul 2017 20:01:14 +0300 Subject: [PATCH] Notification method tests --- notifications/notificationmethod.py | 12 +- notifications/test/test_notificationmethod.py | 158 ++++++++++++++++++ 2 files changed, 165 insertions(+), 5 deletions(-) create mode 100644 notifications/test/test_notificationmethod.py diff --git a/notifications/notificationmethod.py b/notifications/notificationmethod.py index 2a426cfa6..ff2887976 100644 --- a/notifications/notificationmethod.py +++ b/notifications/notificationmethod.py @@ -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. diff --git a/notifications/test/test_notificationmethod.py b/notifications/test/test_notificationmethod.py new file mode 100644 index 000000000..602cfad91 --- /dev/null +++ b/notifications/test/test_notificationmethod.py @@ -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]