Merge pull request #1933 from coreos-inc/test-notifications-api

Add a test for issuing test notifications

test/test_api_usage.py

@@ -21,7 +21,7 @@ from mockldap import MockLdap
 from endpoints.api import api_bp, api
 from endpoints.building import PreparedBuild
 from endpoints.webhooks import webhooks
-from app import app, config_provider
+from app import app, config_provider, notification_queue
 from buildtrigger.basehandler import BuildTriggerHandler
 from initdb import setup_database_for_testing, finished_database_for_testing
 from data import database, model
@@ -40,7 +40,9 @@ from endpoints.api.trigger import (BuildTriggerActivate, BuildTriggerSources, Bu
                                    TriggerBuildList, ActivateBuildTrigger, BuildTrigger,
                                    BuildTriggerList, BuildTriggerAnalyze, BuildTriggerFieldValues)
 from endpoints.api.repoemail import RepositoryAuthorizedEmail
-from endpoints.api.repositorynotification import RepositoryNotification, RepositoryNotificationList
+from endpoints.api.repositorynotification import (RepositoryNotification,
+                                                  RepositoryNotificationList,
+                                                  TestRepositoryNotification)
 from endpoints.api.user import (PrivateRepositories, ConvertToOrganization, Signout, Signin, User,
                                 UserAuthorizationList, UserAuthorization, UserNotification,
                                 UserNotificationList, StarredRepositoryList, StarredRepository)
@@ -2288,6 +2290,28 @@ class TestRepositoryEmail(ApiTestCase):
 
 
 class TestRepositoryNotifications(ApiTestCase):
+  def test_testnotification(self):
+    self.login(ADMIN_ACCESS_USER)
+
+    # Add a notification.
+    json = self.postJsonResponse(RepositoryNotificationList,
+                                 params=dict(repository=ADMIN_ACCESS_USER + '/simple'),
+                                 data=dict(config={'url': 'http://example.com'}, event='repo_push',
+                                           method='webhook', eventConfig={}),
+                                 expected_code=201)
+    uuid = json['uuid']
+
+    self.assertIsNone(notification_queue.get())
+
+    # Issue a test notification.
+    self.postJsonResponse(TestRepositoryNotification,
+                          params=dict(repository=ADMIN_ACCESS_USER + '/simple', uuid=uuid))
+
+    # Ensure the item is in the queue.
+    found = notification_queue.get()
+    self.assertIsNotNone(found)
+    self.assertTrue('notification_uuid' in found['body'])
+
   def test_webhooks(self):
     self.login(ADMIN_ACCESS_USER)