diff options
Diffstat (limited to 'tests/dispatch')
| -rw-r--r-- | tests/dispatch/tests.py | 47 |
1 files changed, 46 insertions, 1 deletions
diff --git a/tests/dispatch/tests.py b/tests/dispatch/tests.py index 18426d8dd1..e91d29abdd 100644 --- a/tests/dispatch/tests.py +++ b/tests/dispatch/tests.py @@ -1,7 +1,9 @@ import weakref from types import TracebackType +from unittest import mock from django.dispatch import Signal, receiver +from django.dispatch.dispatcher import _make_id from django.test import SimpleTestCase from django.test.utils import garbage_collect, override_settings @@ -75,7 +77,15 @@ class DispatcherTests(SimpleTestCase): a_signal.disconnect(receiver_1_arg, sender=object) self.assertTestIsClean(a_signal) - def test_garbage_collected(self): + def test_unweakrefable_sender(self): + sender = object() + a_signal.connect(receiver_1_arg, sender=sender) + result = a_signal.send(sender=sender, val="test") + self.assertEqual(result, [(receiver_1_arg, "test")]) + a_signal.disconnect(receiver_1_arg, sender=sender) + self.assertTestIsClean(a_signal) + + def test_garbage_collected_receiver(self): a = Callable() a_signal.connect(a.a, sender=self) del a @@ -84,6 +94,41 @@ class DispatcherTests(SimpleTestCase): self.assertEqual(result, []) self.assertTestIsClean(a_signal) + def test_garbage_collected_sender(self): + signal = Signal() + + class Sender: + pass + + def make_id(target): + """ + Simulate id() reuse for distinct senders with non-overlapping + lifetimes that would require memory contention to reproduce. + """ + if isinstance(target, Sender): + return 0 + return _make_id(target) + + def first_receiver(attempt, **kwargs): + return attempt + + def second_receiver(attempt, **kwargs): + return attempt + + with mock.patch("django.dispatch.dispatcher._make_id", make_id): + sender = Sender() + signal.connect(first_receiver, sender) + result = signal.send(sender, attempt="first") + self.assertEqual(result, [(first_receiver, "first")]) + + del sender + garbage_collect() + + sender = Sender() + signal.connect(second_receiver, sender) + result = signal.send(sender, attempt="second") + self.assertEqual(result, [(second_receiver, "second")]) + def test_cached_garbaged_collected(self): """ Make sure signal caching sender receivers don't prevent garbage |
