/usr/lib/python2.7/dist-packages/kazoo/tests/test_eventlet_handler.py is in python-kazoo 2.2.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 | import contextlib
import unittest
from nose import SkipTest
from nose.tools import raises
from kazoo.client import KazooClient
from kazoo.handlers import utils
from kazoo.protocol import states as kazoo_states
from kazoo.tests import test_client
from kazoo.tests import test_lock
from kazoo.tests import util as test_util
try:
import eventlet
from eventlet.green import threading
from kazoo.handlers import eventlet as eventlet_handler
EVENTLET_HANDLER_AVAILABLE = True
except ImportError:
EVENTLET_HANDLER_AVAILABLE = False
@contextlib.contextmanager
def start_stop_one(handler=None):
if not handler:
handler = eventlet_handler.SequentialEventletHandler()
handler.start()
try:
yield handler
finally:
handler.stop()
class TestEventletHandler(unittest.TestCase):
def setUp(self):
if not EVENTLET_HANDLER_AVAILABLE:
raise SkipTest('eventlet handler not available.')
super(TestEventletHandler, self).setUp()
def test_started(self):
with start_stop_one() as handler:
self.assertTrue(handler.running)
self.assertNotEqual(0, len(handler._workers))
self.assertFalse(handler.running)
self.assertEqual(0, len(handler._workers))
def test_spawn(self):
captures = []
def cb():
captures.append(1)
with start_stop_one() as handler:
handler.spawn(cb)
self.assertEqual(1, len(captures))
def test_dispatch(self):
captures = []
def cb():
captures.append(1)
with start_stop_one() as handler:
handler.dispatch_callback(kazoo_states.Callback('watch', cb, []))
self.assertEqual(1, len(captures))
def test_async_link(self):
captures = []
def cb(handler):
captures.append(handler)
with start_stop_one() as handler:
r = handler.async_result()
r.rawlink(cb)
r.set(2)
self.assertEqual(1, len(captures))
self.assertEqual(2, r.get())
def test_timeout_raising(self):
handler = eventlet_handler.SequentialEventletHandler()
@raises(handler.timeout_exception)
def raise_it():
raise handler.timeout_exception("This is a timeout")
raise_it()
def test_async_ok(self):
captures = []
def delayed():
captures.append(1)
return 1
def after_delayed(handler):
captures.append(handler)
with start_stop_one() as handler:
r = handler.async_result()
r.rawlink(after_delayed)
w = handler.spawn(utils.wrap(r)(delayed))
w.join()
self.assertEqual(2, len(captures))
self.assertEqual(1, captures[0])
self.assertEqual(1, r.get())
def test_get_with_no_block(self):
handler = eventlet_handler.SequentialEventletHandler()
@raises(handler.timeout_exception)
def test_no_block(r):
r.get(block=False)
with start_stop_one(handler):
r = handler.async_result()
test_no_block(r)
r.set(1)
self.assertEqual(1, r.get())
def test_async_exception(self):
@raises(IOError)
def check_exc(r):
r.get()
def broken():
raise IOError("Failed")
with start_stop_one() as handler:
r = handler.async_result()
w = handler.spawn(utils.wrap(r)(broken))
w.join()
self.assertFalse(r.successful())
check_exc(r)
class TestEventletClient(test_client.TestClient):
def setUp(self):
if not EVENTLET_HANDLER_AVAILABLE:
raise SkipTest('eventlet handler not available.')
super(TestEventletClient, self).setUp()
@staticmethod
def make_event():
return threading.Event()
@staticmethod
def make_condition():
return threading.Condition()
def _makeOne(self, *args):
return eventlet_handler.SequentialEventletHandler(*args)
def _get_client(self, **kwargs):
kwargs["handler"] = self._makeOne()
return KazooClient(self.hosts, **kwargs)
class TestEventletSemaphore(test_lock.TestSemaphore):
def setUp(self):
if not EVENTLET_HANDLER_AVAILABLE:
raise SkipTest('eventlet handler not available.')
super(TestEventletSemaphore, self).setUp()
@staticmethod
def make_condition():
return threading.Condition()
@staticmethod
def make_event():
return threading.Event()
@staticmethod
def make_thread(*args, **kwargs):
return threading.Thread(*args, **kwargs)
def _makeOne(self, *args):
return eventlet_handler.SequentialEventletHandler(*args)
def _get_client(self, **kwargs):
kwargs["handler"] = self._makeOne()
c = KazooClient(self.hosts, **kwargs)
try:
self._clients.append(c)
except AttributeError:
self._client = [c]
return c
class TestEventletLock(test_lock.KazooLockTests):
def setUp(self):
if not EVENTLET_HANDLER_AVAILABLE:
raise SkipTest('eventlet handler not available.')
super(TestEventletLock, self).setUp()
@staticmethod
def make_condition():
return threading.Condition()
@staticmethod
def make_event():
return threading.Event()
@staticmethod
def make_thread(*args, **kwargs):
return threading.Thread(*args, **kwargs)
@staticmethod
def make_wait():
return test_util.Wait(getsleep=(lambda: eventlet.sleep))
def _makeOne(self, *args):
return eventlet_handler.SequentialEventletHandler(*args)
def _get_client(self, **kwargs):
kwargs["handler"] = self._makeOne()
c = KazooClient(self.hosts, **kwargs)
try:
self._clients.append(c)
except AttributeError:
self._client = [c]
return c
|