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

Have QSS only add security scanner notifications once

Browse source
This commit is contained in:
Joseph Schorr 2016-12-05 19:08:20 -05:00
parent 3bbd8ca898
commit 97d150e281
3 changed files with 51 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

21
test/test_queue.py
View file

@ -7,6 +7,7 @@ from functools import wraps
from app import app
from initdb import setup_database_for_testing, finished_database_for_testing
from data.queue import WorkQueue
from datetime import timedelta
QUEUE_NAME = 'testqueuename'
@ -147,6 +148,26 @@ class TestQueue(QueueTestCase):
self.assertEqual(self.reporter.running_count, 1)
self.assertEqual(self.reporter.total, 1)
def test_alive(self):
# No queue item = not alive.
self.assertFalse(self.queue.alive(['abc', 'def']))
# Add a queue item.
self.queue.put(['abc', 'def'], self.TEST_MESSAGE_1, available_after=-1)
self.assertTrue(self.queue.alive(['abc', 'def']))
# Retrieve the queue item.
queue_item = self.queue.get()
self.assertIsNotNone(queue_item)
self.assertTrue(self.queue.alive(['abc', 'def']))
# Make sure it is running by trying to retrieve it again.
self.assertIsNone(self.queue.get())
# Delete the queue item.
self.queue.complete(queue_item)
self.assertFalse(self.queue.alive(['abc', 'def']))
def test_specialized_queue(self):
self.queue.put(['abc', 'def'], self.TEST_MESSAGE_1, available_after=-1)
self.queue.put(['def', 'def'], self.TEST_MESSAGE_2, available_after=-1)