You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
205 lines
6.1 KiB
205 lines
6.1 KiB
import uuid
|
|
|
|
import pytest
|
|
|
|
from kazoo.testing import KazooTestCase
|
|
from 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
|