cbd404d56e
- __next__ for iterator interface - six.next() to get next item - list(dict.keys()) - popen2.popen4 -> subprocess - s2b -> b"..." literals - deprecated assertEquals -> assertEqual - hub_test test_fork using run_python - 1L -> 1 long literal - many PEP-8 fixes
118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
import gc
|
|
import weakref
|
|
|
|
from eventlet.green import thread
|
|
from eventlet import greenthread
|
|
from eventlet import event
|
|
import eventlet
|
|
from eventlet import corolocal
|
|
from eventlet.support import six
|
|
|
|
from tests import LimitedTestCase, skipped
|
|
|
|
|
|
class Locals(LimitedTestCase):
|
|
def passthru(self, *args, **kw):
|
|
self.results.append((args, kw))
|
|
return args, kw
|
|
|
|
def setUp(self):
|
|
self.results = []
|
|
super(Locals, self).setUp()
|
|
|
|
def tearDown(self):
|
|
self.results = []
|
|
super(Locals, self).tearDown()
|
|
|
|
@skipped # cause it relies on internal details of corolocal that are no longer true
|
|
def test_simple(self):
|
|
tls = thread._local()
|
|
g_ids = []
|
|
evt = event.Event()
|
|
def setter(tls, v):
|
|
g_id = id(greenthread.getcurrent())
|
|
g_ids.append(g_id)
|
|
tls.value = v
|
|
evt.wait()
|
|
thread.start_new_thread(setter, args=(tls, 1))
|
|
thread.start_new_thread(setter, args=(tls, 2))
|
|
eventlet.sleep()
|
|
objs = object.__getattribute__(tls, "__objs")
|
|
self.failUnlessEqual(sorted(g_ids), sorted(objs.keys()))
|
|
self.failUnlessEqual(objs[g_ids[0]]['value'], 1)
|
|
self.failUnlessEqual(objs[g_ids[1]]['value'], 2)
|
|
self.failUnlessRaises(AttributeError, lambda: tls.value)
|
|
evt.send("done")
|
|
eventlet.sleep()
|
|
|
|
def test_assignment(self):
|
|
my_local = corolocal.local()
|
|
my_local.a = 1
|
|
def do_something():
|
|
my_local.b = 2
|
|
self.assertEqual(my_local.b, 2)
|
|
try:
|
|
my_local.a
|
|
self.fail()
|
|
except AttributeError:
|
|
pass
|
|
eventlet.spawn(do_something).wait()
|
|
self.assertEqual(my_local.a, 1)
|
|
|
|
def test_calls_init(self):
|
|
init_args = []
|
|
class Init(corolocal.local):
|
|
def __init__(self, *args):
|
|
init_args.append((args, eventlet.getcurrent()))
|
|
|
|
my_local = Init(1,2,3)
|
|
self.assertEqual(init_args[0][0], (1,2,3))
|
|
self.assertEqual(init_args[0][1], eventlet.getcurrent())
|
|
|
|
def do_something():
|
|
my_local.foo = 'bar'
|
|
self.assertEqual(len(init_args), 2, init_args)
|
|
self.assertEqual(init_args[1][0], (1,2,3))
|
|
self.assertEqual(init_args[1][1], eventlet.getcurrent())
|
|
|
|
eventlet.spawn(do_something).wait()
|
|
|
|
def test_calling_methods(self):
|
|
class Caller(corolocal.local):
|
|
def callme(self):
|
|
return self.foo
|
|
|
|
my_local = Caller()
|
|
my_local.foo = "foo1"
|
|
self.assertEqual("foo1", my_local.callme())
|
|
|
|
def do_something():
|
|
my_local.foo = "foo2"
|
|
self.assertEqual("foo2", my_local.callme())
|
|
|
|
eventlet.spawn(do_something).wait()
|
|
|
|
my_local.foo = "foo3"
|
|
self.assertEqual("foo3", my_local.callme())
|
|
|
|
def test_no_leaking(self):
|
|
refs = weakref.WeakKeyDictionary()
|
|
my_local = corolocal.local()
|
|
class X(object):
|
|
pass
|
|
def do_something(i):
|
|
o = X()
|
|
refs[o] = True
|
|
my_local.foo = o
|
|
|
|
p = eventlet.GreenPool()
|
|
for i in six.moves.range(100):
|
|
p.spawn(do_something, i)
|
|
p.waitall()
|
|
del p
|
|
gc.collect()
|
|
eventlet.sleep(0)
|
|
gc.collect()
|
|
# at this point all our coros have terminated
|
|
self.assertEqual(len(refs), 1)
|