1 from tests import LimitedTestCase
2 from eventlet import greenthread
3 from eventlet.support import greenlets as greenlet
8 def passthru(*args, **kw):
9 _g_results.append((args, kw))
14 greenthread.sleep(0.1)
18 class Asserts(object):
19 def assert_dead(self, gt):
20 if hasattr(gt, 'wait'):
21 self.assertRaises(greenlet.GreenletExit, gt.wait)
26 class Spawn(LimitedTestCase, Asserts):
29 super(Spawn, self).tearDown()
32 def test_simple(self):
33 gt = greenthread.spawn(passthru, 1, b=2)
34 self.assertEqual(gt.wait(), ((1,), {'b': 2}))
35 self.assertEqual(_g_results, [((1,), {'b': 2})])
38 gt = greenthread.spawn_n(passthru, 2, b=3)
42 self.assertEqual(_g_results, [((2,), {'b': 3})])
45 gt = greenthread.spawn(passthru, 6)
48 greenthread.sleep(0.001)
49 self.assertEqual(_g_results, [])
53 def test_kill_meth(self):
54 gt = greenthread.spawn(passthru, 6)
57 greenthread.sleep(0.001)
58 self.assertEqual(_g_results, [])
62 def test_kill_n(self):
63 gt = greenthread.spawn_n(passthru, 7)
66 greenthread.sleep(0.001)
67 self.assertEqual(_g_results, [])
74 def link_func(g, *a, **kw):
78 gt = greenthread.spawn(passthru, 5)
79 gt.link(link_func, 4, b=5)
80 self.assertEqual(gt.wait(), ((5,), {}))
81 self.assertEqual(results, [gt, (4,), {'b': 5}])
83 def test_link_after_exited(self):
86 def link_func(g, *a, **kw):
90 gt = greenthread.spawn(passthru, 5)
91 self.assertEqual(gt.wait(), ((5,), {}))
92 gt.link(link_func, 4, b=5)
93 self.assertEqual(results, [gt, (4,), {'b': 5}])
95 def test_link_relinks(self):
96 # test that linking in a linked func doesn't cause infinite recursion.
100 g.link(link_func_pass)
102 def link_func_pass(g):
105 gt = greenthread.spawn(passthru)
108 self.assertEqual(called, [True])
111 class SpawnAfter(Spawn):
112 def test_basic(self):
113 gt = greenthread.spawn_after(0.1, passthru, 20)
114 self.assertEqual(gt.wait(), ((20,), {}))
116 def test_cancel(self):
117 gt = greenthread.spawn_after(0.1, passthru, 21)
121 def test_cancel_already_started(self):
122 gt = greenthread.spawn_after(0, waiter, 22)
125 self.assertEqual(gt.wait(), 22)
127 def test_kill_already_started(self):
128 gt = greenthread.spawn_after(0, waiter, 22)
134 class SpawnAfterLocal(LimitedTestCase, Asserts):
136 super(SpawnAfterLocal, self).setUp()
139 def test_timer_fired(self):
141 greenthread.spawn_after_local(0.1, self.lst.pop)
142 greenthread.sleep(0.2)
144 greenthread.spawn(func)
145 assert self.lst == [1], self.lst
146 greenthread.sleep(0.3)
147 assert self.lst == [], self.lst
149 def test_timer_cancelled_upon_greenlet_exit(self):
151 greenthread.spawn_after_local(0.1, self.lst.pop)
153 greenthread.spawn(func)
154 assert self.lst == [1], self.lst
155 greenthread.sleep(0.2)
156 assert self.lst == [1], self.lst
158 def test_spawn_is_not_cancelled(self):
160 greenthread.spawn(self.lst.pop)
161 # exiting immediatelly, but self.lst.pop must be called
162 greenthread.spawn(func)
163 greenthread.sleep(0.1)
164 assert self.lst == [], self.lst