tests/test__coros_queue.py

   1 from tests import LimitedTestCase, silence_warnings
   2 from unittest import main
   3 import eventlet
   4 from eventlet import coros, spawn, sleep
   5 from eventlet.event import Event
   6
   7
   8 class TestQueue(LimitedTestCase):
   9
  10     @silence_warnings
  11     def test_send_first(self):
  12         q = coros.queue()
  13         q.send('hi')
  14         self.assertEqual(q.wait(), 'hi')
  15
  16     @silence_warnings
  17     def test_send_exception_first(self):
  18         q = coros.queue()
  19         q.send(exc=RuntimeError())
  20         self.assertRaises(RuntimeError, q.wait)
  21
  22     @silence_warnings
  23     def test_send_last(self):
  24         q = coros.queue()
  25
  26         def waiter(q):
  27             timer = eventlet.Timeout(0.1)
  28             self.assertEqual(q.wait(), 'hi2')
  29             timer.cancel()
  30
  31         spawn(waiter, q)
  32         sleep(0)
  33         sleep(0)
  34         q.send('hi2')
  35
  36     @silence_warnings
  37     def test_max_size(self):
  38         q = coros.queue(2)
  39         results = []
  40
  41         def putter(q):
  42             q.send('a')
  43             results.append('a')
  44             q.send('b')
  45             results.append('b')
  46             q.send('c')
  47             results.append('c')
  48
  49         spawn(putter, q)
  50         sleep(0)
  51         self.assertEqual(results, ['a', 'b'])
  52         self.assertEqual(q.wait(), 'a')
  53         sleep(0)
  54         self.assertEqual(results, ['a', 'b', 'c'])
  55         self.assertEqual(q.wait(), 'b')
  56         self.assertEqual(q.wait(), 'c')
  57
  58     @silence_warnings
  59     def test_zero_max_size(self):
  60         q = coros.queue(0)
  61
  62         def sender(evt, q):
  63             q.send('hi')
  64             evt.send('done')
  65
  66         def receiver(evt, q):
  67             x = q.wait()
  68             evt.send(x)
  69
  70         e1 = Event()
  71         e2 = Event()
  72
  73         spawn(sender, e1, q)
  74         sleep(0)
  75         assert not e1.ready()
  76         spawn(receiver, e2, q)
  77         self.assertEqual(e2.wait(), 'hi')
  78         self.assertEqual(e1.wait(), 'done')
  79
  80     @silence_warnings
  81     def test_multiple_waiters(self):
  82         # tests that multiple waiters get their results back
  83         q = coros.queue()
  84
  85         sendings = ['1', '2', '3', '4']
  86         gts = [eventlet.spawn(q.wait)
  87                for x in sendings]
  88
  89         eventlet.sleep(0.01)  # get 'em all waiting
  90
  91         q.send(sendings[0])
  92         q.send(sendings[1])
  93         q.send(sendings[2])
  94         q.send(sendings[3])
  95         results = set()
  96         for i, gt in enumerate(gts):
  97             results.add(gt.wait())
  98         self.assertEqual(results, set(sendings))
  99
 100     @silence_warnings
 101     def test_waiters_that_cancel(self):
 102         q = coros.queue()
 103
 104         def do_receive(q, evt):
 105             eventlet.Timeout(0, RuntimeError())
 106             try:
 107                 result = q.wait()
 108                 evt.send(result)
 109             except RuntimeError:
 110                 evt.send('timed out')
 111
 112         evt = Event()
 113         spawn(do_receive, q, evt)
 114         self.assertEqual(evt.wait(), 'timed out')
 115
 116         q.send('hi')
 117         self.assertEqual(q.wait(), 'hi')
 118
 119     @silence_warnings
 120     def test_senders_that_die(self):
 121         q = coros.queue()
 122
 123         def do_send(q):
 124             q.send('sent')
 125
 126         spawn(do_send, q)
 127         self.assertEqual(q.wait(), 'sent')
 128
 129     @silence_warnings
 130     def test_two_waiters_one_dies(self):
 131         def waiter(q, evt):
 132             evt.send(q.wait())
 133
 134         def do_receive(q, evt):
 135             eventlet.Timeout(0, RuntimeError())
 136             try:
 137                 result = q.wait()
 138                 evt.send(result)
 139             except RuntimeError:
 140                 evt.send('timed out')
 141
 142         q = coros.queue()
 143         dying_evt = Event()
 144         waiting_evt = Event()
 145         spawn(do_receive, q, dying_evt)
 146         spawn(waiter, q, waiting_evt)
 147         sleep(0)
 148         q.send('hi')
 149         self.assertEqual(dying_evt.wait(), 'timed out')
 150         self.assertEqual(waiting_evt.wait(), 'hi')
 151
 152     @silence_warnings
 153     def test_two_bogus_waiters(self):
 154         def do_receive(q, evt):
 155             eventlet.Timeout(0, RuntimeError())
 156             try:
 157                 result = q.wait()
 158                 evt.send(result)
 159             except RuntimeError:
 160                 evt.send('timed out')
 161
 162         q = coros.queue()
 163         e1 = Event()
 164         e2 = Event()
 165         spawn(do_receive, q, e1)
 166         spawn(do_receive, q, e2)
 167         sleep(0)
 168         q.send('sent')
 169         self.assertEqual(e1.wait(), 'timed out')
 170         self.assertEqual(e2.wait(), 'timed out')
 171         self.assertEqual(q.wait(), 'sent')
 172
 173     @silence_warnings
 174     def test_waiting(self):
 175         def do_wait(q, evt):
 176             result = q.wait()
 177             evt.send(result)
 178
 179         q = coros.queue()
 180         e1 = Event()
 181         spawn(do_wait, q, e1)
 182         sleep(0)
 183         self.assertEqual(1, q.waiting())
 184         q.send('hi')
 185         sleep(0)
 186         self.assertEqual(0, q.waiting())
 187         self.assertEqual('hi', e1.wait())
 188         self.assertEqual(0, q.waiting())
 189
 190
 191 class TestChannel(LimitedTestCase):
 192
 193     @silence_warnings
 194     def test_send(self):
 195         sleep(0.1)
 196         channel = coros.queue(0)
 197
 198         events = []
 199
 200         def another_greenlet():
 201             events.append(channel.wait())
 202             events.append(channel.wait())
 203
 204         spawn(another_greenlet)
 205
 206         events.append('sending')
 207         channel.send('hello')
 208         events.append('sent hello')
 209         channel.send('world')
 210         events.append('sent world')
 211
 212         self.assertEqual(['sending', 'hello', 'sent hello', 'world', 'sent world'], events)
 213
 214     @silence_warnings
 215     def test_wait(self):
 216         sleep(0.1)
 217         channel = coros.queue(0)
 218         events = []
 219
 220         def another_greenlet():
 221             events.append('sending hello')
 222             channel.send('hello')
 223             events.append('sending world')
 224             channel.send('world')
 225             events.append('sent world')
 226
 227         spawn(another_greenlet)
 228
 229         events.append('waiting')
 230         events.append(channel.wait())
 231         events.append(channel.wait())
 232
 233         self.assertEqual(['waiting', 'sending hello', 'hello', 'sending world', 'world'], events)
 234         sleep(0)
 235         self.assertEqual(['waiting', 'sending hello', 'hello', 'sending world', 'world', 'sent world'], events)
 236
 237     @silence_warnings
 238     def test_waiters(self):
 239         c = coros.Channel()
 240         w1 = eventlet.spawn(c.wait)
 241         w2 = eventlet.spawn(c.wait)
 242         w3 = eventlet.spawn(c.wait)
 243         sleep(0)
 244         self.assertEqual(c.waiting(), 3)
 245         s1 = eventlet.spawn(c.send, 1)
 246         s2 = eventlet.spawn(c.send, 2)
 247         s3 = eventlet.spawn(c.send, 3)
 248         sleep(0)  # this gets all the sends into a waiting state
 249         self.assertEqual(c.waiting(), 0)
 250
 251         s1.wait()
 252         s2.wait()
 253         s3.wait()
 254         # NOTE: we don't guarantee that waiters are served in order
 255         results = sorted([w1.wait(), w2.wait(), w3.wait()])
 256         self.assertEqual(results, [1, 2, 3])
 257
 258 if __name__ == '__main__':
 259     main()