Python code coverage for Lib/test/test_finalization.py

#	count	content
1	n/a	"""
2	n/a	Tests for object finalization semantics, as outlined in PEP 442.
3	n/a	"""
4	n/a
5	n/a	import contextlib
6	n/a	import gc
7	n/a	import unittest
8	n/a	import weakref
9	n/a
10	n/a	try:
11	n/a	from _testcapi import with_tp_del
12	n/a	except ImportError:
13	n/a	def with_tp_del(cls):
14	n/a	class C(object):
15	n/a	def __new__(cls, args, *kwargs):
16	n/a	raise TypeError('requires _testcapi.with_tp_del')
17	n/a	return C
18	n/a
19	n/a	from test import support
20	n/a
21	n/a
22	n/a	class NonGCSimpleBase:
23	n/a	"""
24	n/a	The base class for all the objects under test, equipped with various
25	n/a	testing features.
26	n/a	"""
27	n/a
28	n/a	survivors = []
29	n/a	del_calls = []
30	n/a	tp_del_calls = []
31	n/a	errors = []
32	n/a
33	n/a	_cleaning = False
34	n/a
35	n/a	__slots__ = ()
36	n/a
37	n/a	@classmethod
38	n/a	def _cleanup(cls):
39	n/a	cls.survivors.clear()
40	n/a	cls.errors.clear()
41	n/a	gc.garbage.clear()
42	n/a	gc.collect()
43	n/a	cls.del_calls.clear()
44	n/a	cls.tp_del_calls.clear()
45	n/a
46	n/a	@classmethod
47	n/a	@contextlib.contextmanager
48	n/a	def test(cls):
49	n/a	"""
50	n/a	A context manager to use around all finalization tests.
51	n/a	"""
52	n/a	with support.disable_gc():
53	n/a	cls.del_calls.clear()
54	n/a	cls.tp_del_calls.clear()
55	n/a	NonGCSimpleBase._cleaning = False
56	n/a	try:
57	n/a	yield
58	n/a	if cls.errors:
59	n/a	raise cls.errors[0]
60	n/a	finally:
61	n/a	NonGCSimpleBase._cleaning = True
62	n/a	cls._cleanup()
63	n/a
64	n/a	def check_sanity(self):
65	n/a	"""
66	n/a	Check the object is sane (non-broken).
67	n/a	"""
68	n/a
69	n/a	def __del__(self):
70	n/a	"""
71	n/a	PEP 442 finalizer. Record that this was called, check the
72	n/a	object is in a sane state, and invoke a side effect.
73	n/a	"""
74	n/a	try:
75	n/a	if not self._cleaning:
76	n/a	self.del_calls.append(id(self))
77	n/a	self.check_sanity()
78	n/a	self.side_effect()
79	n/a	except Exception as e:
80	n/a	self.errors.append(e)
81	n/a
82	n/a	def side_effect(self):
83	n/a	"""
84	n/a	A side effect called on destruction.
85	n/a	"""
86	n/a
87	n/a
88	n/a	class SimpleBase(NonGCSimpleBase):
89	n/a
90	n/a	def __init__(self):
91	n/a	self.id_ = id(self)
92	n/a
93	n/a	def check_sanity(self):
94	n/a	assert self.id_ == id(self)
95	n/a
96	n/a
97	n/a	class NonGC(NonGCSimpleBase):
98	n/a	__slots__ = ()
99	n/a
100	n/a	class NonGCResurrector(NonGCSimpleBase):
101	n/a	__slots__ = ()
102	n/a
103	n/a	def side_effect(self):
104	n/a	"""
105	n/a	Resurrect self by storing self in a class-wide list.
106	n/a	"""
107	n/a	self.survivors.append(self)
108	n/a
109	n/a	class Simple(SimpleBase):
110	n/a	pass
111	n/a
112	n/a	class SimpleResurrector(NonGCResurrector, SimpleBase):
113	n/a	pass
114	n/a
115	n/a
116	n/a	class TestBase:
117	n/a
118	n/a	def setUp(self):
119	n/a	self.old_garbage = gc.garbage[:]
120	n/a	gc.garbage[:] = []
121	n/a
122	n/a	def tearDown(self):
123	n/a	# None of the tests here should put anything in gc.garbage
124	n/a	try:
125	n/a	self.assertEqual(gc.garbage, [])
126	n/a	finally:
127	n/a	del self.old_garbage
128	n/a	gc.collect()
129	n/a
130	n/a	def assert_del_calls(self, ids):
131	n/a	self.assertEqual(sorted(SimpleBase.del_calls), sorted(ids))
132	n/a
133	n/a	def assert_tp_del_calls(self, ids):
134	n/a	self.assertEqual(sorted(SimpleBase.tp_del_calls), sorted(ids))
135	n/a
136	n/a	def assert_survivors(self, ids):
137	n/a	self.assertEqual(sorted(id(x) for x in SimpleBase.survivors), sorted(ids))
138	n/a
139	n/a	def assert_garbage(self, ids):
140	n/a	self.assertEqual(sorted(id(x) for x in gc.garbage), sorted(ids))
141	n/a
142	n/a	def clear_survivors(self):
143	n/a	SimpleBase.survivors.clear()
144	n/a
145	n/a
146	n/a	class SimpleFinalizationTest(TestBase, unittest.TestCase):
147	n/a	"""
148	n/a	Test finalization without refcycles.
149	n/a	"""
150	n/a
151	n/a	def test_simple(self):
152	n/a	with SimpleBase.test():
153	n/a	s = Simple()
154	n/a	ids = [id(s)]
155	n/a	wr = weakref.ref(s)
156	n/a	del s
157	n/a	gc.collect()
158	n/a	self.assert_del_calls(ids)
159	n/a	self.assert_survivors([])
160	n/a	self.assertIs(wr(), None)
161	n/a	gc.collect()
162	n/a	self.assert_del_calls(ids)
163	n/a	self.assert_survivors([])
164	n/a
165	n/a	def test_simple_resurrect(self):
166	n/a	with SimpleBase.test():
167	n/a	s = SimpleResurrector()
168	n/a	ids = [id(s)]
169	n/a	wr = weakref.ref(s)
170	n/a	del s
171	n/a	gc.collect()
172	n/a	self.assert_del_calls(ids)
173	n/a	self.assert_survivors(ids)
174	n/a	self.assertIsNot(wr(), None)
175	n/a	self.clear_survivors()
176	n/a	gc.collect()
177	n/a	self.assert_del_calls(ids)
178	n/a	self.assert_survivors([])
179	n/a	self.assertIs(wr(), None)
180	n/a
181	n/a	def test_non_gc(self):
182	n/a	with SimpleBase.test():
183	n/a	s = NonGC()
184	n/a	self.assertFalse(gc.is_tracked(s))
185	n/a	ids = [id(s)]
186	n/a	del s
187	n/a	gc.collect()
188	n/a	self.assert_del_calls(ids)
189	n/a	self.assert_survivors([])
190	n/a	gc.collect()
191	n/a	self.assert_del_calls(ids)
192	n/a	self.assert_survivors([])
193	n/a
194	n/a	def test_non_gc_resurrect(self):
195	n/a	with SimpleBase.test():
196	n/a	s = NonGCResurrector()
197	n/a	self.assertFalse(gc.is_tracked(s))
198	n/a	ids = [id(s)]
199	n/a	del s
200	n/a	gc.collect()
201	n/a	self.assert_del_calls(ids)
202	n/a	self.assert_survivors(ids)
203	n/a	self.clear_survivors()
204	n/a	gc.collect()
205	n/a	self.assert_del_calls(ids * 2)
206	n/a	self.assert_survivors(ids)
207	n/a
208	n/a
209	n/a	class SelfCycleBase:
210	n/a
211	n/a	def __init__(self):
212	n/a	super().__init__()
213	n/a	self.ref = self
214	n/a
215	n/a	def check_sanity(self):
216	n/a	super().check_sanity()
217	n/a	assert self.ref is self
218	n/a
219	n/a	class SimpleSelfCycle(SelfCycleBase, Simple):
220	n/a	pass
221	n/a
222	n/a	class SelfCycleResurrector(SelfCycleBase, SimpleResurrector):
223	n/a	pass
224	n/a
225	n/a	class SuicidalSelfCycle(SelfCycleBase, Simple):
226	n/a
227	n/a	def side_effect(self):
228	n/a	"""
229	n/a	Explicitly break the reference cycle.
230	n/a	"""
231	n/a	self.ref = None
232	n/a
233	n/a
234	n/a	class SelfCycleFinalizationTest(TestBase, unittest.TestCase):
235	n/a	"""
236	n/a	Test finalization of an object having a single cyclic reference to
237	n/a	itself.
238	n/a	"""
239	n/a
240	n/a	def test_simple(self):
241	n/a	with SimpleBase.test():
242	n/a	s = SimpleSelfCycle()
243	n/a	ids = [id(s)]
244	n/a	wr = weakref.ref(s)
245	n/a	del s
246	n/a	gc.collect()
247	n/a	self.assert_del_calls(ids)
248	n/a	self.assert_survivors([])
249	n/a	self.assertIs(wr(), None)
250	n/a	gc.collect()
251	n/a	self.assert_del_calls(ids)
252	n/a	self.assert_survivors([])
253	n/a
254	n/a	def test_simple_resurrect(self):
255	n/a	# Test that __del__ can resurrect the object being finalized.
256	n/a	with SimpleBase.test():
257	n/a	s = SelfCycleResurrector()
258	n/a	ids = [id(s)]
259	n/a	wr = weakref.ref(s)
260	n/a	del s
261	n/a	gc.collect()
262	n/a	self.assert_del_calls(ids)
263	n/a	self.assert_survivors(ids)
264	n/a	# XXX is this desirable?
265	n/a	self.assertIs(wr(), None)
266	n/a	# When trying to destroy the object a second time, __del__
267	n/a	# isn't called anymore (and the object isn't resurrected).
268	n/a	self.clear_survivors()
269	n/a	gc.collect()
270	n/a	self.assert_del_calls(ids)
271	n/a	self.assert_survivors([])
272	n/a	self.assertIs(wr(), None)
273	n/a
274	n/a	def test_simple_suicide(self):
275	n/a	# Test the GC is able to deal with an object that kills its last
276	n/a	# reference during __del__.
277	n/a	with SimpleBase.test():
278	n/a	s = SuicidalSelfCycle()
279	n/a	ids = [id(s)]
280	n/a	wr = weakref.ref(s)
281	n/a	del s
282	n/a	gc.collect()
283	n/a	self.assert_del_calls(ids)
284	n/a	self.assert_survivors([])
285	n/a	self.assertIs(wr(), None)
286	n/a	gc.collect()
287	n/a	self.assert_del_calls(ids)
288	n/a	self.assert_survivors([])
289	n/a	self.assertIs(wr(), None)
290	n/a
291	n/a
292	n/a	class ChainedBase:
293	n/a
294	n/a	def chain(self, left):
295	n/a	self.suicided = False
296	n/a	self.left = left
297	n/a	left.right = self
298	n/a
299	n/a	def check_sanity(self):
300	n/a	super().check_sanity()
301	n/a	if self.suicided:
302	n/a	assert self.left is None
303	n/a	assert self.right is None
304	n/a	else:
305	n/a	left = self.left
306	n/a	if left.suicided:
307	n/a	assert left.right is None
308	n/a	else:
309	n/a	assert left.right is self
310	n/a	right = self.right
311	n/a	if right.suicided:
312	n/a	assert right.left is None
313	n/a	else:
314	n/a	assert right.left is self
315	n/a
316	n/a	class SimpleChained(ChainedBase, Simple):
317	n/a	pass
318	n/a
319	n/a	class ChainedResurrector(ChainedBase, SimpleResurrector):
320	n/a	pass
321	n/a
322	n/a	class SuicidalChained(ChainedBase, Simple):
323	n/a
324	n/a	def side_effect(self):
325	n/a	"""
326	n/a	Explicitly break the reference cycle.
327	n/a	"""
328	n/a	self.suicided = True
329	n/a	self.left = None
330	n/a	self.right = None
331	n/a
332	n/a
333	n/a	class CycleChainFinalizationTest(TestBase, unittest.TestCase):
334	n/a	"""
335	n/a	Test finalization of a cyclic chain. These tests are similar in
336	n/a	spirit to the self-cycle tests above, but the collectable object
337	n/a	graph isn't trivial anymore.
338	n/a	"""
339	n/a
340	n/a	def build_chain(self, classes):
341	n/a	nodes = [cls() for cls in classes]
342	n/a	for i in range(len(nodes)):
343	n/a	nodes[i].chain(nodes[i-1])
344	n/a	return nodes
345	n/a
346	n/a	def check_non_resurrecting_chain(self, classes):
347	n/a	N = len(classes)
348	n/a	with SimpleBase.test():
349	n/a	nodes = self.build_chain(classes)
350	n/a	ids = [id(s) for s in nodes]
351	n/a	wrs = [weakref.ref(s) for s in nodes]
352	n/a	del nodes
353	n/a	gc.collect()
354	n/a	self.assert_del_calls(ids)
355	n/a	self.assert_survivors([])
356	n/a	self.assertEqual([wr() for wr in wrs], [None] * N)
357	n/a	gc.collect()
358	n/a	self.assert_del_calls(ids)
359	n/a
360	n/a	def check_resurrecting_chain(self, classes):
361	n/a	N = len(classes)
362	n/a	with SimpleBase.test():
363	n/a	nodes = self.build_chain(classes)
364	n/a	N = len(nodes)
365	n/a	ids = [id(s) for s in nodes]
366	n/a	survivor_ids = [id(s) for s in nodes if isinstance(s, SimpleResurrector)]
367	n/a	wrs = [weakref.ref(s) for s in nodes]
368	n/a	del nodes
369	n/a	gc.collect()
370	n/a	self.assert_del_calls(ids)
371	n/a	self.assert_survivors(survivor_ids)
372	n/a	# XXX desirable?
373	n/a	self.assertEqual([wr() for wr in wrs], [None] * N)
374	n/a	self.clear_survivors()
375	n/a	gc.collect()
376	n/a	self.assert_del_calls(ids)
377	n/a	self.assert_survivors([])
378	n/a
379	n/a	def test_homogenous(self):
380	n/a	self.check_non_resurrecting_chain([SimpleChained] * 3)
381	n/a
382	n/a	def test_homogenous_resurrect(self):
383	n/a	self.check_resurrecting_chain([ChainedResurrector] * 3)
384	n/a
385	n/a	def test_homogenous_suicidal(self):
386	n/a	self.check_non_resurrecting_chain([SuicidalChained] * 3)
387	n/a
388	n/a	def test_heterogenous_suicidal_one(self):
389	n/a	self.check_non_resurrecting_chain([SuicidalChained, SimpleChained] * 2)
390	n/a
391	n/a	def test_heterogenous_suicidal_two(self):
392	n/a	self.check_non_resurrecting_chain(
393	n/a	[SuicidalChained] * 2 + [SimpleChained] * 2)
394	n/a
395	n/a	def test_heterogenous_resurrect_one(self):
396	n/a	self.check_resurrecting_chain([ChainedResurrector, SimpleChained] * 2)
397	n/a
398	n/a	def test_heterogenous_resurrect_two(self):
399	n/a	self.check_resurrecting_chain(
400	n/a	[ChainedResurrector, SimpleChained, SuicidalChained] * 2)
401	n/a
402	n/a	def test_heterogenous_resurrect_three(self):
403	n/a	self.check_resurrecting_chain(
404	n/a	[ChainedResurrector] * 2 + [SimpleChained] * 2 + [SuicidalChained] * 2)
405	n/a
406	n/a
407	n/a	# NOTE: the tp_del slot isn't automatically inherited, so we have to call
408	n/a	# with_tp_del() for each instantiated class.
409	n/a
410	n/a	class LegacyBase(SimpleBase):
411	n/a
412	n/a	def __del__(self):
413	n/a	try:
414	n/a	# Do not invoke side_effect here, since we are now exercising
415	n/a	# the tp_del slot.
416	n/a	if not self._cleaning:
417	n/a	self.del_calls.append(id(self))
418	n/a	self.check_sanity()
419	n/a	except Exception as e:
420	n/a	self.errors.append(e)
421	n/a
422	n/a	def __tp_del__(self):
423	n/a	"""
424	n/a	Legacy (pre-PEP 442) finalizer, mapped to a tp_del slot.
425	n/a	"""
426	n/a	try:
427	n/a	if not self._cleaning:
428	n/a	self.tp_del_calls.append(id(self))
429	n/a	self.check_sanity()
430	n/a	self.side_effect()
431	n/a	except Exception as e:
432	n/a	self.errors.append(e)
433	n/a
434	n/a	@with_tp_del
435	n/a	class Legacy(LegacyBase):
436	n/a	pass
437	n/a
438	n/a	@with_tp_del
439	n/a	class LegacyResurrector(LegacyBase):
440	n/a
441	n/a	def side_effect(self):
442	n/a	"""
443	n/a	Resurrect self by storing self in a class-wide list.
444	n/a	"""
445	n/a	self.survivors.append(self)
446	n/a
447	n/a	@with_tp_del
448	n/a	class LegacySelfCycle(SelfCycleBase, LegacyBase):
449	n/a	pass
450	n/a
451	n/a
452	n/a	@support.cpython_only
453	n/a	class LegacyFinalizationTest(TestBase, unittest.TestCase):
454	n/a	"""
455	n/a	Test finalization of objects with a tp_del.
456	n/a	"""
457	n/a
458	n/a	def tearDown(self):
459	n/a	# These tests need to clean up a bit more, since they create
460	n/a	# uncollectable objects.
461	n/a	gc.garbage.clear()
462	n/a	gc.collect()
463	n/a	super().tearDown()
464	n/a
465	n/a	def test_legacy(self):
466	n/a	with SimpleBase.test():
467	n/a	s = Legacy()
468	n/a	ids = [id(s)]
469	n/a	wr = weakref.ref(s)
470	n/a	del s
471	n/a	gc.collect()
472	n/a	self.assert_del_calls(ids)
473	n/a	self.assert_tp_del_calls(ids)
474	n/a	self.assert_survivors([])
475	n/a	self.assertIs(wr(), None)
476	n/a	gc.collect()
477	n/a	self.assert_del_calls(ids)
478	n/a	self.assert_tp_del_calls(ids)
479	n/a
480	n/a	def test_legacy_resurrect(self):
481	n/a	with SimpleBase.test():
482	n/a	s = LegacyResurrector()
483	n/a	ids = [id(s)]
484	n/a	wr = weakref.ref(s)
485	n/a	del s
486	n/a	gc.collect()
487	n/a	self.assert_del_calls(ids)
488	n/a	self.assert_tp_del_calls(ids)
489	n/a	self.assert_survivors(ids)
490	n/a	# weakrefs are cleared before tp_del is called.
491	n/a	self.assertIs(wr(), None)
492	n/a	self.clear_survivors()
493	n/a	gc.collect()
494	n/a	self.assert_del_calls(ids)
495	n/a	self.assert_tp_del_calls(ids * 2)
496	n/a	self.assert_survivors(ids)
497	n/a	self.assertIs(wr(), None)
498	n/a
499	n/a	def test_legacy_self_cycle(self):
500	n/a	# Self-cycles with legacy finalizers end up in gc.garbage.
501	n/a	with SimpleBase.test():
502	n/a	s = LegacySelfCycle()
503	n/a	ids = [id(s)]
504	n/a	wr = weakref.ref(s)
505	n/a	del s
506	n/a	gc.collect()
507	n/a	self.assert_del_calls([])
508	n/a	self.assert_tp_del_calls([])
509	n/a	self.assert_survivors([])
510	n/a	self.assert_garbage(ids)
511	n/a	self.assertIsNot(wr(), None)
512	n/a	# Break the cycle to allow collection
513	n/a	gc.garbage[0].ref = None
514	n/a	self.assert_garbage([])
515	n/a	self.assertIs(wr(), None)
516	n/a
517	n/a
518	n/a	if __name__ == "__main__":
519	n/a	unittest.main()