eventlet/tests/thread_test.py

   1 import gc
   2 import weakref
   3
   4 import eventlet
   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
  10
  11 from tests import LimitedTestCase, skipped
  12
  13
  14 class Locals(LimitedTestCase):
  15     def passthru(self, *args, **kw):
  16         self.results.append((args, kw))
  17         return args, kw
  18
  19     def setUp(self):
  20         self.results = []
  21         super(Locals, self).setUp()
  22
  23     def tearDown(self):
  24         self.results = []
  25         super(Locals, self).tearDown()
  26
  27     @skipped  # cause it relies on internal details of corolocal that are no longer true
  28     def test_simple(self):
  29         tls = thread._local()
  30         g_ids = []
  31         evt = event.Event()
  32
  33         def setter(tls, v):
  34             g_id = id(greenthread.getcurrent())
  35             g_ids.append(g_id)
  36             tls.value = v
  37             evt.wait()
  38
  39         thread.start_new_thread(setter, args=(tls, 1))
  40         thread.start_new_thread(setter, args=(tls, 2))
  41         eventlet.sleep()
  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
  47         evt.send("done")
  48         eventlet.sleep()
  49
  50     def test_assignment(self):
  51         my_local = corolocal.local()
  52         my_local.a = 1
  53
  54         def do_something():
  55             my_local.b = 2
  56             self.assertEqual(my_local.b, 2)
  57             try:
  58                 my_local.a
  59                 self.fail()
  60             except AttributeError:
  61                 pass
  62
  63         eventlet.spawn(do_something).wait()
  64         self.assertEqual(my_local.a, 1)
  65
  66     def test_calls_init(self):
  67         init_args = []
  68
  69         class Init(corolocal.local):
  70             def __init__(self, *args):
  71                 init_args.append((args, eventlet.getcurrent()))
  72
  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())
  76
  77         def do_something():
  78             my_local.foo = 'bar'
  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())
  82
  83         eventlet.spawn(do_something).wait()
  84
  85     def test_calling_methods(self):
  86         class Caller(corolocal.local):
  87             def callme(self):
  88                 return self.foo
  89
  90         my_local = Caller()
  91         my_local.foo = "foo1"
  92         self.assertEqual("foo1", my_local.callme())
  93
  94         def do_something():
  95             my_local.foo = "foo2"
  96             self.assertEqual("foo2", my_local.callme())
  97
  98         eventlet.spawn(do_something).wait()
  99
 100         my_local.foo = "foo3"
 101         self.assertEqual("foo3", my_local.callme())
 102
 103     def test_no_leaking(self):
 104         refs = weakref.WeakKeyDictionary()
 105         my_local = corolocal.local()
 106
 107         class X(object):
 108             pass
 109
 110         def do_something(i):
 111             o = X()
 112             refs[o] = True
 113             my_local.foo = o
 114
 115         p = eventlet.GreenPool()
 116         for i in six.moves.range(100):
 117             p.spawn(do_something, i)
 118         p.waitall()
 119         del p
 120         gc.collect()
 121         eventlet.sleep(0)
 122         gc.collect()
 123         # at this point all our coros have terminated
 124         self.assertEqual(len(refs), 1)