| """license: Apache License 2.0, see LICENSE for more details.""" |
| import uuid |
| |
| from nose import SkipTest |
| from nose.tools import eq_, ok_ |
| |
| from kazoo.testing import KazooTestCase |
| from kazoo.tests.util import TRAVIS_ZK_VERSION |
| |
| |
| class KazooQueueTests(KazooTestCase): |
| |
| def _makeOne(self): |
| path = "/" + uuid.uuid4().hex |
| return self.client.Queue(path) |
| |
| def test_queue_validation(self): |
| queue = self._makeOne() |
| self.assertRaises(TypeError, queue.put, {}) |
| self.assertRaises(TypeError, queue.put, b"one", b"100") |
| self.assertRaises(TypeError, queue.put, b"one", 10.0) |
| self.assertRaises(ValueError, queue.put, b"one", -100) |
| self.assertRaises(ValueError, queue.put, b"one", 100000) |
| |
| def test_empty_queue(self): |
| queue = self._makeOne() |
| eq_(len(queue), 0) |
| self.assertTrue(queue.get() is None) |
| eq_(len(queue), 0) |
| |
| def test_queue(self): |
| queue = self._makeOne() |
| queue.put(b"one") |
| queue.put(b"two") |
| queue.put(b"three") |
| eq_(len(queue), 3) |
| |
| eq_(queue.get(), b"one") |
| eq_(queue.get(), b"two") |
| eq_(queue.get(), b"three") |
| eq_(len(queue), 0) |
| |
| def test_priority(self): |
| queue = self._makeOne() |
| queue.put(b"four", priority=101) |
| queue.put(b"one", priority=0) |
| queue.put(b"two", priority=0) |
| queue.put(b"three", priority=10) |
| |
| eq_(queue.get(), b"one") |
| eq_(queue.get(), b"two") |
| eq_(queue.get(), b"three") |
| eq_(queue.get(), b"four") |
| |
| |
| class KazooLockingQueueTests(KazooTestCase): |
| |
| def setUp(self): |
| KazooTestCase.setUp(self) |
| skip = False |
| if TRAVIS_ZK_VERSION and TRAVIS_ZK_VERSION < (3, 4): |
| skip = True |
| elif TRAVIS_ZK_VERSION and TRAVIS_ZK_VERSION >= (3, 4): |
| skip = False |
| else: |
| ver = self.client.server_version() |
| if ver[1] < 4: |
| skip = True |
| if skip: |
| raise SkipTest("Must use Zookeeper 3.4 or above") |
| |
| def _makeOne(self): |
| path = "/" + uuid.uuid4().hex |
| return self.client.LockingQueue(path) |
| |
| def test_queue_validation(self): |
| queue = self._makeOne() |
| self.assertRaises(TypeError, queue.put, {}) |
| self.assertRaises(TypeError, queue.put, b"one", b"100") |
| self.assertRaises(TypeError, queue.put, b"one", 10.0) |
| self.assertRaises(ValueError, queue.put, b"one", -100) |
| self.assertRaises(ValueError, queue.put, b"one", 100000) |
| self.assertRaises(TypeError, queue.put_all, {}) |
| self.assertRaises(TypeError, queue.put_all, [{}]) |
| self.assertRaises(TypeError, queue.put_all, [b"one"], b"100") |
| self.assertRaises(TypeError, queue.put_all, [b"one"], 10.0) |
| self.assertRaises(ValueError, queue.put_all, [b"one"], -100) |
| self.assertRaises(ValueError, queue.put_all, [b"one"], 100000) |
| |
| def test_empty_queue(self): |
| queue = self._makeOne() |
| eq_(len(queue), 0) |
| self.assertTrue(queue.get(0) is None) |
| eq_(len(queue), 0) |
| |
| def test_queue(self): |
| queue = self._makeOne() |
| queue.put(b"one") |
| queue.put_all([b"two", b"three"]) |
| eq_(len(queue), 3) |
| |
| ok_(not queue.consume()) |
| ok_(not queue.holds_lock()) |
| eq_(queue.get(1), b"one") |
| ok_(queue.holds_lock()) |
| # Without consuming, should return the same element |
| eq_(queue.get(1), b"one") |
| ok_(queue.consume()) |
| ok_(not queue.holds_lock()) |
| eq_(queue.get(1), b"two") |
| ok_(queue.holds_lock()) |
| ok_(queue.consume()) |
| ok_(not queue.holds_lock()) |
| eq_(queue.get(1), b"three") |
| ok_(queue.holds_lock()) |
| ok_(queue.consume()) |
| ok_(not queue.holds_lock()) |
| ok_(not queue.consume()) |
| eq_(len(queue), 0) |
| |
| def test_consume(self): |
| queue = self._makeOne() |
| |
| queue.put(b"one") |
| ok_(not queue.consume()) |
| queue.get(.1) |
| ok_(queue.consume()) |
| ok_(not queue.consume()) |
| |
| def test_holds_lock(self): |
| queue = self._makeOne() |
| |
| ok_(not queue.holds_lock()) |
| queue.put(b"one") |
| queue.get(.1) |
| ok_(queue.holds_lock()) |
| queue.consume() |
| ok_(not queue.holds_lock()) |
| |
| def test_priority(self): |
| queue = self._makeOne() |
| queue.put(b"four", priority=101) |
| queue.put(b"one", priority=0) |
| queue.put(b"two", priority=0) |
| queue.put(b"three", priority=10) |
| |
| eq_(queue.get(1), b"one") |
| ok_(queue.consume()) |
| eq_(queue.get(1), b"two") |
| ok_(queue.consume()) |
| eq_(queue.get(1), b"three") |
| ok_(queue.consume()) |
| eq_(queue.get(1), b"four") |
| ok_(queue.consume()) |
| |
| def test_concurrent_execution(self): |
| queue = self._makeOne() |
| value1 = [] |
| value2 = [] |
| value3 = [] |
| event1 = self.client.handler.event_object() |
| event2 = self.client.handler.event_object() |
| event3 = self.client.handler.event_object() |
| |
| def get_concurrently(value, event): |
| q = self.client.LockingQueue(queue.path) |
| value.append(q.get(.1)) |
| event.set() |
| |
| self.client.handler.spawn(get_concurrently, value1, event1) |
| self.client.handler.spawn(get_concurrently, value2, event2) |
| self.client.handler.spawn(get_concurrently, value3, event3) |
| queue.put(b"one") |
| event1.wait(.2) |
| event2.wait(.2) |
| event3.wait(.2) |
| |
| result = value1 + value2 + value3 |
| eq_(result.count(b"one"), 1) |
| eq_(result.count(None), 2) |
