test_weakset.py 15 KB

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