vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Change notificationworker test to pytest

Browse source
This commit is contained in:
Joseph Schorr 2017-07-12 15:10:58 +03:00
parent 50c2f1fde8
commit 8ec198228c
1 changed files with 25 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
workers/notificationworker/test/test_notificationworker.py
View file

@ -1,45 +1,32 @@
import unittest
from data import model
from workers.notificationworker import NotificationWorker
from initdb import setup_database_for_testing, finished_database_for_testing
from workers.notificationworker.notificationworker import NotificationWorker
from test.fixtures import *
class NotificationWorkerTests(unittest.TestCase):
def setUp(self):
setup_database_for_testing(self)
def test_basic_notification(initialized_db):
# Ensure the public user doesn't have any notifications.
target_user = model.user.get_user('public')
assert len(list(model.notification.list_notifications(target_user))) == 0
def tearDown(self):
finished_database_for_testing(self)
def test_basic_notification(self):
# Ensure the public user doesn't have any notifications.
target_user = model.user.get_user('public')
self.assertEquals(0, len(list(model.notification.list_notifications(target_user))))
# Add a basic build notification.
repo = model.repository.get_repository('devtable', 'simple')
method_data = {
'target': {
'kind': 'user',
'name': 'public',
}
# Add a basic build notification.
repo = model.repository.get_repository('devtable', 'simple')
method_data = {
'target': {
'kind': 'user',
'name': 'public',
}
notification = model.notification.create_repo_notification(repo, 'build_success',
'quay_notification', method_data, {})
}
notification = model.notification.create_repo_notification(repo, 'build_success',
'quay_notification', method_data, {})
notification_uuid = notification.uuid
event_data = {}
notification_uuid = notification.uuid
event_data = {}
# Fire off the queue processing.
worker = NotificationWorker(None)
worker.process_queue_item({
'notification_uuid': notification_uuid,
'event_data': event_data,
})
# Fire off the queue processing.
worker = NotificationWorker(None)
worker.process_queue_item({
'notification_uuid': notification_uuid,
'event_data': event_data,
})
# Ensure the notification was handled.
self.assertEquals(1, len(list(model.notification.list_notifications(target_user))))
if __name__ == '__main__':
unittest.main()
# Ensure the notification was handled.
assert len(list(model.notification.list_notifications(target_user))) == 1