python-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
  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     def test_assignment(self):
  28         my_local = corolocal.local()
  29         my_local.a = 1
  30
  31         def do_something():
  32             my_local.b = 2
  33             self.assertEqual(my_local.b, 2)
  34             try:
  35                 my_local.a
  36                 self.fail()
  37             except AttributeError:
  38                 pass
  39
  40         eventlet.spawn(do_something).wait()
  41         self.assertEqual(my_local.a, 1)
  42
  43     def test_calls_init(self):
  44         init_args = []
  45
  46         class Init(corolocal.local):
  47             def __init__(self, *args):
  48                 init_args.append((args, eventlet.getcurrent()))
  49
  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())
  53
  54         def do_something():
  55             my_local.foo = 'bar'
  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())
  59
  60         eventlet.spawn(do_something).wait()
  61
  62     def test_calling_methods(self):
  63         class Caller(corolocal.local):
  64             def callme(self):
  65                 return self.foo
  66
  67         my_local = Caller()
  68         my_local.foo = "foo1"
  69         self.assertEqual("foo1", my_local.callme())
  70
  71         def do_something():
  72             my_local.foo = "foo2"
  73             self.assertEqual("foo2", my_local.callme())
  74
  75         eventlet.spawn(do_something).wait()
  76
  77         my_local.foo = "foo3"
  78         self.assertEqual("foo3", my_local.callme())
  79
  80     def test_no_leaking(self):
  81         refs = weakref.WeakKeyDictionary()
  82         my_local = corolocal.local()
  83
  84         class X(object):
  85             pass
  86
  87         def do_something(i):
  88             o = X()
  89             refs[o] = True
  90             my_local.foo = o
  91
  92         p = eventlet.GreenPool()
  93         for i in six.moves.range(100):
  94             p.spawn(do_something, i)
  95         p.waitall()
  96         del p
  97         gc.collect()
  98         eventlet.sleep(0)
  99         gc.collect()
 100         # at this point all our coros have terminated
 101         self.assertEqual(len(refs), 1)