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

Complete item queue test

Browse source
This commit is contained in:
Joseph Schorr 2017-01-12 13:21:59 -05:00
parent dcfd379b17
commit 939c122f70
1 changed files with 18 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
test/test_queue.py
View file

@ -244,6 +244,24 @@ class TestQueue(QueueTestCase):
count = self.queue.num_available_jobs_between(now, now, 'abc') count = self.queue.num_available_jobs_between(now, now, 'abc')
self.assertEqual(0, count) self.assertEqual(0, count)
def test_complete(self):
# Add some items.
self.queue.put(['somenamespace', 'abc', 'def'], self.TEST_MESSAGE_1, available_after=-10)
now = datetime.utcnow()
count = self.queue.num_available_jobs_between(now - timedelta(seconds=60), now,
['/somenamespace'])
self.assertEqual(1, count)
# Retrieve it.
item = self.queue.get()
self.assertIsNotNone(item)
self.assertTrue(self.reporter.currently_processing)
# Mark it as complete.
self.queue.complete(item)
self.assertFalse(self.reporter.currently_processing)
def test_cancel(self): def test_cancel(self):
# Add an item. # Add an item.
self.queue.put(['somenamespace', 'abc', 'def'], self.TEST_MESSAGE_1, available_after=-10) self.queue.put(['somenamespace', 'abc', 'def'], self.TEST_MESSAGE_1, available_after=-10)