1 from unittest import TestCase, main
4 from eventlet import Queue
5 from eventlet import pools
6 from eventlet.support import six
9 class IntPool(pools.Pool):
11 self.current_integer = getattr(self, 'current_integer', 0) + 1
12 return self.current_integer
15 class TestIntPool(TestCase):
17 self.pool = IntPool(min_size=0, max_size=4)
19 def test_integers(self):
20 # Do not actually use this pattern in your code. The pool will be
21 # exhausted, and unrestoreable.
22 # If you do a get, you should ALWAYS do a put, probably like this:
24 # thing = self.pool.get()
27 # self.pool.put(thing)
29 # with self.pool.some_api_name() as thing:
31 self.assertEqual(self.pool.get(), 1)
32 self.assertEqual(self.pool.get(), 2)
33 self.assertEqual(self.pool.get(), 3)
34 self.assertEqual(self.pool.get(), 4)
37 self.assertEqual(self.pool.free(), 4)
38 gotten = self.pool.get()
39 self.assertEqual(self.pool.free(), 3)
41 self.assertEqual(self.pool.free(), 4)
43 def test_exhaustion(self):
49 gotten = self.pool.get()
53 eventlet.spawn(consumer)
55 one, two, three, four = (
56 self.pool.get(), self.pool.get(), self.pool.get(), self.pool.get())
57 self.assertEqual(self.pool.free(), 0)
59 # Let consumer run; nothing will be in the pool, so he will wait
65 # wait for the consumer
66 self.assertEqual(waiter.get(), one)
68 def test_blocks_on_pool(self):
76 # No one should be waiting yet.
77 self.assertEqual(self.pool.waiting(), 0)
78 # The call to the next get will unschedule this routine.
80 # So this put should never be called.
83 killable = eventlet.spawn(greedy)
85 # no one should be waiting yet.
86 self.assertEqual(self.pool.waiting(), 0)
91 # Greedy should be blocking on the last get
92 self.assertEqual(self.pool.waiting(), 1)
94 # Send will never be called, so balance should be 0.
95 self.assertFalse(not waiter.full())
97 eventlet.kill(killable)
99 def test_ordering(self):
100 # normal case is that items come back out in the
101 # same order they are put
102 one, two = self.pool.get(), self.pool.get()
105 self.assertEqual(self.pool.get(), one)
106 self.assertEqual(self.pool.get(), two)
108 def test_putting_to_queue(self):
109 timer = eventlet.Timeout(0.1)
112 self.pool = IntPool(min_size=0, max_size=size)
116 def just_put(pool_item, index):
117 self.pool.put(pool_item)
119 for index in six.moves.range(size + 1):
120 pool_item = self.pool.get()
121 eventlet.spawn(just_put, pool_item, index)
123 for _ in six.moves.range(size + 1):
126 self.assertEqual(sorted(results), list(six.moves.range(size + 1)))
130 def test_resize(self):
131 pool = IntPool(max_size=2)
134 self.assertEqual(pool.free(), 0)
136 # verify that the pool discards excess items put into it
140 self.assertEqual(pool.free(), 1)
142 # resize larger and assert that there are more free items
144 self.assertEqual(pool.free(), 2)
146 def test_create_contention(self):
154 p = pools.Pool(max_size=4, create=sleep_create)
158 self.assertEqual(x, "slept")
161 gp = eventlet.GreenPool()
162 for i in six.moves.range(100):
165 self.assertEqual(creates[0], 4)
168 class TestAbstract(TestCase):
171 def test_abstract(self):
172 # Going for 100% coverage here
173 # A Pool cannot be used without overriding create()
175 self.assertRaises(NotImplementedError, pool.get)
178 class TestIntPool2(TestCase):
182 self.pool = IntPool(min_size=3, max_size=3)
184 def test_something(self):
185 self.assertEqual(len(self.pool.free_items), 3)
186 # Cover the clause in get where we get from the free list instead of creating
188 gotten = self.pool.get()
189 self.assertEqual(gotten, 1)
192 class TestOrderAsStack(TestCase):
196 self.pool = IntPool(max_size=3, order_as_stack=True)
198 def test_ordering(self):
199 # items come out in the reverse order they are put
200 one, two = self.pool.get(), self.pool.get()
203 self.assertEqual(self.pool.get(), two)
204 self.assertEqual(self.pool.get(), one)
207 class RaisePool(pools.Pool):
212 class TestCreateRaises(TestCase):
216 self.pool = RaisePool(max_size=3)
219 self.assertEqual(self.pool.free(), 3)
220 self.assertRaises(RuntimeError, self.pool.get)
221 self.assertEqual(self.pool.free(), 3)
224 ALWAYS = RuntimeError('I always fail')
225 SOMETIMES = RuntimeError('I fail half the time')
228 class TestTookTooLong(Exception):
231 if __name__ == '__main__':