import gc

import sys

import unittest

import UserList

import weakref

import operator

from test import test_support

# Used in ReferencesTestCase.test_ref_created_during_del() .

ref_from_del = None

class C:

    def method(self):

        pass

class Callable:

    bar = None

    def __call__(self, x):

        self.bar = x

def create_function():

    def f(): pass

    return f

def create_bound_method():

    return C().method

def create_unbound_method():

    return C.method

class TestBase(unittest.TestCase):

    def setUp(self):

        self.cbcalled = 0

    def callback(self, ref):

        self.cbcalled += 1

class ReferencesTestCase(TestBase):

    def test_basic_ref(self):

        self.check_basic_ref(C)

        self.check_basic_ref(create_function)

        self.check_basic_ref(create_bound_method)

        self.check_basic_ref(create_unbound_method)

        # Just make sure the tp_repr handler doesn't raise an exception.

        # Live reference:

        o = C()

        wr = weakref.ref(o)

        `wr`

        # Dead reference:

        del o

        `wr`

    def test_basic_callback(self):

        self.check_basic_callback(C)

        self.check_basic_callback(create_function)

        self.check_basic_callback(create_bound_method)

        self.check_basic_callback(create_unbound_method)

    def test_multiple_callbacks(self):

        o = C()

        ref1 = weakref.ref(o, self.callback)

        ref2 = weakref.ref(o, self.callback)

        del o

        self.assert_(ref1() is None,

                     "expected reference to be invalidated")

        self.assert_(ref2() is None,

                     "expected reference to be invalidated")

        self.assert_(self.cbcalled == 2,

                     "callback not called the right number of times")

    def test_multiple_selfref_callbacks(self):

        # Make sure all references are invalidated before callbacks are called

        #

        # What's important here is that we're using the first

        # reference in the callback invoked on the second reference

        # (the most recently created ref is cleaned up first).  This

        # tests that all references to the object are invalidated

        # before any of the callbacks are invoked, so that we only

        # have one invocation of _weakref.c:cleanup_helper() active

        # for a particular object at a time.

        #

        def callback(object, self=self):

            self.ref()

        c = C()

        self.ref = weakref.ref(c, callback)

        ref1 = weakref.ref(c, callback)

        del c

    def test_proxy_ref(self):

        o = C()

        o.bar = 1

        ref1 = weakref.proxy(o, self.callback)

        ref2 = weakref.proxy(o, self.callback)

        del o

        def check(proxy):

            proxy.bar

        self.assertRaises(weakref.ReferenceError, check, ref1)

        self.assertRaises(weakref.ReferenceError, check, ref2)

        self.assertRaises(weakref.ReferenceError, bool, weakref.proxy(C()))

        self.assert_(self.cbcalled == 2)

    def check_basic_ref(self, factory):

        o = factory()

        ref = weakref.ref(o)

        self.assert_(ref() is not None,

                     "weak reference to live object should be live")

        o2 = ref()

        self.assert_(o is o2,

                     "<ref>() should return original object if live")

    def check_basic_callback(self, factory):

        self.cbcalled = 0

        o = factory()

        ref = weakref.ref(o, self.callback)

        del o

        self.assert_(self.cbcalled == 1,

                     "callback did not properly set 'cbcalled'")

        self.assert_(ref() is None,

                     "ref2 should be dead after deleting object reference")

    def test_ref_reuse(self):

        o = C()

        ref1 = weakref.ref(o)

        # create a proxy to make sure that there's an intervening creation

        # between these two; it should make no difference

        proxy = weakref.proxy(o)

        ref2 = weakref.ref(o)

        self.assert_(ref1 is ref2,

                     "reference object w/out callback should be re-used")

        o = C()

        proxy = weakref.proxy(o)

        ref1 = weakref.ref(o)

        ref2 = weakref.ref(o)

        self.assert_(ref1 is ref2,

                     "reference object w/out callback should be re-used")

        self.assert_(weakref.getweakrefcount(o) == 2,

                     "wrong weak ref count for object")

        del proxy

        self.assert_(weakref.getweakrefcount(o) == 1,

                     "wrong weak ref count for object after deleting proxy")

    def test_proxy_reuse(self):

        o = C()

        proxy1 = weakref.proxy(o)

        ref = weakref.ref(o)

        proxy2 = weakref.proxy(o)

        self.assert_(proxy1 is proxy2,

                     "proxy object w/out callback should have been re-used")

    def test_basic_proxy(self):

        o = C()

        self.check_proxy(o, weakref.proxy(o))

        L = UserList.UserList()

        p = weakref.proxy(L)

        self.failIf(p, "proxy for empty UserList should be false")

        p.append(12)

        self.assertEqual(len(L), 1)

        self.failUnless(p, "proxy for non-empty UserList should be true")

        p[:] = [2, 3]

        self.assertEqual(len(L), 2)

        self.assertEqual(len(p), 2)

        self.failUnless(3 in p,

                        "proxy didn't support __contains__() properly")

        p[1] = 5

        self.assertEqual(L[1], 5)

        self.assertEqual(p[1], 5)

        L2 = UserList.UserList(L)

        p2 = weakref.proxy(L2)

        self.assertEqual(p, p2)

        ## self.assertEqual(repr(L2), repr(p2))

        L3 = UserList.UserList(range(10))

        p3 = weakref.proxy(L3)

        self.assertEqual(L3[:], p3[:])

        self.assertEqual(L3[5:], p3[5:])

        self.assertEqual(L3[:5], p3[:5])

        self.assertEqual(L3[2:5], p3[2:5])

    def test_proxy_index(self):

        class C:

            def __index__(self):

                return 10

        o = C()

        p = weakref.proxy(o)

        self.assertEqual(operator.index(p), 10)

    def test_proxy_div(self):

        class C:

            def __floordiv__(self, other):

                return 42

            def __ifloordiv__(self, other):

                return 21

        o = C()

        p = weakref.proxy(o)

        self.assertEqual(p // 5, 42)

        p //= 5

        self.assertEqual(p, 21)

    # The PyWeakref_* C API is documented as allowing either NULL or

    # None as the value for the callback, where either means "no

    # callback".  The "no callback" ref and proxy objects are supposed

    # to be shared so long as they exist by all callers so long as

    # they are active.  In Python 2.3.3 and earlier, this guarantee

    # was not honored, and was broken in different ways for

    # PyWeakref_NewRef() and PyWeakref_NewProxy().  (Two tests.)

    def test_shared_ref_without_callback(self):

        self.check_shared_without_callback(weakref.ref)

    def test_shared_proxy_without_callback(self):

        self.check_shared_without_callback(weakref.proxy)

    def check_shared_without_callback(self, makeref):

        o = Object(1)

        p1 = makeref(o, None)

        p2 = makeref(o, None)

        self.assert_(p1 is p2, "both callbacks were None in the C API")

        del p1, p2

        p1 = makeref(o)

        p2 = makeref(o, None)

        self.assert_(p1 is p2, "callbacks were NULL, None in the C API")

        del p1, p2

        p1 = makeref(o)

        p2 = makeref(o)

        self.assert_(p1 is p2, "both callbacks were NULL in the C API")

        del p1, p2

        p1 = makeref(o, None)

        p2 = makeref(o)

        self.assert_(p1 is p2, "callbacks were None, NULL in the C API")

    def test_callable_proxy(self):

        o = Callable()

        ref1 = weakref.proxy(o)

        self.check_proxy(o, ref1)

        self.assert_(type(ref1) is weakref.CallableProxyType,

                     "proxy is not of callable type")

        ref1('twinkies!')

        self.assert_(o.bar == 'twinkies!',

                     "call through proxy not passed through to original")

        ref1(x='Splat.')

        self.assert_(o.bar == 'Splat.',

                     "call through proxy not passed through to original")

        # expect due to too few args

        self.assertRaises(TypeError, ref1)

        # expect due to too many args

        self.assertRaises(TypeError, ref1, 1, 2, 3)

    def check_proxy(self, o, proxy):

        o.foo = 1

        self.assert_(proxy.foo == 1,

                     "proxy does not reflect attribute addition")

        o.foo = 2

        self.assert_(proxy.foo == 2,

                     "proxy does not reflect attribute modification")

        del o.foo

        self.assert_(not hasattr(proxy, 'foo'),

                     "proxy does not reflect attribute removal")

        proxy.foo = 1

        self.assert_(o.foo == 1,

                     "object does not reflect attribute addition via proxy")

        proxy.foo = 2

        self.assert_(

            o.foo == 2,

            "object does not reflect attribute modification via proxy")

        del proxy.foo

        self.assert_(not hasattr(o, 'foo'),

                     "object does not reflect attribute removal via proxy")

    def test_proxy_deletion(self):

        # Test clearing of SF bug #762891

        class Foo:

            result = None

            def __delitem__(self, accessor):

                self.result = accessor

        g = Foo()

        f = weakref.proxy(g)

        del f[0]

        self.assertEqual(f.result, 0)

    def test_proxy_bool(self):

        # Test clearing of SF bug #1170766

        class List(list): pass

        lyst = List()

        self.assertEqual(bool(weakref.proxy(lyst)), bool(lyst))

    def test_getweakrefcount(self):

        o = C()

        ref1 = weakref.ref(o)

        ref2 = weakref.ref(o, self.callback)

        self.assert_(weakref.getweakrefcount(o) == 2,

                     "got wrong number of weak reference objects")

        proxy1 = weakref.proxy(o)

        proxy2 = weakref.proxy(o, self.callback)

        self.assert_(weakref.getweakrefcount(o) == 4,

                     "got wrong number of weak reference objects")

        del ref1, ref2, proxy1, proxy2

        self.assert_(weakref.getweakrefcount(o) == 0,

                     "weak reference objects not unlinked from"

                     " referent when discarded.")

        # assumes ints do not support weakrefs

        self.assert_(weakref.getweakrefcount(1) == 0,

                     "got wrong number of weak reference objects for int")

    def test_getweakrefs(self):

        o = C()

        ref1 = weakref.ref(o, self.callback)

        ref2 = weakref.ref(o, self.callback)

        del ref1

        self.assert_(weakref.getweakrefs(o) == [ref2],

                     "list of refs does not match")

        o = C()

        ref1 = weakref.ref(o, self.callback)

        ref2 = weakref.ref(o, self.callback)

        del ref2

        self.assert_(weakref.getweakrefs(o) == [ref1],

                     "list of refs does not match")

        del ref1

        self.assert_(weakref.getweakrefs(o) == [],

                     "list of refs not cleared")

        # assumes ints do not support weakrefs

        self.assert_(weakref.getweakrefs(1) == [],

                     "list of refs does not match for int")

    def test_newstyle_number_ops(self):

        class F(float):

            pass

        f = F(2.0)

        p = weakref.proxy(f)

        self.assert_(p + 1.0 == 3.0)

        self.assert_(1.0 + p == 3.0)  # this used to SEGV

    def test_callbacks_protected(self):

        # Callbacks protected from already-set exceptions?

        # Regression test for SF bug #478534.

        class BogusError(Exception):

            pass

        data = {}

        def remove(k):

            del data[k]

        def encapsulate():

            f = lambda : ()

            data[weakref.ref(f, remove)] = None

            raise BogusError

        try:

            encapsulate()

        except BogusError:

            pass

        else:

            self.fail("exception not properly restored")

        try:

            encapsulate()

        except BogusError:

            pass

        else:

            self.fail("exception not properly restored")

    def test_sf_bug_840829(self):

        # "weakref callbacks and gc corrupt memory"

        # subtype_dealloc erroneously exposed a new-style instance

        # already in the process of getting deallocated to gc,

        # causing double-deallocation if the instance had a weakref

        # callback that triggered gc.

        # If the bug exists, there probably won't be an obvious symptom

        # in a release build.  In a debug build, a segfault will occur

        # when the second attempt to remove the instance from the "list

        # of all objects" occurs.

        import gc

        class C(object):

            pass

        c = C()

        wr = weakref.ref(c, lambda ignore: gc.collect())

        del c

        # There endeth the first part.  It gets worse.

        del wr

        c1 = C()

        c1.i = C()

        wr = weakref.ref(c1.i, lambda ignore: gc.collect())

        c2 = C()

        c2.c1 = c1

        del c1  # still alive because c2 points to it

        # Now when subtype_dealloc gets called on c2, it's not enough just

        # that c2 is immune from gc while the weakref callbacks associated

        # with c2 execute (there are none in this 2nd half of the test, btw).

        # subtype_dealloc goes on to call the base classes' deallocs too,

        # so any gc triggered by weakref callbacks associated with anything

        # torn down by a base class dealloc can also trigger double

        # deallocation of c2.

        del c2

    def test_callback_in_cycle_1(self):

        import gc

        class J(object):

            pass

        class II(object):

            def acallback(self, ignore):

                self.J

        I = II()

        I.J = J

        I.wr = weakref.ref(J, I.acallback)

        # Now J and II are each in a self-cycle (as all new-style class

        # objects are, since their __mro__ points back to them).  I holds

        # both a weak reference (I.wr) and a strong reference (I.J) to class

        # J.  I is also in a cycle (I.wr points to a weakref that references

        # I.acallback).  When we del these three, they all become trash, but

        # the cycles prevent any of them from getting cleaned up immediately.

        # Instead they have to wait for cyclic gc to deduce that they're

        # trash.

        #

        # gc used to call tp_clear on all of them, and the order in which

        # it does that is pretty accidental.  The exact order in which we

        # built up these things manages to provoke gc into running tp_clear

        # in just the right order (I last).  Calling tp_clear on II leaves

        # behind an insane class object (its __mro__ becomes NULL).  Calling

        # tp_clear on J breaks its self-cycle, but J doesn't get deleted

        # just then because of the strong reference from I.J.  Calling

        # tp_clear on I starts to clear I's __dict__, and just happens to

        # clear I.J first -- I.wr is still intact.  That removes the last

        # reference to J, which triggers the weakref callback.  The callback

        # tries to do "self.J", and instances of new-style classes look up

        # attributes ("J") in the class dict first.  The class (II) wants to

        # search II.__mro__, but that's NULL.   The result was a segfault in

        # a release build, and an assert failure in a debug build.

        del I, J, II

        gc.collect()

    def test_callback_in_cycle_2(self):

        import gc

        # This is just like test_callback_in_cycle_1, except that II is an

        # old-style class.  The symptom is different then:  an instance of an

        # old-style class looks in its own __dict__ first.  'J' happens to

        # get cleared from I.__dict__ before 'wr', and 'J' was never in II's

        # __dict__, so the attribute isn't found.  The difference is that

        # the old-style II doesn't have a NULL __mro__ (it doesn't have any

        # __mro__), so no segfault occurs.  Instead it got:

        #    test_callback_in_cycle_2 (__main__.ReferencesTestCase) ...

        #    Exception exceptions.AttributeError:

        #   "II instance has no attribute 'J'" in <bound method II.acallback

        #       of <?.II instance at 0x00B9B4B8>> ignored

        class J(object):

            pass

        class II:

            def acallback(self, ignore):

                self.J

        I = II()

        I.J = J

        I.wr = weakref.ref(J, I.acallback)

        del I, J, II

        gc.collect()

    def test_callback_in_cycle_3(self):

        import gc

        # This one broke the first patch that fixed the last two.  In this

        # case, the objects reachable from the callback aren't also reachable

        # from the object (c1) *triggering* the callback:  you can get to

        # c1 from c2, but not vice-versa.  The result was that c2's __dict__

        # got tp_clear'ed by the time the c2.cb callback got invoked.

        class C:

            def cb(self, ignore):

                self.me

                self.c1

                self.wr

        c1, c2 = C(), C()

        c2.me = c2

        c2.c1 = c1

        c2.wr = weakref.ref(c1, c2.cb)

        del c1, c2

        gc.collect()

    def test_callback_in_cycle_4(self):

        import gc

        # Like test_callback_in_cycle_3, except c2 and c1 have different

        # classes.  c2's class (C) isn't reachable from c1 then, so protecting

        # objects reachable from the dying object (c1) isn't enough to stop

        # c2's class (C) from getting tp_clear'ed before c2.cb is invoked.

        # The result was a segfault (C.__mro__ was NULL when the callback

        # tried to look up self.me).

        class C(object):

            def cb(self, ignore):

                self.me

                self.c1

                self.wr