5 from eventlet import corolocal
6 from eventlet import event
7 from eventlet import greenthread
8 from eventlet.green import thread
9 from eventlet.support import six
11 from tests import LimitedTestCase
14 class Locals(LimitedTestCase):
15 def passthru(self, *args, **kw):
16 self.results.append((args, kw))
21 super(Locals, self).setUp()
25 super(Locals, self).tearDown()
27 def test_assignment(self):
28 my_local = corolocal.local()
33 self.assertEqual(my_local.b, 2)
37 except AttributeError:
40 eventlet.spawn(do_something).wait()
41 self.assertEqual(my_local.a, 1)
43 def test_calls_init(self):
46 class Init(corolocal.local):
47 def __init__(self, *args):
48 init_args.append((args, eventlet.getcurrent()))
50 my_local = Init(1, 2, 3)
51 self.assertEqual(init_args[0][0], (1, 2, 3))
52 self.assertEqual(init_args[0][1], eventlet.getcurrent())
56 self.assertEqual(len(init_args), 2, init_args)
57 self.assertEqual(init_args[1][0], (1, 2, 3))
58 self.assertEqual(init_args[1][1], eventlet.getcurrent())
60 eventlet.spawn(do_something).wait()
62 def test_calling_methods(self):
63 class Caller(corolocal.local):
69 self.assertEqual("foo1", my_local.callme())
73 self.assertEqual("foo2", my_local.callme())
75 eventlet.spawn(do_something).wait()
78 self.assertEqual("foo3", my_local.callme())
80 def test_no_leaking(self):
81 refs = weakref.WeakKeyDictionary()
82 my_local = corolocal.local()
92 p = eventlet.GreenPool()
93 for i in six.moves.range(100):
94 p.spawn(do_something, i)
100 # at this point all our coros have terminated
101 self.assertEqual(len(refs), 1)