|
|
import uuid
import pytest
from kazoo.testing import KazooTestCasefrom kazoo.tests.util import CI_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() with pytest.raises(TypeError): queue.put({}) with pytest.raises(TypeError): queue.put(b"one", b"100") with pytest.raises(TypeError): queue.put(b"one", 10.0) with pytest.raises(ValueError): queue.put(b"one", -100) with pytest.raises(ValueError): queue.put(b"one", 100000)
def test_empty_queue(self): queue = self._makeOne() assert len(queue) == 0 assert queue.get() is None assert len(queue) == 0
def test_queue(self): queue = self._makeOne() queue.put(b"one") queue.put(b"two") queue.put(b"three") assert len(queue) == 3
assert queue.get() == b"one" assert queue.get() == b"two" assert queue.get() == b"three" assert 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)
assert queue.get() == b"one" assert queue.get() == b"two" assert queue.get() == b"three" assert queue.get() == b"four"
class KazooLockingQueueTests(KazooTestCase): def setUp(self): KazooTestCase.setUp(self) skip = False if CI_ZK_VERSION and CI_ZK_VERSION < (3, 4): skip = True elif CI_ZK_VERSION and CI_ZK_VERSION >= (3, 4): skip = False else: ver = self.client.server_version() if ver[1] < 4: skip = True if skip: pytest.skip("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() with pytest.raises(TypeError): queue.put({}) with pytest.raises(TypeError): queue.put(b"one", b"100") with pytest.raises(TypeError): queue.put(b"one", 10.0) with pytest.raises(ValueError): queue.put(b"one", -100) with pytest.raises(ValueError): queue.put(b"one", 100000) with pytest.raises(TypeError): queue.put_all({}) with pytest.raises(TypeError): queue.put_all([{}]) with pytest.raises(TypeError): queue.put_all([b"one"], b"100") with pytest.raises(TypeError): queue.put_all([b"one"], 10.0) with pytest.raises(ValueError): queue.put_all([b"one"], -100) with pytest.raises(ValueError): queue.put_all([b"one"], 100000)
def test_empty_queue(self): queue = self._makeOne() assert len(queue) == 0 assert queue.get(0) is None assert len(queue) == 0
def test_queue(self): queue = self._makeOne() queue.put(b"one") queue.put_all([b"two", b"three"]) assert len(queue) == 3
assert not queue.consume() assert not queue.holds_lock() assert queue.get(1) == b"one" assert queue.holds_lock() # Without consuming, should return the same element assert queue.get(1) == b"one" assert queue.consume() assert not queue.holds_lock() assert queue.get(1) == b"two" assert queue.holds_lock() assert queue.consume() assert not queue.holds_lock() assert queue.get(1) == b"three" assert queue.holds_lock() assert queue.consume() assert not queue.holds_lock() assert not queue.consume() assert len(queue) == 0
def test_consume(self): queue = self._makeOne()
queue.put(b"one") assert not queue.consume() queue.get(0.1) assert queue.consume() assert not queue.consume()
def test_release(self): queue = self._makeOne()
queue.put(b"one") assert queue.get(1) == b"one" assert queue.holds_lock() assert queue.release() assert not queue.holds_lock() assert queue.get(1) == b"one" assert queue.consume() assert not queue.release() assert len(queue) == 0
def test_holds_lock(self): queue = self._makeOne()
assert not queue.holds_lock() queue.put(b"one") queue.get(0.1) assert queue.holds_lock() queue.consume() assert 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)
assert queue.get(1) == b"one" assert queue.consume() assert queue.get(1) == b"two" assert queue.consume() assert queue.get(1) == b"three" assert queue.consume() assert queue.get(1) == b"four" assert 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(0.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(0.2) event2.wait(0.2) event3.wait(0.2)
result = value1 + value2 + value3 assert result.count(b"one") == 1 assert result.count(None) == 2
|
