8 from test
import test_support
10 # Used in ReferencesTestCase.test_ref_created_during_del() .
21 def __call__(self
, x
):
25 def create_function():
29 def create_bound_method():
32 def create_unbound_method():
36 class TestBase(unittest
.TestCase
):
41 def callback(self
, ref
):
45 class ReferencesTestCase(TestBase
):
47 def test_basic_ref(self
):
48 self
.check_basic_ref(C
)
49 self
.check_basic_ref(create_function
)
50 self
.check_basic_ref(create_bound_method
)
51 self
.check_basic_ref(create_unbound_method
)
53 # Just make sure the tp_repr handler doesn't raise an exception.
62 def test_basic_callback(self
):
63 self
.check_basic_callback(C
)
64 self
.check_basic_callback(create_function
)
65 self
.check_basic_callback(create_bound_method
)
66 self
.check_basic_callback(create_unbound_method
)
68 def test_multiple_callbacks(self
):
70 ref1
= weakref
.ref(o
, self
.callback
)
71 ref2
= weakref
.ref(o
, self
.callback
)
73 self
.assertTrue(ref1() is None,
74 "expected reference to be invalidated")
75 self
.assertTrue(ref2() is None,
76 "expected reference to be invalidated")
77 self
.assertTrue(self
.cbcalled
== 2,
78 "callback not called the right number of times")
80 def test_multiple_selfref_callbacks(self
):
81 # Make sure all references are invalidated before callbacks are called
83 # What's important here is that we're using the first
84 # reference in the callback invoked on the second reference
85 # (the most recently created ref is cleaned up first). This
86 # tests that all references to the object are invalidated
87 # before any of the callbacks are invoked, so that we only
88 # have one invocation of _weakref.c:cleanup_helper() active
89 # for a particular object at a time.
91 def callback(object, self
=self
):
94 self
.ref
= weakref
.ref(c
, callback
)
95 ref1
= weakref
.ref(c
, callback
)
98 def test_proxy_ref(self
):
101 ref1
= weakref
.proxy(o
, self
.callback
)
102 ref2
= weakref
.proxy(o
, self
.callback
)
108 self
.assertRaises(weakref
.ReferenceError, check
, ref1
)
109 self
.assertRaises(weakref
.ReferenceError, check
, ref2
)
110 self
.assertRaises(weakref
.ReferenceError, bool, weakref
.proxy(C()))
111 self
.assertTrue(self
.cbcalled
== 2)
113 def check_basic_ref(self
, factory
):
116 self
.assertTrue(ref() is not None,
117 "weak reference to live object should be live")
119 self
.assertTrue(o
is o2
,
120 "<ref>() should return original object if live")
122 def check_basic_callback(self
, factory
):
125 ref
= weakref
.ref(o
, self
.callback
)
127 self
.assertTrue(self
.cbcalled
== 1,
128 "callback did not properly set 'cbcalled'")
129 self
.assertTrue(ref() is None,
130 "ref2 should be dead after deleting object reference")
132 def test_ref_reuse(self
):
134 ref1
= weakref
.ref(o
)
135 # create a proxy to make sure that there's an intervening creation
136 # between these two; it should make no difference
137 proxy
= weakref
.proxy(o
)
138 ref2
= weakref
.ref(o
)
139 self
.assertTrue(ref1
is ref2
,
140 "reference object w/out callback should be re-used")
143 proxy
= weakref
.proxy(o
)
144 ref1
= weakref
.ref(o
)
145 ref2
= weakref
.ref(o
)
146 self
.assertTrue(ref1
is ref2
,
147 "reference object w/out callback should be re-used")
148 self
.assertTrue(weakref
.getweakrefcount(o
) == 2,
149 "wrong weak ref count for object")
151 self
.assertTrue(weakref
.getweakrefcount(o
) == 1,
152 "wrong weak ref count for object after deleting proxy")
154 def test_proxy_reuse(self
):
156 proxy1
= weakref
.proxy(o
)
158 proxy2
= weakref
.proxy(o
)
159 self
.assertTrue(proxy1
is proxy2
,
160 "proxy object w/out callback should have been re-used")
162 def test_basic_proxy(self
):
164 self
.check_proxy(o
, weakref
.proxy(o
))
166 L
= UserList
.UserList()
168 self
.assertFalse(p
, "proxy for empty UserList should be false")
170 self
.assertEqual(len(L
), 1)
171 self
.assertTrue(p
, "proxy for non-empty UserList should be true")
173 self
.assertEqual(len(L
), 2)
174 self
.assertEqual(len(p
), 2)
175 self
.assertTrue(3 in p
,
176 "proxy didn't support __contains__() properly")
178 self
.assertEqual(L
[1], 5)
179 self
.assertEqual(p
[1], 5)
180 L2
= UserList
.UserList(L
)
181 p2
= weakref
.proxy(L2
)
182 self
.assertEqual(p
, p2
)
183 ## self.assertEqual(repr(L2), repr(p2))
184 L3
= UserList
.UserList(range(10))
185 p3
= weakref
.proxy(L3
)
186 self
.assertEqual(L3
[:], p3
[:])
187 self
.assertEqual(L3
[5:], p3
[5:])
188 self
.assertEqual(L3
[:5], p3
[:5])
189 self
.assertEqual(L3
[2:5], p3
[2:5])
191 def test_proxy_unicode(self
):
196 def __unicode__(self
):
199 self
.assertTrue("__unicode__" in dir(weakref
.proxy(instance
)))
200 self
.assertEqual(unicode(weakref
.proxy(instance
)), u
"unicode")
202 def test_proxy_index(self
):
208 self
.assertEqual(operator
.index(p
), 10)
210 def test_proxy_div(self
):
212 def __floordiv__(self
, other
):
214 def __ifloordiv__(self
, other
):
218 self
.assertEqual(p
// 5, 42)
220 self
.assertEqual(p
, 21)
222 # The PyWeakref_* C API is documented as allowing either NULL or
223 # None as the value for the callback, where either means "no
224 # callback". The "no callback" ref and proxy objects are supposed
225 # to be shared so long as they exist by all callers so long as
226 # they are active. In Python 2.3.3 and earlier, this guarantee
227 # was not honored, and was broken in different ways for
228 # PyWeakref_NewRef() and PyWeakref_NewProxy(). (Two tests.)
230 def test_shared_ref_without_callback(self
):
231 self
.check_shared_without_callback(weakref
.ref
)
233 def test_shared_proxy_without_callback(self
):
234 self
.check_shared_without_callback(weakref
.proxy
)
236 def check_shared_without_callback(self
, makeref
):
238 p1
= makeref(o
, None)
239 p2
= makeref(o
, None)
240 self
.assertTrue(p1
is p2
, "both callbacks were None in the C API")
243 p2
= makeref(o
, None)
244 self
.assertTrue(p1
is p2
, "callbacks were NULL, None in the C API")
248 self
.assertTrue(p1
is p2
, "both callbacks were NULL in the C API")
250 p1
= makeref(o
, None)
252 self
.assertTrue(p1
is p2
, "callbacks were None, NULL in the C API")
254 def test_callable_proxy(self
):
256 ref1
= weakref
.proxy(o
)
258 self
.check_proxy(o
, ref1
)
260 self
.assertTrue(type(ref1
) is weakref
.CallableProxyType
,
261 "proxy is not of callable type")
263 self
.assertTrue(o
.bar
== 'twinkies!',
264 "call through proxy not passed through to original")
266 self
.assertTrue(o
.bar
== 'Splat.',
267 "call through proxy not passed through to original")
269 # expect due to too few args
270 self
.assertRaises(TypeError, ref1
)
272 # expect due to too many args
273 self
.assertRaises(TypeError, ref1
, 1, 2, 3)
275 def check_proxy(self
, o
, proxy
):
277 self
.assertTrue(proxy
.foo
== 1,
278 "proxy does not reflect attribute addition")
280 self
.assertTrue(proxy
.foo
== 2,
281 "proxy does not reflect attribute modification")
283 self
.assertTrue(not hasattr(proxy
, 'foo'),
284 "proxy does not reflect attribute removal")
287 self
.assertTrue(o
.foo
== 1,
288 "object does not reflect attribute addition via proxy")
292 "object does not reflect attribute modification via proxy")
294 self
.assertTrue(not hasattr(o
, 'foo'),
295 "object does not reflect attribute removal via proxy")
297 def test_proxy_deletion(self
):
298 # Test clearing of SF bug #762891
301 def __delitem__(self
, accessor
):
302 self
.result
= accessor
306 self
.assertEqual(f
.result
, 0)
308 def test_proxy_bool(self
):
309 # Test clearing of SF bug #1170766
310 class List(list): pass
312 self
.assertEqual(bool(weakref
.proxy(lyst
)), bool(lyst
))
314 def test_getweakrefcount(self
):
316 ref1
= weakref
.ref(o
)
317 ref2
= weakref
.ref(o
, self
.callback
)
318 self
.assertTrue(weakref
.getweakrefcount(o
) == 2,
319 "got wrong number of weak reference objects")
321 proxy1
= weakref
.proxy(o
)
322 proxy2
= weakref
.proxy(o
, self
.callback
)
323 self
.assertTrue(weakref
.getweakrefcount(o
) == 4,
324 "got wrong number of weak reference objects")
326 del ref1
, ref2
, proxy1
, proxy2
327 self
.assertTrue(weakref
.getweakrefcount(o
) == 0,
328 "weak reference objects not unlinked from"
329 " referent when discarded.")
331 # assumes ints do not support weakrefs
332 self
.assertTrue(weakref
.getweakrefcount(1) == 0,
333 "got wrong number of weak reference objects for int")
335 def test_getweakrefs(self
):
337 ref1
= weakref
.ref(o
, self
.callback
)
338 ref2
= weakref
.ref(o
, self
.callback
)
340 self
.assertTrue(weakref
.getweakrefs(o
) == [ref2
],
341 "list of refs does not match")
344 ref1
= weakref
.ref(o
, self
.callback
)
345 ref2
= weakref
.ref(o
, self
.callback
)
347 self
.assertTrue(weakref
.getweakrefs(o
) == [ref1
],
348 "list of refs does not match")
351 self
.assertTrue(weakref
.getweakrefs(o
) == [],
352 "list of refs not cleared")
354 # assumes ints do not support weakrefs
355 self
.assertTrue(weakref
.getweakrefs(1) == [],
356 "list of refs does not match for int")
358 def test_newstyle_number_ops(self
):
363 self
.assertTrue(p
+ 1.0 == 3.0)
364 self
.assertTrue(1.0 + p
== 3.0) # this used to SEGV
366 def test_callbacks_protected(self
):
367 # Callbacks protected from already-set exceptions?
368 # Regression test for SF bug #478534.
369 class BogusError(Exception):
376 data
[weakref
.ref(f
, remove
)] = None
383 self
.fail("exception not properly restored")
389 self
.fail("exception not properly restored")
391 def test_sf_bug_840829(self
):
392 # "weakref callbacks and gc corrupt memory"
393 # subtype_dealloc erroneously exposed a new-style instance
394 # already in the process of getting deallocated to gc,
395 # causing double-deallocation if the instance had a weakref
396 # callback that triggered gc.
397 # If the bug exists, there probably won't be an obvious symptom
398 # in a release build. In a debug build, a segfault will occur
399 # when the second attempt to remove the instance from the "list
400 # of all objects" occurs.
408 wr
= weakref
.ref(c
, lambda ignore
: gc
.collect())
411 # There endeth the first part. It gets worse.
416 wr
= weakref
.ref(c1
.i
, lambda ignore
: gc
.collect())
420 del c1
# still alive because c2 points to it
422 # Now when subtype_dealloc gets called on c2, it's not enough just
423 # that c2 is immune from gc while the weakref callbacks associated
424 # with c2 execute (there are none in this 2nd half of the test, btw).
425 # subtype_dealloc goes on to call the base classes' deallocs too,
426 # so any gc triggered by weakref callbacks associated with anything
427 # torn down by a base class dealloc can also trigger double
428 # deallocation of c2.
431 def test_callback_in_cycle_1(self
):
438 def acallback(self
, ignore
):
443 I
.wr
= weakref
.ref(J
, I
.acallback
)
445 # Now J and II are each in a self-cycle (as all new-style class
446 # objects are, since their __mro__ points back to them). I holds
447 # both a weak reference (I.wr) and a strong reference (I.J) to class
448 # J. I is also in a cycle (I.wr points to a weakref that references
449 # I.acallback). When we del these three, they all become trash, but
450 # the cycles prevent any of them from getting cleaned up immediately.
451 # Instead they have to wait for cyclic gc to deduce that they're
454 # gc used to call tp_clear on all of them, and the order in which
455 # it does that is pretty accidental. The exact order in which we
456 # built up these things manages to provoke gc into running tp_clear
457 # in just the right order (I last). Calling tp_clear on II leaves
458 # behind an insane class object (its __mro__ becomes NULL). Calling
459 # tp_clear on J breaks its self-cycle, but J doesn't get deleted
460 # just then because of the strong reference from I.J. Calling
461 # tp_clear on I starts to clear I's __dict__, and just happens to
462 # clear I.J first -- I.wr is still intact. That removes the last
463 # reference to J, which triggers the weakref callback. The callback
464 # tries to do "self.J", and instances of new-style classes look up
465 # attributes ("J") in the class dict first. The class (II) wants to
466 # search II.__mro__, but that's NULL. The result was a segfault in
467 # a release build, and an assert failure in a debug build.
471 def test_callback_in_cycle_2(self
):
474 # This is just like test_callback_in_cycle_1, except that II is an
475 # old-style class. The symptom is different then: an instance of an
476 # old-style class looks in its own __dict__ first. 'J' happens to
477 # get cleared from I.__dict__ before 'wr', and 'J' was never in II's
478 # __dict__, so the attribute isn't found. The difference is that
479 # the old-style II doesn't have a NULL __mro__ (it doesn't have any
480 # __mro__), so no segfault occurs. Instead it got:
481 # test_callback_in_cycle_2 (__main__.ReferencesTestCase) ...
482 # Exception exceptions.AttributeError:
483 # "II instance has no attribute 'J'" in <bound method II.acallback
484 # of <?.II instance at 0x00B9B4B8>> ignored
490 def acallback(self
, ignore
):
495 I
.wr
= weakref
.ref(J
, I
.acallback
)
500 def test_callback_in_cycle_3(self
):
503 # This one broke the first patch that fixed the last two. In this
504 # case, the objects reachable from the callback aren't also reachable
505 # from the object (c1) *triggering* the callback: you can get to
506 # c1 from c2, but not vice-versa. The result was that c2's __dict__
507 # got tp_clear'ed by the time the c2.cb callback got invoked.
510 def cb(self
, ignore
):
519 c2
.wr
= weakref
.ref(c1
, c2
.cb
)
524 def test_callback_in_cycle_4(self
):
527 # Like test_callback_in_cycle_3, except c2 and c1 have different
528 # classes. c2's class (C) isn't reachable from c1 then, so protecting
529 # objects reachable from the dying object (c1) isn't enough to stop
530 # c2's class (C) from getting tp_clear'ed before c2.cb is invoked.
531 # The result was a segfault (C.__mro__ was NULL when the callback
532 # tried to look up self.me).
535 def cb(self
, ignore
):
547 c2
.wr
= weakref
.ref(c1
, c2
.cb
)
552 def test_callback_in_cycle_resurrection(self
):
555 # Do something nasty in a weakref callback: resurrect objects
556 # from dead cycles. For this to be attempted, the weakref and
557 # its callback must also be part of the cyclic trash (else the
558 # objects reachable via the callback couldn't be in cyclic trash
559 # to begin with -- the callback would act like an external root).
560 # But gc clears trash weakrefs with callbacks early now, which
561 # disables the callbacks, so the callbacks shouldn't get called
562 # at all (and so nothing actually gets resurrected).
566 def __init__(self
, value
):
567 self
.attribute
= value
569 def acallback(self
, ignore
):
575 c1
.wr
= weakref
.ref(c2
, c1
.acallback
)
576 c2
.wr
= weakref
.ref(c1
, c2
.acallback
)
578 def C_went_away(ignore
):
579 alist
.append("C went away")
580 wr
= weakref
.ref(C
, C_went_away
)
582 del c1
, c2
, C
# make them all trash
583 self
.assertEqual(alist
, []) # del isn't enough to reclaim anything
586 # c1.wr and c2.wr were part of the cyclic trash, so should have
587 # been cleared without their callbacks executing. OTOH, the weakref
588 # to C is bound to a function local (wr), and wasn't trash, so that
589 # callback should have been invoked when C went away.
590 self
.assertEqual(alist
, ["C went away"])
591 # The remaining weakref should be dead now (its callback ran).
592 self
.assertEqual(wr(), None)
596 self
.assertEqual(alist
, [])
598 def test_callbacks_on_callback(self
):
601 # Set up weakref callbacks *on* weakref callbacks.
603 def safe_callback(ignore
):
604 alist
.append("safe_callback called")
607 def cb(self
, ignore
):
608 alist
.append("cb called")
614 c
.wr
= weakref
.ref(d
, callback
) # this won't trigger
615 d
.wr
= weakref
.ref(callback
, d
.cb
) # ditto
616 external_wr
= weakref
.ref(callback
, safe_callback
) # but this will
617 self
.assertTrue(external_wr() is callback
)
619 # The weakrefs attached to c and d should get cleared, so that
620 # C.cb is never called. But external_wr isn't part of the cyclic
621 # trash, and no cyclic trash is reachable from it, so safe_callback
622 # should get invoked when the bound method object callback (c.cb)
623 # -- which is itself a callback, and also part of the cyclic trash --
624 # gets reclaimed at the end of gc.
626 del callback
, c
, d
, C
627 self
.assertEqual(alist
, []) # del isn't enough to clean up cycles
629 self
.assertEqual(alist
, ["safe_callback called"])
630 self
.assertEqual(external_wr(), None)
634 self
.assertEqual(alist
, [])
636 def test_gc_during_ref_creation(self
):
637 self
.check_gc_during_creation(weakref
.ref
)
639 def test_gc_during_proxy_creation(self
):
640 self
.check_gc_during_creation(weakref
.proxy
)
642 def check_gc_during_creation(self
, makeref
):
643 thresholds
= gc
.get_threshold()
644 gc
.set_threshold(1, 1, 1)
656 a
.wr
= makeref(referenced
)
659 # now make sure the object and the ref get labeled as
662 weakref
.ref(referenced
, callback
)
665 gc
.set_threshold(*thresholds
)
667 def test_ref_created_during_del(self
):
669 # A weakref created in an object's __del__() would crash the
670 # interpreter when the weakref was cleaned up since it would refer to
671 # non-existent memory. This test should not segfault the interpreter.
672 class Target(object):
675 ref_from_del
= weakref
.ref(self
)
681 # <weakref to class>.__init__() doesn't check errors correctly
682 r
= weakref
.ref(Exception)
683 self
.assertRaises(TypeError, r
.__init
__, 0, 0, 0, 0, 0)
684 # No exception should be raised here
688 class SubclassableWeakrefTestCase(TestBase
):
690 def test_subclass_refs(self
):
691 class MyRef(weakref
.ref
):
692 def __init__(self
, ob
, callback
=None, value
=42):
694 super(MyRef
, self
).__init
__(ob
, callback
)
697 return super(MyRef
, self
).__call
__()
699 mr
= MyRef(o
, value
=24)
700 self
.assertTrue(mr() is o
)
701 self
.assertTrue(mr
.called
)
702 self
.assertEqual(mr
.value
, 24)
704 self
.assertTrue(mr() is None)
705 self
.assertTrue(mr
.called
)
707 def test_subclass_refs_dont_replace_standard_refs(self
):
708 class MyRef(weakref
.ref
):
713 self
.assertTrue(r1
is not r2
)
714 self
.assertEqual(weakref
.getweakrefs(o
), [r2
, r1
])
715 self
.assertEqual(weakref
.getweakrefcount(o
), 2)
717 self
.assertEqual(weakref
.getweakrefcount(o
), 3)
718 refs
= weakref
.getweakrefs(o
)
719 self
.assertEqual(len(refs
), 3)
720 self
.assertTrue(r2
is refs
[0])
721 self
.assertTrue(r1
in refs
[1:])
722 self
.assertTrue(r3
in refs
[1:])
724 def test_subclass_refs_dont_conflate_callbacks(self
):
725 class MyRef(weakref
.ref
):
730 self
.assertTrue(r1
is not r2
)
731 refs
= weakref
.getweakrefs(o
)
732 self
.assertTrue(r1
in refs
)
733 self
.assertTrue(r2
in refs
)
735 def test_subclass_refs_with_slots(self
):
736 class MyRef(weakref
.ref
):
737 __slots__
= "slot1", "slot2"
738 def __new__(type, ob
, callback
, slot1
, slot2
):
739 return weakref
.ref
.__new
__(type, ob
, callback
)
740 def __init__(self
, ob
, callback
, slot1
, slot2
):
744 return self
.slot1
+ self
.slot2
746 r
= MyRef(o
, None, "abc", "def")
747 self
.assertEqual(r
.slot1
, "abc")
748 self
.assertEqual(r
.slot2
, "def")
749 self
.assertEqual(r
.meth(), "abcdef")
750 self
.assertFalse(hasattr(r
, "__dict__"))
752 def test_subclass_refs_with_cycle(self
):
754 # An instance of a weakref subclass can have attributes.
755 # If such a weakref holds the only strong reference to the object,
756 # deleting the weakref will delete the object. In this case,
757 # the callback must not be called, because the ref object is
759 class MyRef(weakref
.ref
):
762 # Use a local callback, for "regrtest -R::"
763 # to detect refcounting problems
768 r1
= MyRef(o
, callback
)
772 del r1
# Used to crash here
774 self
.assertEqual(self
.cbcalled
, 0)
776 # Same test, with two weakrefs to the same object
777 # (since code paths are different)
779 r1
= MyRef(o
, callback
)
780 r2
= MyRef(o
, callback
)
786 del r1
# Used to crash here
788 self
.assertEqual(self
.cbcalled
, 0)
792 def __init__(self
, arg
):
795 return "<Object %r>" % self
.arg
798 class MappingTestCase(TestBase
):
802 def test_weak_values(self
):
804 # This exercises d.copy(), d.items(), d[], del d[], len(d).
806 dict, objects
= self
.make_weak_valued_dict()
808 self
.assertTrue(weakref
.getweakrefcount(o
) == 1,
809 "wrong number of weak references to %r!" % o
)
810 self
.assertTrue(o
is dict[o
.arg
],
811 "wrong object returned by weak dict!")
812 items1
= dict.items()
813 items2
= dict.copy().items()
816 self
.assertTrue(items1
== items2
,
817 "cloning of weak-valued dictionary did not work!")
819 self
.assertTrue(len(dict) == self
.COUNT
)
821 self
.assertTrue(len(dict) == (self
.COUNT
- 1),
822 "deleting object did not cause dictionary update")
824 self
.assertTrue(len(dict) == 0,
825 "deleting the values did not clear the dictionary")
826 # regression on SF bug #447152:
827 dict = weakref
.WeakValueDictionary()
828 self
.assertRaises(KeyError, dict.__getitem
__, 1)
830 self
.assertRaises(KeyError, dict.__getitem
__, 2)
832 def test_weak_keys(self
):
834 # This exercises d.copy(), d.items(), d[] = v, d[], del d[],
835 # len(d), d.has_key().
837 dict, objects
= self
.make_weak_keyed_dict()
839 self
.assertTrue(weakref
.getweakrefcount(o
) == 1,
840 "wrong number of weak references to %r!" % o
)
841 self
.assertTrue(o
.arg
is dict[o
],
842 "wrong object returned by weak dict!")
843 items1
= dict.items()
844 items2
= dict.copy().items()
845 self
.assertTrue(set(items1
) == set(items2
),
846 "cloning of weak-keyed dictionary did not work!")
848 self
.assertTrue(len(dict) == self
.COUNT
)
850 self
.assertTrue(len(dict) == (self
.COUNT
- 1),
851 "deleting object did not cause dictionary update")
853 self
.assertTrue(len(dict) == 0,
854 "deleting the keys did not clear the dictionary")
856 dict[o
] = "What is the meaning of the universe?"
857 self
.assertTrue(dict.has_key(o
))
858 self
.assertTrue(not dict.has_key(34))
860 def test_weak_keyed_iters(self
):
861 dict, objects
= self
.make_weak_keyed_dict()
862 self
.check_iters(dict)
865 refs
= dict.keyrefs()
866 self
.assertEqual(len(refs
), len(objects
))
867 objects2
= list(objects
)
870 self
.assertTrue(dict.has_key(ob
))
871 self
.assertTrue(ob
in dict)
872 self
.assertEqual(ob
.arg
, dict[ob
])
874 self
.assertEqual(len(objects2
), 0)
877 objects2
= list(objects
)
878 self
.assertEqual(len(list(dict.iterkeyrefs())), len(objects
))
879 for wr
in dict.iterkeyrefs():
881 self
.assertTrue(dict.has_key(ob
))
882 self
.assertTrue(ob
in dict)
883 self
.assertEqual(ob
.arg
, dict[ob
])
885 self
.assertEqual(len(objects2
), 0)
887 def test_weak_valued_iters(self
):
888 dict, objects
= self
.make_weak_valued_dict()
889 self
.check_iters(dict)
892 refs
= dict.valuerefs()
893 self
.assertEqual(len(refs
), len(objects
))
894 objects2
= list(objects
)
897 self
.assertEqual(ob
, dict[ob
.arg
])
898 self
.assertEqual(ob
.arg
, dict[ob
.arg
].arg
)
900 self
.assertEqual(len(objects2
), 0)
902 # Test itervaluerefs()
903 objects2
= list(objects
)
904 self
.assertEqual(len(list(dict.itervaluerefs())), len(objects
))
905 for wr
in dict.itervaluerefs():
907 self
.assertEqual(ob
, dict[ob
.arg
])
908 self
.assertEqual(ob
.arg
, dict[ob
.arg
].arg
)
910 self
.assertEqual(len(objects2
), 0)
912 def check_iters(self
, dict):
915 for item
in dict.iteritems():
917 self
.assertTrue(len(items
) == 0, "iteritems() did not touch all items")
919 # key iterator, via __iter__():
923 self
.assertTrue(len(keys
) == 0, "__iter__() did not touch all keys")
925 # key iterator, via iterkeys():
927 for k
in dict.iterkeys():
929 self
.assertTrue(len(keys
) == 0, "iterkeys() did not touch all keys")
932 values
= dict.values()
933 for v
in dict.itervalues():
935 self
.assertTrue(len(values
) == 0,
936 "itervalues() did not touch all values")
938 def test_make_weak_keyed_dict_from_dict(self
):
940 dict = weakref
.WeakKeyDictionary({o
:364})
941 self
.assertTrue(dict[o
] == 364)
943 def test_make_weak_keyed_dict_from_weak_keyed_dict(self
):
945 dict = weakref
.WeakKeyDictionary({o
:364})
946 dict2
= weakref
.WeakKeyDictionary(dict)
947 self
.assertTrue(dict[o
] == 364)
949 def make_weak_keyed_dict(self
):
950 dict = weakref
.WeakKeyDictionary()
951 objects
= map(Object
, range(self
.COUNT
))
956 def make_weak_valued_dict(self
):
957 dict = weakref
.WeakValueDictionary()
958 objects
= map(Object
, range(self
.COUNT
))
963 def check_popitem(self
, klass
, key1
, value1
, key2
, value2
):
965 weakdict
[key1
] = value1
966 weakdict
[key2
] = value2
967 self
.assertTrue(len(weakdict
) == 2)
968 k
, v
= weakdict
.popitem()
969 self
.assertTrue(len(weakdict
) == 1)
971 self
.assertTrue(v
is value1
)
973 self
.assertTrue(v
is value2
)
974 k
, v
= weakdict
.popitem()
975 self
.assertTrue(len(weakdict
) == 0)
977 self
.assertTrue(v
is value1
)
979 self
.assertTrue(v
is value2
)
981 def test_weak_valued_dict_popitem(self
):
982 self
.check_popitem(weakref
.WeakValueDictionary
,
983 "key1", C(), "key2", C())
985 def test_weak_keyed_dict_popitem(self
):
986 self
.check_popitem(weakref
.WeakKeyDictionary
,
987 C(), "value 1", C(), "value 2")
989 def check_setdefault(self
, klass
, key
, value1
, value2
):
990 self
.assertTrue(value1
is not value2
,
992 " -- value parameters must be distinct objects")
994 o
= weakdict
.setdefault(key
, value1
)
995 self
.assertTrue(o
is value1
)
996 self
.assertTrue(weakdict
.has_key(key
))
997 self
.assertTrue(weakdict
.get(key
) is value1
)
998 self
.assertTrue(weakdict
[key
] is value1
)
1000 o
= weakdict
.setdefault(key
, value2
)
1001 self
.assertTrue(o
is value1
)
1002 self
.assertTrue(weakdict
.has_key(key
))
1003 self
.assertTrue(weakdict
.get(key
) is value1
)
1004 self
.assertTrue(weakdict
[key
] is value1
)
1006 def test_weak_valued_dict_setdefault(self
):
1007 self
.check_setdefault(weakref
.WeakValueDictionary
,
1010 def test_weak_keyed_dict_setdefault(self
):
1011 self
.check_setdefault(weakref
.WeakKeyDictionary
,
1012 C(), "value 1", "value 2")
1014 def check_update(self
, klass
, dict):
1016 # This exercises d.update(), len(d), d.keys(), d.has_key(),
1020 weakdict
.update(dict)
1021 self
.assertTrue(len(weakdict
) == len(dict))
1022 for k
in weakdict
.keys():
1023 self
.assertTrue(dict.has_key(k
),
1024 "mysterious new key appeared in weak dict")
1026 self
.assertTrue(v
is weakdict
[k
])
1027 self
.assertTrue(v
is weakdict
.get(k
))
1028 for k
in dict.keys():
1029 self
.assertTrue(weakdict
.has_key(k
),
1030 "original key disappeared in weak dict")
1032 self
.assertTrue(v
is weakdict
[k
])
1033 self
.assertTrue(v
is weakdict
.get(k
))
1035 def test_weak_valued_dict_update(self
):
1036 self
.check_update(weakref
.WeakValueDictionary
,
1037 {1: C(), 'a': C(), C(): C()})
1039 def test_weak_keyed_dict_update(self
):
1040 self
.check_update(weakref
.WeakKeyDictionary
,
1041 {C(): 1, C(): 2, C(): 3})
1043 def test_weak_keyed_delitem(self
):
1044 d
= weakref
.WeakKeyDictionary()
1049 self
.assertTrue(len(d
) == 2)
1051 self
.assertTrue(len(d
) == 1)
1052 self
.assertTrue(d
.keys() == [o2
])
1054 def test_weak_valued_delitem(self
):
1055 d
= weakref
.WeakValueDictionary()
1059 d
['something else'] = o2
1060 self
.assertTrue(len(d
) == 2)
1062 self
.assertTrue(len(d
) == 1)
1063 self
.assertTrue(d
.items() == [('something else', o2
)])
1065 def test_weak_keyed_bad_delitem(self
):
1066 d
= weakref
.WeakKeyDictionary()
1068 # An attempt to delete an object that isn't there should raise
1069 # KeyError. It didn't before 2.3.
1070 self
.assertRaises(KeyError, d
.__delitem
__, o
)
1071 self
.assertRaises(KeyError, d
.__getitem
__, o
)
1073 # If a key isn't of a weakly referencable type, __getitem__ and
1074 # __setitem__ raise TypeError. __delitem__ should too.
1075 self
.assertRaises(TypeError, d
.__delitem
__, 13)
1076 self
.assertRaises(TypeError, d
.__getitem
__, 13)
1077 self
.assertRaises(TypeError, d
.__setitem
__, 13, 13)
1079 def test_weak_keyed_cascading_deletes(self
):
1080 # SF bug 742860. For some reason, before 2.3 __delitem__ iterated
1081 # over the keys via self.data.iterkeys(). If things vanished from
1082 # the dict during this (or got added), that caused a RuntimeError.
1084 d
= weakref
.WeakKeyDictionary()
1088 def __init__(self
, i
):
1091 return hash(self
.value
)
1092 def __eq__(self
, other
):
1094 # Side effect that mutates the dict, by removing the
1095 # last strong reference to a key.
1097 return self
.value
== other
.value
1099 objs
= [C(i
) for i
in range(4)]
1102 del o
# now the only strong references to keys are in objs
1103 # Find the order in which iterkeys sees the keys.
1105 # Reverse it, so that the iteration implementation of __delitem__
1106 # has to keep looping to find the first object we delete.
1109 # Turn on mutation in C.__eq__. The first time thru the loop,
1110 # under the iterkeys() business the first comparison will delete
1111 # the last item iterkeys() would see, and that causes a
1112 # RuntimeError: dictionary changed size during iteration
1113 # when the iterkeys() loop goes around to try comparing the next
1114 # key. After this was fixed, it just deletes the last object *our*
1115 # "for o in obj" loop would have gotten to.
1121 self
.assertEqual(len(d
), 0)
1122 self
.assertEqual(count
, 2)
1124 from test
import mapping_tests
1126 class WeakValueDictionaryTestCase(mapping_tests
.BasicTestMappingProtocol
):
1127 """Check that WeakValueDictionary conforms to the mapping protocol"""
1128 __ref
= {"key1":Object(1), "key2":Object(2), "key3":Object(3)}
1129 type2test
= weakref
.WeakValueDictionary
1130 def _reference(self
):
1131 return self
.__ref
.copy()
1133 class WeakKeyDictionaryTestCase(mapping_tests
.BasicTestMappingProtocol
):
1134 """Check that WeakKeyDictionary conforms to the mapping protocol"""
1135 __ref
= {Object("key1"):1, Object("key2"):2, Object("key3"):3}
1136 type2test
= weakref
.WeakKeyDictionary
1137 def _reference(self
):
1138 return self
.__ref
.copy()
1140 libreftest
= """ Doctest for examples in the library reference: weakref.rst
1143 >>> class Dict(dict):
1146 >>> obj = Dict(red=1, green=2, blue=3) # this object is weak referencable
1147 >>> r = weakref.ref(obj)
1148 >>> print r() is obj
1156 >>> r = weakref.ref(o)
1165 >>> class ExtendedRef(weakref.ref):
1166 ... def __init__(self, ob, callback=None, **annotations):
1167 ... super(ExtendedRef, self).__init__(ob, callback)
1168 ... self.__counter = 0
1169 ... for k, v in annotations.iteritems():
1170 ... setattr(self, k, v)
1171 ... def __call__(self):
1172 ... '''Return a pair containing the referent and the number of
1173 ... times the reference has been called.
1175 ... ob = super(ExtendedRef, self).__call__()
1176 ... if ob is not None:
1177 ... self.__counter += 1
1178 ... ob = (ob, self.__counter)
1181 >>> class A: # not in docs from here, just testing the ExtendedRef
1185 >>> r = ExtendedRef(a, foo=1, bar="baz")
1199 >>> _id2obj_dict = weakref.WeakValueDictionary()
1200 >>> def remember(obj):
1202 ... _id2obj_dict[oid] = obj
1205 >>> def id2obj(oid):
1206 ... return _id2obj_dict[oid]
1208 >>> a = A() # from here, just testing
1209 >>> a_id = remember(a)
1210 >>> id2obj(a_id) is a
1215 ... except KeyError:
1218 ... print 'WeakValueDictionary error'
1223 __test__
= {'libreftest' : libreftest
}
1226 test_support
.run_unittest(
1229 WeakValueDictionaryTestCase
,
1230 WeakKeyDictionaryTestCase
,
1231 SubclassableWeakrefTestCase
,
1233 test_support
.run_doctest(sys
.modules
[__name__
])
1236 if __name__
== "__main__":