Lib/test/test_weakref.py

   1 import sys
   2 import unittest
   3 import UserList
   4 import weakref
   5
   6 from test import test_support
   7
   8
   9 class C:
  10     def method(self):
  11         pass
  12
  13
  14 class Callable:
  15     bar = None
  16
  17     def __call__(self, x):
  18         self.bar = x
  19
  20
  21 def create_function():
  22     def f(): pass
  23     return f
  24
  25 def create_bound_method():
  26     return C().method
  27
  28 def create_unbound_method():
  29     return C.method
  30
  31
  32 class TestBase(unittest.TestCase):
  33
  34     def setUp(self):
  35         self.cbcalled = 0
  36
  37     def callback(self, ref):
  38         self.cbcalled += 1
  39
  40
  41 class ReferencesTestCase(TestBase):
  42
  43     def test_basic_ref(self):
  44         self.check_basic_ref(C)
  45         self.check_basic_ref(create_function)
  46         self.check_basic_ref(create_bound_method)
  47         self.check_basic_ref(create_unbound_method)
  48
  49         # Just make sure the tp_repr handler doesn't raise an exception.
  50         # Live reference:
  51         o = C()
  52         wr = weakref.ref(o)
  53         `wr`
  54         # Dead reference:
  55         del o
  56         `wr`
  57
  58     def test_basic_callback(self):
  59         self.check_basic_callback(C)
  60         self.check_basic_callback(create_function)
  61         self.check_basic_callback(create_bound_method)
  62         self.check_basic_callback(create_unbound_method)
  63
  64     def test_multiple_callbacks(self):
  65         o = C()
  66         ref1 = weakref.ref(o, self.callback)
  67         ref2 = weakref.ref(o, self.callback)
  68         del o
  69         self.assert_(ref1() is None,
  70                      "expected reference to be invalidated")
  71         self.assert_(ref2() is None,
  72                      "expected reference to be invalidated")
  73         self.assert_(self.cbcalled == 2,
  74                      "callback not called the right number of times")
  75
  76     def test_multiple_selfref_callbacks(self):
  77         # Make sure all references are invalidated before callbacks are called
  78         #
  79         # What's important here is that we're using the first
  80         # reference in the callback invoked on the second reference
  81         # (the most recently created ref is cleaned up first).  This
  82         # tests that all references to the object are invalidated
  83         # before any of the callbacks are invoked, so that we only
  84         # have one invocation of _weakref.c:cleanup_helper() active
  85         # for a particular object at a time.
  86         #
  87         def callback(object, self=self):
  88             self.ref()
  89         c = C()
  90         self.ref = weakref.ref(c, callback)
  91         ref1 = weakref.ref(c, callback)
  92         del c
  93
  94     def test_proxy_ref(self):
  95         o = C()
  96         o.bar = 1
  97         ref1 = weakref.proxy(o, self.callback)
  98         ref2 = weakref.proxy(o, self.callback)
  99         del o
 100
 101         def check(proxy):
 102             proxy.bar
 103
 104         self.assertRaises(weakref.ReferenceError, check, ref1)
 105         self.assertRaises(weakref.ReferenceError, check, ref2)
 106         self.assert_(self.cbcalled == 2)
 107
 108     def check_basic_ref(self, factory):
 109         o = factory()
 110         ref = weakref.ref(o)
 111         self.assert_(ref() is not None,
 112                      "weak reference to live object should be live")
 113         o2 = ref()
 114         self.assert_(o is o2,
 115                      "<ref>() should return original object if live")
 116
 117     def check_basic_callback(self, factory):
 118         self.cbcalled = 0
 119         o = factory()
 120         ref = weakref.ref(o, self.callback)
 121         del o
 122         self.assert_(self.cbcalled == 1,
 123                      "callback did not properly set 'cbcalled'")
 124         self.assert_(ref() is None,
 125                      "ref2 should be dead after deleting object reference")
 126
 127     def test_ref_reuse(self):
 128         o = C()
 129         ref1 = weakref.ref(o)
 130         # create a proxy to make sure that there's an intervening creation
 131         # between these two; it should make no difference
 132         proxy = weakref.proxy(o)
 133         ref2 = weakref.ref(o)
 134         self.assert_(ref1 is ref2,
 135                      "reference object w/out callback should be re-used")
 136
 137         o = C()
 138         proxy = weakref.proxy(o)
 139         ref1 = weakref.ref(o)
 140         ref2 = weakref.ref(o)
 141         self.assert_(ref1 is ref2,
 142                      "reference object w/out callback should be re-used")
 143         self.assert_(weakref.getweakrefcount(o) == 2,
 144                      "wrong weak ref count for object")
 145         del proxy
 146         self.assert_(weakref.getweakrefcount(o) == 1,
 147                      "wrong weak ref count for object after deleting proxy")
 148
 149     def test_proxy_reuse(self):
 150         o = C()
 151         proxy1 = weakref.proxy(o)
 152         ref = weakref.ref(o)
 153         proxy2 = weakref.proxy(o)
 154         self.assert_(proxy1 is proxy2,
 155                      "proxy object w/out callback should have been re-used")
 156
 157     def test_basic_proxy(self):
 158         o = C()
 159         self.check_proxy(o, weakref.proxy(o))
 160
 161         L = UserList.UserList()
 162         p = weakref.proxy(L)
 163         self.failIf(p, "proxy for empty UserList should be false")
 164         p.append(12)
 165         self.assertEqual(len(L), 1)
 166         self.failUnless(p, "proxy for non-empty UserList should be true")
 167         p[:] = [2, 3]
 168         self.assertEqual(len(L), 2)
 169         self.assertEqual(len(p), 2)
 170         self.failUnless(3 in p, "proxy didn't support __contains__() properly")
 171         p[1] = 5
 172         self.assertEqual(L[1], 5)
 173         self.assertEqual(p[1], 5)
 174         L2 = UserList.UserList(L)
 175         p2 = weakref.proxy(L2)
 176         self.assertEqual(p, p2)
 177         ## self.assertEqual(`L2`, `p2`)
 178         L3 = UserList.UserList(range(10))
 179         p3 = weakref.proxy(L3)
 180         self.assertEqual(L3[:], p3[:])
 181         self.assertEqual(L3[5:], p3[5:])
 182         self.assertEqual(L3[:5], p3[:5])
 183         self.assertEqual(L3[2:5], p3[2:5])
 184
 185     def test_callable_proxy(self):
 186         o = Callable()
 187         ref1 = weakref.proxy(o)
 188
 189         self.check_proxy(o, ref1)
 190
 191         self.assert_(type(ref1) is weakref.CallableProxyType,
 192                      "proxy is not of callable type")
 193         ref1('twinkies!')
 194         self.assert_(o.bar == 'twinkies!',
 195                      "call through proxy not passed through to original")
 196         ref1(x='Splat.')
 197         self.assert_(o.bar == 'Splat.',
 198                      "call through proxy not passed through to original")
 199
 200         # expect due to too few args
 201         self.assertRaises(TypeError, ref1)
 202
 203         # expect due to too many args
 204         self.assertRaises(TypeError, ref1, 1, 2, 3)
 205
 206     def check_proxy(self, o, proxy):
 207         o.foo = 1
 208         self.assert_(proxy.foo == 1,
 209                      "proxy does not reflect attribute addition")
 210         o.foo = 2
 211         self.assert_(proxy.foo == 2,
 212                      "proxy does not reflect attribute modification")
 213         del o.foo
 214         self.assert_(not hasattr(proxy, 'foo'),
 215                      "proxy does not reflect attribute removal")
 216
 217         proxy.foo = 1
 218         self.assert_(o.foo == 1,
 219                      "object does not reflect attribute addition via proxy")
 220         proxy.foo = 2
 221         self.assert_(
 222             o.foo == 2,
 223             "object does not reflect attribute modification via proxy")
 224         del proxy.foo
 225         self.assert_(not hasattr(o, 'foo'),
 226                      "object does not reflect attribute removal via proxy")
 227
 228     def test_getweakrefcount(self):
 229         o = C()
 230         ref1 = weakref.ref(o)
 231         ref2 = weakref.ref(o, self.callback)
 232         self.assert_(weakref.getweakrefcount(o) == 2,
 233                      "got wrong number of weak reference objects")
 234
 235         proxy1 = weakref.proxy(o)
 236         proxy2 = weakref.proxy(o, self.callback)
 237         self.assert_(weakref.getweakrefcount(o) == 4,
 238                      "got wrong number of weak reference objects")
 239
 240     def test_getweakrefs(self):
 241         o = C()
 242         ref1 = weakref.ref(o, self.callback)
 243         ref2 = weakref.ref(o, self.callback)
 244         del ref1
 245         self.assert_(weakref.getweakrefs(o) == [ref2],
 246                      "list of refs does not match")
 247
 248         o = C()
 249         ref1 = weakref.ref(o, self.callback)
 250         ref2 = weakref.ref(o, self.callback)
 251         del ref2
 252         self.assert_(weakref.getweakrefs(o) == [ref1],
 253                      "list of refs does not match")
 254
 255     def test_newstyle_number_ops(self):
 256         class F(float):
 257             pass
 258         f = F(2.0)
 259         p = weakref.proxy(f)
 260         self.assert_(p + 1.0 == 3.0)
 261         self.assert_(1.0 + p == 3.0)  # this used to SEGV
 262
 263     def test_callbacks_protected(self):
 264         # Callbacks protected from already-set exceptions?
 265         # Regression test for SF bug #478534.
 266         class BogusError(Exception):
 267             pass
 268         data = {}
 269         def remove(k):
 270             del data[k]
 271         def encapsulate():
 272             f = lambda : ()
 273             data[weakref.ref(f, remove)] = None
 274             raise BogusError
 275         try:
 276             encapsulate()
 277         except BogusError:
 278             pass
 279         else:
 280             self.fail("exception not properly restored")
 281         try:
 282             encapsulate()
 283         except BogusError:
 284             pass
 285         else:
 286             self.fail("exception not properly restored")
 287
 288
 289 class Object:
 290     def __init__(self, arg):
 291         self.arg = arg
 292     def __repr__(self):
 293         return "<Object %r>" % self.arg
 294
 295
 296 class MappingTestCase(TestBase):
 297
 298     COUNT = 10
 299
 300     def test_weak_values(self):
 301         #
 302         #  This exercises d.copy(), d.items(), d[], del d[], len(d).
 303         #
 304         dict, objects = self.make_weak_valued_dict()
 305         for o in objects:
 306             self.assert_(weakref.getweakrefcount(o) == 1,
 307                          "wrong number of weak references to %r!" % o)
 308             self.assert_(o is dict[o.arg],
 309                          "wrong object returned by weak dict!")
 310         items1 = dict.items()
 311         items2 = dict.copy().items()
 312         items1.sort()
 313         items2.sort()
 314         self.assert_(items1 == items2,
 315                      "cloning of weak-valued dictionary did not work!")
 316         del items1, items2
 317         self.assert_(len(dict) == self.COUNT)
 318         del objects[0]
 319         self.assert_(len(dict) == (self.COUNT - 1),
 320                      "deleting object did not cause dictionary update")
 321         del objects, o
 322         self.assert_(len(dict) == 0,
 323                      "deleting the values did not clear the dictionary")
 324         # regression on SF bug #447152:
 325         dict = weakref.WeakValueDictionary()
 326         self.assertRaises(KeyError, dict.__getitem__, 1)
 327         dict[2] = C()
 328         self.assertRaises(KeyError, dict.__getitem__, 2)
 329
 330     def test_weak_keys(self):
 331         #
 332         #  This exercises d.copy(), d.items(), d[] = v, d[], del d[],
 333         #  len(d), d.has_key().
 334         #
 335         dict, objects = self.make_weak_keyed_dict()
 336         for o in objects:
 337             self.assert_(weakref.getweakrefcount(o) == 1,
 338                          "wrong number of weak references to %r!" % o)
 339             self.assert_(o.arg is dict[o],
 340                          "wrong object returned by weak dict!")
 341         items1 = dict.items()
 342         items2 = dict.copy().items()
 343         items1.sort()
 344         items2.sort()
 345         self.assert_(items1 == items2,
 346                      "cloning of weak-keyed dictionary did not work!")
 347         del items1, items2
 348         self.assert_(len(dict) == self.COUNT)
 349         del objects[0]
 350         self.assert_(len(dict) == (self.COUNT - 1),
 351                      "deleting object did not cause dictionary update")
 352         del objects, o
 353         self.assert_(len(dict) == 0,
 354                      "deleting the keys did not clear the dictionary")
 355         o = Object(42)
 356         dict[o] = "What is the meaning of the universe?"
 357         self.assert_(dict.has_key(o))
 358         self.assert_(not dict.has_key(34))
 359
 360     def test_weak_keyed_iters(self):
 361         dict, objects = self.make_weak_keyed_dict()
 362         self.check_iters(dict)
 363
 364     def test_weak_valued_iters(self):
 365         dict, objects = self.make_weak_valued_dict()
 366         self.check_iters(dict)
 367
 368     def check_iters(self, dict):
 369         # item iterator:
 370         items = dict.items()
 371         for item in dict.iteritems():
 372             items.remove(item)
 373         self.assert_(len(items) == 0, "iteritems() did not touch all items")
 374
 375         # key iterator, via __iter__():
 376         keys = dict.keys()
 377         for k in dict:
 378             keys.remove(k)
 379         self.assert_(len(keys) == 0, "__iter__() did not touch all keys")
 380
 381         # key iterator, via iterkeys():
 382         keys = dict.keys()
 383         for k in dict.iterkeys():
 384             keys.remove(k)
 385         self.assert_(len(keys) == 0, "iterkeys() did not touch all keys")
 386
 387         # value iterator:
 388         values = dict.values()
 389         for v in dict.itervalues():
 390             values.remove(v)
 391         self.assert_(len(values) == 0, "itervalues() did not touch all values")
 392
 393     def test_make_weak_keyed_dict_from_dict(self):
 394         o = Object(3)
 395         dict = weakref.WeakKeyDictionary({o:364})
 396         self.assert_(dict[o] == 364)
 397
 398     def test_make_weak_keyed_dict_from_weak_keyed_dict(self):
 399         o = Object(3)
 400         dict = weakref.WeakKeyDictionary({o:364})
 401         dict2 = weakref.WeakKeyDictionary(dict)
 402         self.assert_(dict[o] == 364)
 403
 404     def make_weak_keyed_dict(self):
 405         dict = weakref.WeakKeyDictionary()
 406         objects = map(Object, range(self.COUNT))
 407         for o in objects:
 408             dict[o] = o.arg
 409         return dict, objects
 410
 411     def make_weak_valued_dict(self):
 412         dict = weakref.WeakValueDictionary()
 413         objects = map(Object, range(self.COUNT))
 414         for o in objects:
 415             dict[o.arg] = o
 416         return dict, objects
 417
 418     def check_popitem(self, klass, key1, value1, key2, value2):
 419         weakdict = klass()
 420         weakdict[key1] = value1
 421         weakdict[key2] = value2
 422         self.assert_(len(weakdict) == 2)
 423         k, v = weakdict.popitem()
 424         self.assert_(len(weakdict) == 1)
 425         if k is key1:
 426             self.assert_(v is value1)
 427         else:
 428             self.assert_(v is value2)
 429         k, v = weakdict.popitem()
 430         self.assert_(len(weakdict) == 0)
 431         if k is key1:
 432             self.assert_(v is value1)
 433         else:
 434             self.assert_(v is value2)
 435
 436     def test_weak_valued_dict_popitem(self):
 437         self.check_popitem(weakref.WeakValueDictionary,
 438                            "key1", C(), "key2", C())
 439
 440     def test_weak_keyed_dict_popitem(self):
 441         self.check_popitem(weakref.WeakKeyDictionary,
 442                            C(), "value 1", C(), "value 2")
 443
 444     def check_setdefault(self, klass, key, value1, value2):
 445         self.assert_(value1 is not value2,
 446                      "invalid test"
 447                      " -- value parameters must be distinct objects")
 448         weakdict = klass()
 449         o = weakdict.setdefault(key, value1)
 450         self.assert_(o is value1)
 451         self.assert_(weakdict.has_key(key))
 452         self.assert_(weakdict.get(key) is value1)
 453         self.assert_(weakdict[key] is value1)
 454
 455         o = weakdict.setdefault(key, value2)
 456         self.assert_(o is value1)
 457         self.assert_(weakdict.has_key(key))
 458         self.assert_(weakdict.get(key) is value1)
 459         self.assert_(weakdict[key] is value1)
 460
 461     def test_weak_valued_dict_setdefault(self):
 462         self.check_setdefault(weakref.WeakValueDictionary,
 463                               "key", C(), C())
 464
 465     def test_weak_keyed_dict_setdefault(self):
 466         self.check_setdefault(weakref.WeakKeyDictionary,
 467                               C(), "value 1", "value 2")
 468
 469     def check_update(self, klass, dict):
 470         #
 471         #  This exercises d.update(), len(d), d.keys(), d.has_key(),
 472         #  d.get(), d[].
 473         #
 474         weakdict = klass()
 475         weakdict.update(dict)
 476         self.assert_(len(weakdict) == len(dict))
 477         for k in weakdict.keys():
 478             self.assert_(dict.has_key(k),
 479                          "mysterious new key appeared in weak dict")
 480             v = dict.get(k)
 481             self.assert_(v is weakdict[k])
 482             self.assert_(v is weakdict.get(k))
 483         for k in dict.keys():
 484             self.assert_(weakdict.has_key(k),
 485                          "original key disappeared in weak dict")
 486             v = dict[k]
 487             self.assert_(v is weakdict[k])
 488             self.assert_(v is weakdict.get(k))
 489
 490     def test_weak_valued_dict_update(self):
 491         self.check_update(weakref.WeakValueDictionary,
 492                           {1: C(), 'a': C(), C(): C()})
 493
 494     def test_weak_keyed_dict_update(self):
 495         self.check_update(weakref.WeakKeyDictionary,
 496                           {C(): 1, C(): 2, C(): 3})
 497
 498     def test_weak_keyed_delitem(self):
 499         d = weakref.WeakKeyDictionary()
 500         o1 = Object('1')
 501         o2 = Object('2')
 502         d[o1] = 'something'
 503         d[o2] = 'something'
 504         self.assert_(len(d) == 2)
 505         del d[o1]
 506         self.assert_(len(d) == 1)
 507         self.assert_(d.keys() == [o2])
 508
 509     def test_weak_valued_delitem(self):
 510         d = weakref.WeakValueDictionary()
 511         o1 = Object('1')
 512         o2 = Object('2')
 513         d['something'] = o1
 514         d['something else'] = o2
 515         self.assert_(len(d) == 2)
 516         del d['something']
 517         self.assert_(len(d) == 1)
 518         self.assert_(d.items() == [('something else', o2)])
 519
 520 from test_userdict import TestMappingProtocol
 521
 522 class WeakValueDictionaryTestCase(TestMappingProtocol):
 523     """Check that WeakValueDictionary class conforms to the mapping protocol"""
 524     __ref = {"key1":Object(1), "key2":Object(2), "key3":Object(3)}
 525     _tested_class = weakref.WeakValueDictionary
 526     def _reference(self):
 527         return self.__ref.copy()
 528
 529 class WeakKeyDictionaryTestCase(TestMappingProtocol):
 530     """Check that WeakKeyDictionary class conforms to the mapping protocol"""
 531     __ref = {Object("key1"):1, Object("key2"):2, Object("key3"):3}
 532     _tested_class = weakref.WeakKeyDictionary
 533     def _reference(self):
 534         return self.__ref.copy()
 535
 536 def test_main():
 537     suite = unittest.TestSuite()
 538     suite.addTest(unittest.makeSuite(ReferencesTestCase))
 539     suite.addTest(unittest.makeSuite(MappingTestCase))
 540     suite.addTest(unittest.makeSuite(WeakValueDictionaryTestCase))
 541     suite.addTest(unittest.makeSuite(WeakKeyDictionaryTestCase))
 542     test_support.run_suite(suite)
 543
 544
 545 if __name__ == "__main__":
 546     test_main()