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, skipped
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 @skipped # cause it relies on internal details of corolocal that are no longer true
28 def test_simple(self):
34 g_id = id(greenthread.getcurrent())
39 thread.start_new_thread(setter, args=(tls, 1))
40 thread.start_new_thread(setter, args=(tls, 2))
42 objs = object.__getattribute__(tls, "__objs")
43 assert sorted(g_ids) == sorted(objs.keys())
44 assert objs[g_ids[0]]['value'] == 1
45 assert objs[g_ids[1]]['value'] == 2
46 assert getattr(tls, 'value', None) is None
50 def test_assignment(self):
51 my_local = corolocal.local()
56 self.assertEqual(my_local.b, 2)
60 except AttributeError:
63 eventlet.spawn(do_something).wait()
64 self.assertEqual(my_local.a, 1)
66 def test_calls_init(self):
69 class Init(corolocal.local):
70 def __init__(self, *args):
71 init_args.append((args, eventlet.getcurrent()))
73 my_local = Init(1, 2, 3)
74 self.assertEqual(init_args[0][0], (1, 2, 3))
75 self.assertEqual(init_args[0][1], eventlet.getcurrent())
79 self.assertEqual(len(init_args), 2, init_args)
80 self.assertEqual(init_args[1][0], (1, 2, 3))
81 self.assertEqual(init_args[1][1], eventlet.getcurrent())
83 eventlet.spawn(do_something).wait()
85 def test_calling_methods(self):
86 class Caller(corolocal.local):
92 self.assertEqual("foo1", my_local.callme())
96 self.assertEqual("foo2", my_local.callme())
98 eventlet.spawn(do_something).wait()
100 my_local.foo = "foo3"
101 self.assertEqual("foo3", my_local.callme())
103 def test_no_leaking(self):
104 refs = weakref.WeakKeyDictionary()
105 my_local = corolocal.local()
115 p = eventlet.GreenPool()
116 for i in six.moves.range(100):
117 p.spawn(do_something, i)
123 # at this point all our coros have terminated
124 self.assertEqual(len(refs), 1)