mirror of
https://github.com/django/django.git
synced 2025-05-17 20:36:28 +00:00
760121dcb1
As documented, the id() function can return the same value for distinct objects with non-overlapping lifetimes which can result in signals being sent to the wrong receivers if two distinct senders happen to have a colliding id() value. Since reproduction of the issue requires memory constrained circumstances where the same exact id() is reused for two senders of the same signal the test opt to simulate the collision by systematically making the same id for Sender instances. Note that we explicitly avoid keeping a strong reference to senders that cannot be weakly referenced as that would unexpectedly prevent them from being garbage collected. This means that id(sender) collisions could still occur for such objects but Django itself doesn't make use of them. Thanks Sjoerd Job Postmus for the reduced test case and Mariusz for the review. Co-authored-by: And Clover <and@doxdesk.com>
285 lines
9.6 KiB
Python
285 lines
9.6 KiB
Python
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
|
|
|
|
|
|
def receiver_1_arg(val, **kwargs):
|
|
return val
|
|
|
|
|
|
class Callable:
|
|
def __call__(self, val, **kwargs):
|
|
return val
|
|
|
|
def a(self, val, **kwargs):
|
|
return val
|
|
|
|
|
|
a_signal = Signal()
|
|
b_signal = Signal()
|
|
c_signal = Signal()
|
|
d_signal = Signal(use_caching=True)
|
|
|
|
|
|
class DispatcherTests(SimpleTestCase):
|
|
def assertTestIsClean(self, signal):
|
|
"""Assert that everything has been cleaned up automatically"""
|
|
# Note that dead weakref cleanup happens as side effect of using
|
|
# the signal's receivers through the signals API. So, first do a
|
|
# call to an API method to force cleanup.
|
|
self.assertFalse(signal.has_listeners())
|
|
self.assertEqual(signal.receivers, [])
|
|
|
|
@override_settings(DEBUG=True)
|
|
def test_cannot_connect_no_kwargs(self):
|
|
def receiver_no_kwargs(sender):
|
|
pass
|
|
|
|
msg = "Signal receivers must accept keyword arguments (**kwargs)."
|
|
with self.assertRaisesMessage(ValueError, msg):
|
|
a_signal.connect(receiver_no_kwargs)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
@override_settings(DEBUG=True)
|
|
def test_cannot_connect_non_callable(self):
|
|
msg = "Signal receivers must be callable."
|
|
with self.assertRaisesMessage(TypeError, msg):
|
|
a_signal.connect(object())
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_send(self):
|
|
a_signal.connect(receiver_1_arg, sender=self)
|
|
result = a_signal.send(sender=self, val="test")
|
|
self.assertEqual(result, [(receiver_1_arg, "test")])
|
|
a_signal.disconnect(receiver_1_arg, sender=self)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_send_no_receivers(self):
|
|
result = a_signal.send(sender=self, val="test")
|
|
self.assertEqual(result, [])
|
|
|
|
def test_send_connected_no_sender(self):
|
|
a_signal.connect(receiver_1_arg)
|
|
result = a_signal.send(sender=self, val="test")
|
|
self.assertEqual(result, [(receiver_1_arg, "test")])
|
|
a_signal.disconnect(receiver_1_arg)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_send_different_no_sender(self):
|
|
a_signal.connect(receiver_1_arg, sender=object)
|
|
result = a_signal.send(sender=self, val="test")
|
|
self.assertEqual(result, [])
|
|
a_signal.disconnect(receiver_1_arg, sender=object)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
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
|
|
garbage_collect()
|
|
result = a_signal.send(sender=self, val="test")
|
|
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
|
|
collection of senders.
|
|
"""
|
|
|
|
class sender:
|
|
pass
|
|
|
|
wref = weakref.ref(sender)
|
|
d_signal.connect(receiver_1_arg)
|
|
d_signal.send(sender, val="garbage")
|
|
del sender
|
|
garbage_collect()
|
|
try:
|
|
self.assertIsNone(wref())
|
|
finally:
|
|
# Disconnect after reference check since it flushes the tested cache.
|
|
d_signal.disconnect(receiver_1_arg)
|
|
|
|
def test_multiple_registration(self):
|
|
a = Callable()
|
|
a_signal.connect(a)
|
|
a_signal.connect(a)
|
|
a_signal.connect(a)
|
|
a_signal.connect(a)
|
|
a_signal.connect(a)
|
|
a_signal.connect(a)
|
|
result = a_signal.send(sender=self, val="test")
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(len(a_signal.receivers), 1)
|
|
del a
|
|
del result
|
|
garbage_collect()
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_uid_registration(self):
|
|
def uid_based_receiver_1(**kwargs):
|
|
pass
|
|
|
|
def uid_based_receiver_2(**kwargs):
|
|
pass
|
|
|
|
a_signal.connect(uid_based_receiver_1, dispatch_uid="uid")
|
|
a_signal.connect(uid_based_receiver_2, dispatch_uid="uid")
|
|
self.assertEqual(len(a_signal.receivers), 1)
|
|
a_signal.disconnect(dispatch_uid="uid")
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_send_robust_success(self):
|
|
a_signal.connect(receiver_1_arg)
|
|
result = a_signal.send_robust(sender=self, val="test")
|
|
self.assertEqual(result, [(receiver_1_arg, "test")])
|
|
a_signal.disconnect(receiver_1_arg)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_send_robust_no_receivers(self):
|
|
result = a_signal.send_robust(sender=self, val="test")
|
|
self.assertEqual(result, [])
|
|
|
|
def test_send_robust_ignored_sender(self):
|
|
a_signal.connect(receiver_1_arg)
|
|
result = a_signal.send_robust(sender=self, val="test")
|
|
self.assertEqual(result, [(receiver_1_arg, "test")])
|
|
a_signal.disconnect(receiver_1_arg)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_send_robust_fail(self):
|
|
def fails(val, **kwargs):
|
|
raise ValueError("this")
|
|
|
|
a_signal.connect(fails)
|
|
try:
|
|
with self.assertLogs("django.dispatch", "ERROR") as cm:
|
|
result = a_signal.send_robust(sender=self, val="test")
|
|
err = result[0][1]
|
|
self.assertIsInstance(err, ValueError)
|
|
self.assertEqual(err.args, ("this",))
|
|
self.assertIs(hasattr(err, "__traceback__"), True)
|
|
self.assertIsInstance(err.__traceback__, TracebackType)
|
|
|
|
log_record = cm.records[0]
|
|
self.assertEqual(
|
|
log_record.getMessage(),
|
|
"Error calling "
|
|
"DispatcherTests.test_send_robust_fail.<locals>.fails in "
|
|
"Signal.send_robust() (this)",
|
|
)
|
|
self.assertIsNotNone(log_record.exc_info)
|
|
_, exc_value, _ = log_record.exc_info
|
|
self.assertIsInstance(exc_value, ValueError)
|
|
self.assertEqual(str(exc_value), "this")
|
|
finally:
|
|
a_signal.disconnect(fails)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_disconnection(self):
|
|
receiver_1 = Callable()
|
|
receiver_2 = Callable()
|
|
receiver_3 = Callable()
|
|
a_signal.connect(receiver_1)
|
|
a_signal.connect(receiver_2)
|
|
a_signal.connect(receiver_3)
|
|
a_signal.disconnect(receiver_1)
|
|
del receiver_2
|
|
garbage_collect()
|
|
a_signal.disconnect(receiver_3)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_values_returned_by_disconnection(self):
|
|
receiver_1 = Callable()
|
|
receiver_2 = Callable()
|
|
a_signal.connect(receiver_1)
|
|
receiver_1_disconnected = a_signal.disconnect(receiver_1)
|
|
receiver_2_disconnected = a_signal.disconnect(receiver_2)
|
|
self.assertTrue(receiver_1_disconnected)
|
|
self.assertFalse(receiver_2_disconnected)
|
|
self.assertTestIsClean(a_signal)
|
|
|
|
def test_has_listeners(self):
|
|
self.assertFalse(a_signal.has_listeners())
|
|
self.assertFalse(a_signal.has_listeners(sender=object()))
|
|
receiver_1 = Callable()
|
|
a_signal.connect(receiver_1)
|
|
self.assertTrue(a_signal.has_listeners())
|
|
self.assertTrue(a_signal.has_listeners(sender=object()))
|
|
a_signal.disconnect(receiver_1)
|
|
self.assertFalse(a_signal.has_listeners())
|
|
self.assertFalse(a_signal.has_listeners(sender=object()))
|
|
|
|
|
|
class ReceiverTestCase(SimpleTestCase):
|
|
def test_receiver_single_signal(self):
|
|
@receiver(a_signal)
|
|
def f(val, **kwargs):
|
|
self.state = val
|
|
|
|
self.state = False
|
|
a_signal.send(sender=self, val=True)
|
|
self.assertTrue(self.state)
|
|
|
|
def test_receiver_signal_list(self):
|
|
@receiver([a_signal, b_signal, c_signal])
|
|
def f(val, **kwargs):
|
|
self.state.append(val)
|
|
|
|
self.state = []
|
|
a_signal.send(sender=self, val="a")
|
|
c_signal.send(sender=self, val="c")
|
|
b_signal.send(sender=self, val="b")
|
|
self.assertIn("a", self.state)
|
|
self.assertIn("b", self.state)
|
|
self.assertIn("c", self.state)
|