Home | History | Annotate | Download | only in test
      1 import unittest
      2 from test import test_support
      3 from weakref import proxy, ref, WeakSet
      4 import operator
      5 import copy
      6 import string
      7 import os
      8 from random import randrange, shuffle
      9 import sys
     10 import warnings
     11 import collections
     12 import gc
     13 import contextlib
     14 
     15 
     16 class Foo:
     17     pass
     18 
     19 class SomeClass(object):
     20     def __init__(self, value):
     21         self.value = value
     22     def __eq__(self, other):
     23         if type(other) != type(self):
     24             return False
     25         return other.value == self.value
     26 
     27     def __ne__(self, other):
     28         return not self.__eq__(other)
     29 
     30     def __hash__(self):
     31         return hash((SomeClass, self.value))
     32 
     33 class RefCycle(object):
     34     def __init__(self):
     35         self.cycle = self
     36 
     37 class TestWeakSet(unittest.TestCase):
     38 
     39     def setUp(self):
     40         # need to keep references to them
     41         self.items = [SomeClass(c) for c in ('a', 'b', 'c')]
     42         self.items2 = [SomeClass(c) for c in ('x', 'y', 'z')]
     43         self.letters = [SomeClass(c) for c in string.ascii_letters]
     44         self.ab_items = [SomeClass(c) for c in 'ab']
     45         self.abcde_items = [SomeClass(c) for c in 'abcde']
     46         self.def_items = [SomeClass(c) for c in 'def']
     47         self.ab_weakset = WeakSet(self.ab_items)
     48         self.abcde_weakset = WeakSet(self.abcde_items)
     49         self.def_weakset = WeakSet(self.def_items)
     50         self.s = WeakSet(self.items)
     51         self.d = dict.fromkeys(self.items)
     52         self.obj = SomeClass('F')
     53         self.fs = WeakSet([self.obj])
     54 
     55     def test_methods(self):
     56         weaksetmethods = dir(WeakSet)
     57         for method in dir(set):
     58             if method == 'test_c_api' or method.startswith('_'):
     59                 continue
     60             self.assertIn(method, weaksetmethods,
     61                          "WeakSet missing method " + method)
     62 
     63     def test_new_or_init(self):
     64         self.assertRaises(TypeError, WeakSet, [], 2)
     65 
     66     def test_len(self):
     67         self.assertEqual(len(self.s), len(self.d))
     68         self.assertEqual(len(self.fs), 1)
     69         del self.obj
     70         self.assertEqual(len(self.fs), 0)
     71 
     72     def test_contains(self):
     73         for c in self.letters:
     74             self.assertEqual(c in self.s, c in self.d)
     75         # 1 is not weakref'able, but that TypeError is caught by __contains__
     76         self.assertNotIn(1, self.s)
     77         self.assertIn(self.obj, self.fs)
     78         del self.obj
     79         self.assertNotIn(SomeClass('F'), self.fs)
     80 
     81     def test_union(self):
     82         u = self.s.union(self.items2)
     83         for c in self.letters:
     84             self.assertEqual(c in u, c in self.d or c in self.items2)
     85         self.assertEqual(self.s, WeakSet(self.items))
     86         self.assertEqual(type(u), WeakSet)
     87         self.assertRaises(TypeError, self.s.union, [[]])
     88         for C in set, frozenset, dict.fromkeys, list, tuple:
     89             x = WeakSet(self.items + self.items2)
     90             c = C(self.items2)
     91             self.assertEqual(self.s.union(c), x)
     92             del c
     93         self.assertEqual(len(u), len(self.items) + len(self.items2))
     94         self.items2.pop()
     95         gc.collect()
     96         self.assertEqual(len(u), len(self.items) + len(self.items2))
     97 
     98     def test_or(self):
     99         i = self.s.union(self.items2)
    100         self.assertEqual(self.s | set(self.items2), i)
    101         self.assertEqual(self.s | frozenset(self.items2), i)
    102 
    103     def test_intersection(self):
    104         s = WeakSet(self.letters)
    105         i = s.intersection(self.items2)
    106         for c in self.letters:
    107             self.assertEqual(c in i, c in self.items2 and c in self.letters)
    108         self.assertEqual(s, WeakSet(self.letters))
    109         self.assertEqual(type(i), WeakSet)
    110         for C in set, frozenset, dict.fromkeys, list, tuple:
    111             x = WeakSet([])
    112             self.assertEqual(i.intersection(C(self.items)), x)
    113         self.assertEqual(len(i), len(self.items2))
    114         self.items2.pop()
    115         gc.collect()
    116         self.assertEqual(len(i), len(self.items2))
    117 
    118     def test_isdisjoint(self):
    119         self.assertTrue(self.s.isdisjoint(WeakSet(self.items2)))
    120         self.assertTrue(not self.s.isdisjoint(WeakSet(self.letters)))
    121 
    122     def test_and(self):
    123         i = self.s.intersection(self.items2)
    124         self.assertEqual(self.s & set(self.items2), i)
    125         self.assertEqual(self.s & frozenset(self.items2), i)
    126 
    127     def test_difference(self):
    128         i = self.s.difference(self.items2)
    129         for c in self.letters:
    130             self.assertEqual(c in i, c in self.d and c not in self.items2)
    131         self.assertEqual(self.s, WeakSet(self.items))
    132         self.assertEqual(type(i), WeakSet)
    133         self.assertRaises(TypeError, self.s.difference, [[]])
    134 
    135     def test_sub(self):
    136         i = self.s.difference(self.items2)
    137         self.assertEqual(self.s - set(self.items2), i)
    138         self.assertEqual(self.s - frozenset(self.items2), i)
    139 
    140     def test_symmetric_difference(self):
    141         i = self.s.symmetric_difference(self.items2)
    142         for c in self.letters:
    143             self.assertEqual(c in i, (c in self.d) ^ (c in self.items2))
    144         self.assertEqual(self.s, WeakSet(self.items))
    145         self.assertEqual(type(i), WeakSet)
    146         self.assertRaises(TypeError, self.s.symmetric_difference, [[]])
    147         self.assertEqual(len(i), len(self.items) + len(self.items2))
    148         self.items2.pop()
    149         gc.collect()
    150         self.assertEqual(len(i), len(self.items) + len(self.items2))
    151 
    152     def test_xor(self):
    153         i = self.s.symmetric_difference(self.items2)
    154         self.assertEqual(self.s ^ set(self.items2), i)
    155         self.assertEqual(self.s ^ frozenset(self.items2), i)
    156 
    157     def test_sub_and_super(self):
    158         self.assertTrue(self.ab_weakset <= self.abcde_weakset)
    159         self.assertTrue(self.abcde_weakset <= self.abcde_weakset)
    160         self.assertTrue(self.abcde_weakset >= self.ab_weakset)
    161         self.assertFalse(self.abcde_weakset <= self.def_weakset)
    162         self.assertFalse(self.abcde_weakset >= self.def_weakset)
    163         self.assertTrue(set('a').issubset('abc'))
    164         self.assertTrue(set('abc').issuperset('a'))
    165         self.assertFalse(set('a').issubset('cbs'))
    166         self.assertFalse(set('cbs').issuperset('a'))
    167 
    168     def test_lt(self):
    169         self.assertTrue(self.ab_weakset < self.abcde_weakset)
    170         self.assertFalse(self.abcde_weakset < self.def_weakset)
    171         self.assertFalse(self.ab_weakset < self.ab_weakset)
    172         self.assertFalse(WeakSet() < WeakSet())
    173 
    174     def test_gt(self):
    175         self.assertTrue(self.abcde_weakset > self.ab_weakset)
    176         self.assertFalse(self.abcde_weakset > self.def_weakset)
    177         self.assertFalse(self.ab_weakset > self.ab_weakset)
    178         self.assertFalse(WeakSet() > WeakSet())
    179 
    180     def test_gc(self):
    181         # Create a nest of cycles to exercise overall ref count check
    182         s = WeakSet(Foo() for i in range(1000))
    183         for elem in s:
    184             elem.cycle = s
    185             elem.sub = elem
    186             elem.set = WeakSet([elem])
    187 
    188     def test_subclass_with_custom_hash(self):
    189         # Bug #1257731
    190         class H(WeakSet):
    191             def __hash__(self):
    192                 return int(id(self) & 0x7fffffff)
    193         s=H()
    194         f=set()
    195         f.add(s)
    196         self.assertIn(s, f)
    197         f.remove(s)
    198         f.add(s)
    199         f.discard(s)
    200 
    201     def test_init(self):
    202         s = WeakSet()
    203         s.__init__(self.items)
    204         self.assertEqual(s, self.s)
    205         s.__init__(self.items2)
    206         self.assertEqual(s, WeakSet(self.items2))
    207         self.assertRaises(TypeError, s.__init__, s, 2);
    208         self.assertRaises(TypeError, s.__init__, 1);
    209 
    210     def test_constructor_identity(self):
    211         s = WeakSet(self.items)
    212         t = WeakSet(s)
    213         self.assertNotEqual(id(s), id(t))
    214 
    215     def test_hash(self):
    216         self.assertRaises(TypeError, hash, self.s)
    217 
    218     def test_clear(self):
    219         self.s.clear()
    220         self.assertEqual(self.s, WeakSet([]))
    221         self.assertEqual(len(self.s), 0)
    222 
    223     def test_copy(self):
    224         dup = self.s.copy()
    225         self.assertEqual(self.s, dup)
    226         self.assertNotEqual(id(self.s), id(dup))
    227 
    228     def test_add(self):
    229         x = SomeClass('Q')
    230         self.s.add(x)
    231         self.assertIn(x, self.s)
    232         dup = self.s.copy()
    233         self.s.add(x)
    234         self.assertEqual(self.s, dup)
    235         self.assertRaises(TypeError, self.s.add, [])
    236         self.fs.add(Foo())
    237         self.assertTrue(len(self.fs) == 1)
    238         self.fs.add(self.obj)
    239         self.assertTrue(len(self.fs) == 1)
    240 
    241     def test_remove(self):
    242         x = SomeClass('a')
    243         self.s.remove(x)
    244         self.assertNotIn(x, self.s)
    245         self.assertRaises(KeyError, self.s.remove, x)
    246         self.assertRaises(TypeError, self.s.remove, [])
    247 
    248     def test_discard(self):
    249         a, q = SomeClass('a'), SomeClass('Q')
    250         self.s.discard(a)
    251         self.assertNotIn(a, self.s)
    252         self.s.discard(q)
    253         self.assertRaises(TypeError, self.s.discard, [])
    254 
    255     def test_pop(self):
    256         for i in range(len(self.s)):
    257             elem = self.s.pop()
    258             self.assertNotIn(elem, self.s)
    259         self.assertRaises(KeyError, self.s.pop)
    260 
    261     def test_update(self):
    262         retval = self.s.update(self.items2)
    263         self.assertEqual(retval, None)
    264         for c in (self.items + self.items2):
    265             self.assertIn(c, self.s)
    266         self.assertRaises(TypeError, self.s.update, [[]])
    267 
    268     def test_update_set(self):
    269         self.s.update(set(self.items2))
    270         for c in (self.items + self.items2):
    271             self.assertIn(c, self.s)
    272 
    273     def test_ior(self):
    274         self.s |= set(self.items2)
    275         for c in (self.items + self.items2):
    276             self.assertIn(c, self.s)
    277 
    278     def test_intersection_update(self):
    279         retval = self.s.intersection_update(self.items2)
    280         self.assertEqual(retval, None)
    281         for c in (self.items + self.items2):
    282             if c in self.items2 and c in self.items:
    283                 self.assertIn(c, self.s)
    284             else:
    285                 self.assertNotIn(c, self.s)
    286         self.assertRaises(TypeError, self.s.intersection_update, [[]])
    287 
    288     def test_iand(self):
    289         self.s &= set(self.items2)
    290         for c in (self.items + self.items2):
    291             if c in self.items2 and c in self.items:
    292                 self.assertIn(c, self.s)
    293             else:
    294                 self.assertNotIn(c, self.s)
    295 
    296     def test_difference_update(self):
    297         retval = self.s.difference_update(self.items2)
    298         self.assertEqual(retval, None)
    299         for c in (self.items + self.items2):
    300             if c in self.items and c not in self.items2:
    301                 self.assertIn(c, self.s)
    302             else:
    303                 self.assertNotIn(c, self.s)
    304         self.assertRaises(TypeError, self.s.difference_update, [[]])
    305         self.assertRaises(TypeError, self.s.symmetric_difference_update, [[]])
    306 
    307     def test_isub(self):
    308         self.s -= set(self.items2)
    309         for c in (self.items + self.items2):
    310             if c in self.items and c not in self.items2:
    311                 self.assertIn(c, self.s)
    312             else:
    313                 self.assertNotIn(c, self.s)
    314 
    315     def test_symmetric_difference_update(self):
    316         retval = self.s.symmetric_difference_update(self.items2)
    317         self.assertEqual(retval, None)
    318         for c in (self.items + self.items2):
    319             if (c in self.items) ^ (c in self.items2):
    320                 self.assertIn(c, self.s)
    321             else:
    322                 self.assertNotIn(c, self.s)
    323         self.assertRaises(TypeError, self.s.symmetric_difference_update, [[]])
    324 
    325     def test_ixor(self):
    326         self.s ^= set(self.items2)
    327         for c in (self.items + self.items2):
    328             if (c in self.items) ^ (c in self.items2):
    329                 self.assertIn(c, self.s)
    330             else:
    331                 self.assertNotIn(c, self.s)
    332 
    333     def test_inplace_on_self(self):
    334         t = self.s.copy()
    335         t |= t
    336         self.assertEqual(t, self.s)
    337         t &= t
    338         self.assertEqual(t, self.s)
    339         t -= t
    340         self.assertEqual(t, WeakSet())
    341         t = self.s.copy()
    342         t ^= t
    343         self.assertEqual(t, WeakSet())
    344 
    345     def test_eq(self):
    346         # issue 5964
    347         self.assertTrue(self.s == self.s)
    348         self.assertTrue(self.s == WeakSet(self.items))
    349         self.assertFalse(self.s == set(self.items))
    350         self.assertFalse(self.s == list(self.items))
    351         self.assertFalse(self.s == tuple(self.items))
    352         self.assertFalse(self.s == 1)
    353 
    354     def test_weak_destroy_while_iterating(self):
    355         # Issue #7105: iterators shouldn't crash when a key is implicitly removed
    356         # Create new items to be sure no-one else holds a reference
    357         items = [SomeClass(c) for c in ('a', 'b', 'c')]
    358         s = WeakSet(items)
    359         it = iter(s)
    360         next(it)             # Trigger internal iteration
    361         # Destroy an item
    362         del items[-1]
    363         gc.collect()    # just in case
    364         # We have removed either the first consumed items, or another one
    365         self.assertIn(len(list(it)), [len(items), len(items) - 1])
    366         del it
    367         # The removal has been committed
    368         self.assertEqual(len(s), len(items))
    369 
    370     def test_weak_destroy_and_mutate_while_iterating(self):
    371         # Issue #7105: iterators shouldn't crash when a key is implicitly removed
    372         items = [SomeClass(c) for c in string.ascii_letters]
    373         s = WeakSet(items)
    374         @contextlib.contextmanager
    375         def testcontext():
    376             try:
    377                 it = iter(s)
    378                 next(it)
    379                 # Schedule an item for removal and recreate it
    380                 u = SomeClass(str(items.pop()))
    381                 gc.collect()      # just in case
    382                 yield u
    383             finally:
    384                 it = None           # should commit all removals
    385 
    386         with testcontext() as u:
    387             self.assertNotIn(u, s)
    388         with testcontext() as u:
    389             self.assertRaises(KeyError, s.remove, u)
    390         self.assertNotIn(u, s)
    391         with testcontext() as u:
    392             s.add(u)
    393         self.assertIn(u, s)
    394         t = s.copy()
    395         with testcontext() as u:
    396             s.update(t)
    397         self.assertEqual(len(s), len(t))
    398         with testcontext() as u:
    399             s.clear()
    400         self.assertEqual(len(s), 0)
    401 
    402     def test_len_cycles(self):
    403         N = 20
    404         items = [RefCycle() for i in range(N)]
    405         s = WeakSet(items)
    406         del items
    407         it = iter(s)
    408         try:
    409             next(it)
    410         except StopIteration:
    411             pass
    412         gc.collect()
    413         n1 = len(s)
    414         del it
    415         gc.collect()
    416         n2 = len(s)
    417         # one item may be kept alive inside the iterator
    418         self.assertIn(n1, (0, 1))
    419         self.assertEqual(n2, 0)
    420 
    421     def test_len_race(self):
    422         # Extended sanity checks for len() in the face of cyclic collection
    423         self.addCleanup(gc.set_threshold, *gc.get_threshold())
    424         for th in range(1, 100):
    425             N = 20
    426             gc.collect(0)
    427             gc.set_threshold(th, th, th)
    428             items = [RefCycle() for i in range(N)]
    429             s = WeakSet(items)
    430             del items
    431             # All items will be collected at next garbage collection pass
    432             it = iter(s)
    433             try:
    434                 next(it)
    435             except StopIteration:
    436                 pass
    437             n1 = len(s)
    438             del it
    439             n2 = len(s)
    440             self.assertGreaterEqual(n1, 0)
    441             self.assertLessEqual(n1, N)
    442             self.assertGreaterEqual(n2, 0)
    443             self.assertLessEqual(n2, n1)
    444 
    445 
    446 def test_main(verbose=None):
    447     test_support.run_unittest(TestWeakSet)
    448 
    449 if __name__ == "__main__":
    450     test_main(verbose=True)
    451