summaryrefslogtreecommitdiff
path: root/tests/dispatch
diff options
context:
space:
mode:
Diffstat (limited to 'tests/dispatch')
-rw-r--r--tests/dispatch/tests.py47
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