Added Signal.receiver_connected and .receiver_disconnected per-Signal signals.

Deprecates the global 'receiver_connected' signal.

1 files changed, 139 insertions, 0 deletions

diff --git a/tests/test_signals.py b/tests/test_signals.py
index 9a5a5ce..22bf397 100644
--- a/tests/test_signals.py
+++ b/tests/test_signals.py
@@ -16,6 +16,21 @@ def collect():
         time.sleep(0.1)
 
 
+class Sentinel(list):
+    """A signal receipt accumulator."""
+
+    def make_receiver(self, key):
+        """Return a generic signal receiver function logging as *key*
+
+        When connected to a signal, appends (key, sender, kw) to the Sentinel.
+
+        """
+        def receiver(*sentby, **kw):
+            self.append((key, sentby[0], kw))
+        receiver.func_name = 'receiver_%s' % key
+        return receiver
+
+
 def test_meta_connect():
     sentinel = []
     def meta_received(sender, **kw):
@@ -38,6 +53,130 @@ def test_meta_connect():
     blinker.receiver_connected._clear_state()
 
 
+def _test_signal_signals(sender):
+    sentinel = Sentinel()
+    sig = blinker.Signal()
+
+    connected = sentinel.make_receiver('receiver_connected')
+    disconnected = sentinel.make_receiver('receiver_disconnected')
+    receiver1 = sentinel.make_receiver('receiver1')
+    receiver2 = sentinel.make_receiver('receiver2')
+
+    assert not sig.receiver_connected.receivers
+    assert not sig.receiver_disconnected.receivers
+    sig.receiver_connected.connect(connected)
+    sig.receiver_disconnected.connect(disconnected)
+
+    assert sig.receiver_connected.receivers
+    assert not sentinel
+
+    for receiver, weak in [(receiver1, True), (receiver2, False)]:
+        sig.connect(receiver, sender=sender, weak=weak)
+
+        expected = ('receiver_connected',
+                    sig,
+                    dict(receiver=receiver, sender=sender, weak=weak))
+
+        assert sentinel[-1] == expected
+
+    # disconnect from explicit sender
+    sig.disconnect(receiver1, sender=sender)
+
+    expected = ('receiver_disconnected',
+                sig,
+                dict(receiver=receiver1, sender=sender))
+    assert sentinel[-1] == expected
+
+    # disconnect from ANY and all senders (implicit disconnect signature)
+    sig.disconnect(receiver2)
+    assert sentinel[-1] == ('receiver_disconnected',
+                            sig,
+                            dict(receiver=receiver2, sender=blinker.ANY))
+
+
+def test_signal_signals_any_sender():
+    _test_signal_signals(blinker.ANY)
+
+
+def test_signal_signals_strong_sender():
+    _test_signal_signals("squiznart")
+
+
+def test_signal_weak_receiver_vanishes():
+    # non-edge-case path for weak receivers is exercised in the ANY sender
+    # test above.
+    sentinel = Sentinel()
+    sig = blinker.Signal()
+
+    connected = sentinel.make_receiver('receiver_connected')
+    disconnected = sentinel.make_receiver('receiver_disconnected')
+    receiver1 = sentinel.make_receiver('receiver1')
+    receiver2 = sentinel.make_receiver('receiver2')
+
+    sig.receiver_connected.connect(connected)
+    sig.receiver_disconnected.connect(disconnected)
+
+    # explicit disconnect on a weak does emit the signal
+    sig.connect(receiver1, weak=True)
+    sig.disconnect(receiver1)
+
+    assert len(sentinel) == 2
+    assert sentinel[-1][2]['receiver'] is receiver1
+
+    del sentinel[:]
+    sig.connect(receiver2, weak=True)
+    assert len(sentinel) == 1
+
+    del sentinel[:]  # holds a ref to receiver2
+    del receiver2
+    collect()
+
+    # no disconnect signal is fired
+    assert len(sentinel) == 0
+
+    # and everything really is disconnected
+    sig.send('abc')
+    assert len(sentinel) == 0
+
+
+def test_signal_signals_weak_sender():
+    sentinel = Sentinel()
+    sig = blinker.Signal()
+
+    connected = sentinel.make_receiver('receiver_connected')
+    disconnected = sentinel.make_receiver('receiver_disconnected')
+    receiver1 = sentinel.make_receiver('receiver1')
+    receiver2 = sentinel.make_receiver('receiver2')
+
+    class Sender(object):
+        """A weakref-able object."""
+
+    sig.receiver_connected.connect(connected)
+    sig.receiver_disconnected.connect(disconnected)
+
+    sender1 = Sender()
+    sig.connect(receiver1, sender=sender1, weak=False)
+    # regular disconnect of weak-able sender works fine
+    sig.disconnect(receiver1, sender=sender1)
+
+    assert len(sentinel) == 2
+
+    del sentinel[:]
+    sender2 = Sender()
+    sig.connect(receiver2, sender=sender2, weak=False)
+
+    # force sender2 to go out of scope
+    del sender2
+    collect()
+
+    # no disconnect signal is fired
+    assert len(sentinel) == 1
+
+    # and everything really is disconnected
+    sig.send('abc')
+    assert len(sentinel) == 1
+
+
 def test_meta_connect_failure():
     def meta_received(sender, **kw):
         raise TypeError('boom')