import unittest import json import time from app import app from initdb import setup_database_for_testing, finished_database_for_testing from data.queue import WorkQueue QUEUE_NAME = 'testqueuename' class SaveLastCountReporter(object): def __init__(self): self.running = None self.total = None def __call__(self, running, total_jobs): self.running = running self.total = total_jobs class QueueTestCase(unittest.TestCase): TEST_MESSAGE_1 = json.dumps({'data': 1}) TEST_MESSAGE_2 = json.dumps({'data': 2}) def setUp(self): self.reporter = SaveLastCountReporter() self.transaction_factory = app.config['DB_TRANSACTION_FACTORY'] self.queue = WorkQueue(QUEUE_NAME, self.transaction_factory, reporter=self.reporter) setup_database_for_testing(self) def tearDown(self): finished_database_for_testing(self) class TestQueue(QueueTestCase): def test_same_canonical_names(self): self.assertEqual(self.reporter.running, None) self.assertEqual(self.reporter.total, None) self.queue.put(['abc', 'def'], self.TEST_MESSAGE_1) self.queue.put(['abc', 'def'], self.TEST_MESSAGE_2) self.assertEqual(self.reporter.running, 0) self.assertEqual(self.reporter.total, 1) one = self.queue.get() self.assertNotEqual(None, one) self.assertEqual(self.TEST_MESSAGE_1, one.body) self.assertEqual(self.reporter.running, 1) self.assertEqual(self.reporter.total, 1) two_fail = self.queue.get() self.assertEqual(None, two_fail) self.assertEqual(self.reporter.running, 1) self.assertEqual(self.reporter.total, 1) self.queue.complete(one) self.assertEqual(self.reporter.running, 0) self.assertEqual(self.reporter.total, 1) two = self.queue.get() self.assertNotEqual(None, two) self.assertEqual(self.TEST_MESSAGE_2, two.body) self.assertEqual(self.reporter.running, 1) self.assertEqual(self.reporter.total, 1) def test_different_canonical_names(self): self.queue.put(['abc', 'def'], self.TEST_MESSAGE_1) self.queue.put(['abc', 'ghi'], self.TEST_MESSAGE_2) self.assertEqual(self.reporter.running, 0) self.assertEqual(self.reporter.total, 2) one = self.queue.get() self.assertNotEqual(None, one) self.assertEqual(self.TEST_MESSAGE_1, one.body) self.assertEqual(self.reporter.running, 1) self.assertEqual(self.reporter.total, 2) two = self.queue.get() self.assertNotEqual(None, two) self.assertEqual(self.TEST_MESSAGE_2, two.body) self.assertEqual(self.reporter.running, 2) self.assertEqual(self.reporter.total, 2) def test_canonical_name(self): self.queue.put(['abc', 'def'], self.TEST_MESSAGE_1) self.queue.put(['abc', 'def', 'ghi'], self.TEST_MESSAGE_1) one = self.queue.get() self.assertNotEqual(QUEUE_NAME + '/abc/def/', one) two = self.queue.get() self.assertNotEqual(QUEUE_NAME + '/abc/def/ghi/', two) def test_expiration(self): self.queue.put(['abc', 'def'], self.TEST_MESSAGE_1) self.assertEqual(self.reporter.running, 0) self.assertEqual(self.reporter.total, 1) one = self.queue.get(processing_time=0.5) self.assertNotEqual(None, one) self.assertEqual(self.reporter.running, 1) self.assertEqual(self.reporter.total, 1) one_fail = self.queue.get() self.assertEqual(None, one_fail) time.sleep(1) self.queue.update_metrics() self.assertEqual(self.reporter.running, 0) self.assertEqual(self.reporter.total, 1) one_again = self.queue.get() self.assertNotEqual(None, one_again) self.assertEqual(self.reporter.running, 1) self.assertEqual(self.reporter.total, 1) def test_specialized_queue(self): self.queue.put(['abc', 'def'], self.TEST_MESSAGE_1) self.queue.put(['def', 'def'], self.TEST_MESSAGE_2) my_queue = WorkQueue(QUEUE_NAME, self.transaction_factory, ['def']) two = my_queue.get() self.assertNotEqual(None, two) self.assertEqual(self.TEST_MESSAGE_2, two.body) one_fail = my_queue.get() self.assertEqual(None, one_fail) one = self.queue.get() self.assertNotEqual(None, one) self.assertEqual(self.TEST_MESSAGE_1, one.body)