""" Tests for object finalization semantics, as outlined in PEP 442. """ import contextlib import gc import unittest import weakref try: from _testcapi import with_tp_del except ImportError: def with_tp_del(cls): class C(object): def __new__(cls, *args, **kwargs): raise TypeError('requires _testcapi.with_tp_del') return C try: from _testcapi import without_gc except ImportError: def without_gc(cls): class C: def __new__(cls, *args, **kwargs): raise TypeError('requires _testcapi.without_gc') return C from test import support class NonGCSimpleBase: """ The base class for all the objects under test, equipped with various testing features. """ survivors = [] del_calls = [] tp_del_calls = [] errors = [] _cleaning = False __slots__ = () @classmethod def _cleanup(cls): cls.survivors.clear() cls.errors.clear() gc.garbage.clear() gc.collect() cls.del_calls.clear() cls.tp_del_calls.clear() @classmethod @contextlib.contextmanager def test(cls): """ A context manager to use around all finalization tests. """ with support.disable_gc(): cls.del_calls.clear() cls.tp_del_calls.clear() NonGCSimpleBase._cleaning = False try: yield if cls.errors: raise cls.errors[0] finally: NonGCSimpleBase._cleaning = True cls._cleanup() def check_sanity(self): """ Check the object is sane (non-broken). """ def __del__(self): """ PEP 442 finalizer. Record that this was called, check the object is in a sane state, and invoke a side effect. """ try: if not self._cleaning: self.del_calls.append(id(self)) self.check_sanity() self.side_effect() except Exception as e: self.errors.append(e) def side_effect(self): """ A side effect called on destruction. """ class SimpleBase(NonGCSimpleBase): def __init__(self): self.id_ = id(self) def check_sanity(self): assert self.id_ == id(self) @without_gc class NonGC(NonGCSimpleBase): __slots__ = () @without_gc class NonGCResurrector(NonGCSimpleBase): __slots__ = () def side_effect(self): """ Resurrect self by storing self in a class-wide list. """ self.survivors.append(self) class Simple(SimpleBase): pass # Can't inherit from NonGCResurrector, in case importing without_gc fails. class SimpleResurrector(SimpleBase): def side_effect(self): """ Resurrect self by storing self in a class-wide list. """ self.survivors.append(self) class TestBase: def setUp(self): self.old_garbage = gc.garbage[:] gc.garbage[:] = [] def tearDown(self): # None of the tests here should put anything in gc.garbage try: self.assertEqual(gc.garbage, []) finally: del self.old_garbage gc.collect() def assert_del_calls(self, ids): self.assertEqual(sorted(SimpleBase.del_calls), sorted(ids)) def assert_tp_del_calls(self, ids): self.assertEqual(sorted(SimpleBase.tp_del_calls), sorted(ids)) def assert_survivors(self, ids): self.assertEqual(sorted(id(x) for x in SimpleBase.survivors), sorted(ids)) def assert_garbage(self, ids): self.assertEqual(sorted(id(x) for x in gc.garbage), sorted(ids)) def clear_survivors(self): SimpleBase.survivors.clear() class SimpleFinalizationTest(TestBase, unittest.TestCase): """ Test finalization without refcycles. """ def test_simple(self): with SimpleBase.test(): s = Simple() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) self.assertIs(wr(), None) gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) def test_simple_resurrect(self): with SimpleBase.test(): s = SimpleResurrector() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls(ids) self.assert_survivors(ids) self.assertIsNot(wr(), None) self.clear_survivors() gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) self.assertIs(wr(), None) @support.cpython_only def test_non_gc(self): with SimpleBase.test(): s = NonGC() self.assertFalse(gc.is_tracked(s)) ids = [id(s)] del s gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) @support.cpython_only def test_non_gc_resurrect(self): with SimpleBase.test(): s = NonGCResurrector() self.assertFalse(gc.is_tracked(s)) ids = [id(s)] del s gc.collect() self.assert_del_calls(ids) self.assert_survivors(ids) self.clear_survivors() gc.collect() self.assert_del_calls(ids * 2) self.assert_survivors(ids) class SelfCycleBase: def __init__(self): super().__init__() self.ref = self def check_sanity(self): super().check_sanity() assert self.ref is self class SimpleSelfCycle(SelfCycleBase, Simple): pass class SelfCycleResurrector(SelfCycleBase, SimpleResurrector): pass class SuicidalSelfCycle(SelfCycleBase, Simple): def side_effect(self): """ Explicitly break the reference cycle. """ self.ref = None class SelfCycleFinalizationTest(TestBase, unittest.TestCase): """ Test finalization of an object having a single cyclic reference to itself. """ def test_simple(self): with SimpleBase.test(): s = SimpleSelfCycle() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) self.assertIs(wr(), None) gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) def test_simple_resurrect(self): # Test that __del__ can resurrect the object being finalized. with SimpleBase.test(): s = SelfCycleResurrector() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls(ids) self.assert_survivors(ids) # XXX is this desirable? self.assertIs(wr(), None) # When trying to destroy the object a second time, __del__ # isn't called anymore (and the object isn't resurrected). self.clear_survivors() gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) self.assertIs(wr(), None) def test_simple_suicide(self): # Test the GC is able to deal with an object that kills its last # reference during __del__. with SimpleBase.test(): s = SuicidalSelfCycle() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) self.assertIs(wr(), None) gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) self.assertIs(wr(), None) class ChainedBase: def chain(self, left): self.suicided = False self.left = left left.right = self def check_sanity(self): super().check_sanity() if self.suicided: assert self.left is None assert self.right is None else: left = self.left if left.suicided: assert left.right is None else: assert left.right is self right = self.right if right.suicided: assert right.left is None else: assert right.left is self class SimpleChained(ChainedBase, Simple): pass class ChainedResurrector(ChainedBase, SimpleResurrector): pass class SuicidalChained(ChainedBase, Simple): def side_effect(self): """ Explicitly break the reference cycle. """ self.suicided = True self.left = None self.right = None class CycleChainFinalizationTest(TestBase, unittest.TestCase): """ Test finalization of a cyclic chain. These tests are similar in spirit to the self-cycle tests above, but the collectable object graph isn't trivial anymore. """ def build_chain(self, classes): nodes = [cls() for cls in classes] for i in range(len(nodes)): nodes[i].chain(nodes[i-1]) return nodes def check_non_resurrecting_chain(self, classes): N = len(classes) with SimpleBase.test(): nodes = self.build_chain(classes) ids = [id(s) for s in nodes] wrs = [weakref.ref(s) for s in nodes] del nodes gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) self.assertEqual([wr() for wr in wrs], [None] * N) gc.collect() self.assert_del_calls(ids) def check_resurrecting_chain(self, classes): N = len(classes) with SimpleBase.test(): nodes = self.build_chain(classes) N = len(nodes) ids = [id(s) for s in nodes] survivor_ids = [id(s) for s in nodes if isinstance(s, SimpleResurrector)] wrs = [weakref.ref(s) for s in nodes] del nodes gc.collect() self.assert_del_calls(ids) self.assert_survivors(survivor_ids) # XXX desirable? self.assertEqual([wr() for wr in wrs], [None] * N) self.clear_survivors() gc.collect() self.assert_del_calls(ids) self.assert_survivors([]) def test_homogenous(self): self.check_non_resurrecting_chain([SimpleChained] * 3) def test_homogenous_resurrect(self): self.check_resurrecting_chain([ChainedResurrector] * 3) def test_homogenous_suicidal(self): self.check_non_resurrecting_chain([SuicidalChained] * 3) def test_heterogenous_suicidal_one(self): self.check_non_resurrecting_chain([SuicidalChained, SimpleChained] * 2) def test_heterogenous_suicidal_two(self): self.check_non_resurrecting_chain( [SuicidalChained] * 2 + [SimpleChained] * 2) def test_heterogenous_resurrect_one(self): self.check_resurrecting_chain([ChainedResurrector, SimpleChained] * 2) def test_heterogenous_resurrect_two(self): self.check_resurrecting_chain( [ChainedResurrector, SimpleChained, SuicidalChained] * 2) def test_heterogenous_resurrect_three(self): self.check_resurrecting_chain( [ChainedResurrector] * 2 + [SimpleChained] * 2 + [SuicidalChained] * 2) # NOTE: the tp_del slot isn't automatically inherited, so we have to call # with_tp_del() for each instantiated class. class LegacyBase(SimpleBase): def __del__(self): try: # Do not invoke side_effect here, since we are now exercising # the tp_del slot. if not self._cleaning: self.del_calls.append(id(self)) self.check_sanity() except Exception as e: self.errors.append(e) def __tp_del__(self): """ Legacy (pre-PEP 442) finalizer, mapped to a tp_del slot. """ try: if not self._cleaning: self.tp_del_calls.append(id(self)) self.check_sanity() self.side_effect() except Exception as e: self.errors.append(e) @with_tp_del class Legacy(LegacyBase): pass @with_tp_del class LegacyResurrector(LegacyBase): def side_effect(self): """ Resurrect self by storing self in a class-wide list. """ self.survivors.append(self) @with_tp_del class LegacySelfCycle(SelfCycleBase, LegacyBase): pass @support.cpython_only class LegacyFinalizationTest(TestBase, unittest.TestCase): """ Test finalization of objects with a tp_del. """ def tearDown(self): # These tests need to clean up a bit more, since they create # uncollectable objects. gc.garbage.clear() gc.collect() super().tearDown() def test_legacy(self): with SimpleBase.test(): s = Legacy() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls(ids) self.assert_tp_del_calls(ids) self.assert_survivors([]) self.assertIs(wr(), None) gc.collect() self.assert_del_calls(ids) self.assert_tp_del_calls(ids) def test_legacy_resurrect(self): with SimpleBase.test(): s = LegacyResurrector() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls(ids) self.assert_tp_del_calls(ids) self.assert_survivors(ids) # weakrefs are cleared before tp_del is called. self.assertIs(wr(), None) self.clear_survivors() gc.collect() self.assert_del_calls(ids) self.assert_tp_del_calls(ids * 2) self.assert_survivors(ids) self.assertIs(wr(), None) def test_legacy_self_cycle(self): # Self-cycles with legacy finalizers end up in gc.garbage. with SimpleBase.test(): s = LegacySelfCycle() ids = [id(s)] wr = weakref.ref(s) del s gc.collect() self.assert_del_calls([]) self.assert_tp_del_calls([]) self.assert_survivors([]) self.assert_garbage(ids) self.assertIsNot(wr(), None) # Break the cycle to allow collection gc.garbage[0].ref = None self.assert_garbage([]) self.assertIs(wr(), None) if __name__ == "__main__": unittest.main() SIMPAN PERUBAHAN