import weakref
from eventlet.green import thread
from eventlet import greenthread
from eventlet import event
import eventlet
from eventlet import corolocal

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.assertEquals("foo1", my_local.callme())
        
        def do_something():
            my_local.foo = "foo2"
            self.assertEquals("foo2", my_local.callme())
            
        eventlet.spawn(do_something).wait()        
            
        my_local.foo = "foo3"
        self.assertEquals("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 xrange(100):
            p.spawn(do_something, i)
        p.waitall()
        del p
        # at this point all our coros have terminated
        self.assertEqual(len(refs), 1)