Add python-eventlet package to MOS 8.0 repository

[packages/trusty/python-eventlet.git] / eventlet / tests / thread_test.py

diff --git a/eventlet/tests/thread_test.py b/eventlet/tests/thread_test.py
deleted file mode 100644 (file)
index caf8dbf..0000000
--- a/eventlet/tests/thread_test.py
+++ /dev/null
@@ -1,124 +0,0 @@
-import gc
-import weakref
-
-import eventlet
-from eventlet import corolocal
-from eventlet import event
-from eventlet import greenthread
-from eventlet.green import thread
-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")
-        assert sorted(g_ids) == sorted(objs.keys())
-        assert objs[g_ids[0]]['value'] == 1
-        assert objs[g_ids[1]]['value'] == 2
-        assert getattr(tls, 'value', None) is None
-        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)