path: root/tests/test_signals.py

import gc
import sys
import time

import blinker

from nose.tools import assert_raises


jython = sys.platform.startswith('java')


def collect():
    if jython:
        gc.collect()
        time.sleep(0.1)


def test_meta_connect():
    sentinel = []
    def meta_received(sender, **kw):
        sentinel.append(dict(kw, sender=sender))

    assert not blinker.receiver_connected.receivers
    blinker.receiver_connected.connect(meta_received)
    assert not sentinel

    def receiver(sender, **kw):
        pass
    sig = blinker.Signal()
    sig.connect(receiver)

    assert sentinel == [dict(sender=sig,
                             receiver_arg=receiver,
                             sender_arg=blinker.ANY,
                             weak_arg=True)]

    blinker.receiver_connected._clear_state()


def test_meta_connect_failure():
    def meta_received(sender, **kw):
        raise TypeError('boom')

    assert not blinker.receiver_connected.receivers
    blinker.receiver_connected.connect(meta_received)

    def receiver(sender, **kw):
        pass
    sig = blinker.Signal()

    assert_raises(TypeError, sig.connect, receiver)
    assert not sig.receivers
    assert not sig._by_receiver
    assert sig._by_sender == {blinker.base.ANY_ID: set()}

    blinker.receiver_connected._clear_state()


def test_singletons():
    ns = blinker.Namespace()
    assert not ns
    s1 = ns.signal('abc')
    assert s1 is ns.signal('abc')
    assert s1 is not ns.signal('def')
    assert 'abc' in ns
    collect()

    # weak by default, already out of scope
    assert 'def' not in ns
    del s1
    collect()

    assert 'abc' not in ns


def test_weak_receiver():
    sentinel = []
    def received(sender, **kw):
        sentinel.append(kw)

    sig = blinker.Signal()

    # XXX: weirdly, under jython an explicit weak=True causes this test
    #      to fail, leaking a strong ref to the receiver somewhere.
    #      http://bugs.jython.org/issue1586
    if jython:
        sig.connect(received)  # weak=True by default.
    else:
        sig.connect(received, weak=True)

    del received
    collect()

    assert not sentinel
    sig.send()
    assert not sentinel
    assert not sig.receivers
    values_are_empty_sets_(sig._by_receiver)
    values_are_empty_sets_(sig._by_sender)


def test_strong_receiver():
    sentinel = []
    def received(sender):
        sentinel.append(sender)
    fn_id = id(received)

    sig = blinker.Signal()
    sig.connect(received, weak=False)

    del received
    collect()

    assert not sentinel
    sig.send()
    assert sentinel
    assert [id(fn) for fn in sig.receivers.values()] == [fn_id]


def test_instancemethod_receiver():
    sentinel = []

    class Receiver(object):
        def __init__(self, bucket):
            self.bucket = bucket
        def received(self, sender):
            self.bucket.append(sender)

    receiver = Receiver(sentinel)

    sig = blinker.Signal()
    sig.connect(receiver.received)

    assert not sentinel
    sig.send()
    assert sentinel
    del receiver
    collect()

    sig.send()
    assert len(sentinel) == 1


def test_filtered_receiver():
    sentinel = []
    def received(sender):
        sentinel.append(sender)

    sig = blinker.Signal()

    sig.connect(received, 123)

    assert not sentinel
    sig.send()
    assert not sentinel
    sig.send(123)
    assert sentinel == [123]
    sig.send()
    assert sentinel == [123]

    sig.disconnect(received, 123)
    sig.send(123)
    assert sentinel == [123]

    sig.connect(received, 123)
    sig.send(123)
    assert sentinel == [123, 123]

    sig.disconnect(received)
    sig.send(123)
    assert sentinel == [123, 123]


def test_filtered_receiver_weakref():
    sentinel = []
    def received(sender):
        sentinel.append(sender)

    class Object(object):
        pass
    obj = Object()

    sig = blinker.Signal()
    sig.connect(received, obj)

    assert not sentinel
    sig.send(obj)
    assert sentinel == [obj]
    del sentinel[:]
    del obj
    collect()

    # general index isn't cleaned up
    assert sig.receivers
    # but receiver/sender pairs are
    values_are_empty_sets_(sig._by_receiver)
    values_are_empty_sets_(sig._by_sender)


def test_decorated_receiver():
    sentinel = []

    class Object(object):
        pass
    obj = Object()

    sig = blinker.Signal()

    @sig.connect_via(obj)
    def receiver(sender, **kw):
        sentinel.append(kw)

    assert not sentinel
    sig.send()
    assert not sentinel
    sig.send(1)
    assert not sentinel
    sig.send(obj)
    assert sig.receivers

    del receiver
    collect()
    assert sig.receivers


def test_no_double_send():
    sentinel = []
    def received(sender):
        sentinel.append(sender)

    sig = blinker.Signal()

    sig.connect(received, 123)
    sig.connect(received)

    assert not sentinel
    sig.send()
    assert sentinel == [None]
    sig.send(123)
    assert sentinel == [None, 123]
    sig.send()
    assert sentinel == [None, 123, None]


def test_has_receivers():
    received = lambda sender: None

    sig = blinker.Signal()
    assert not sig.has_receivers_for(None)
    assert not sig.has_receivers_for(blinker.ANY)

    sig.connect(received, 'xyz')
    assert not sig.has_receivers_for(None)
    assert not sig.has_receivers_for(blinker.ANY)
    assert sig.has_receivers_for('xyz')

    class Object(object):
        pass
    o = Object()

    sig.connect(received, o)
    assert sig.has_receivers_for(o)

    del received
    collect()

    assert not sig.has_receivers_for('xyz')
    assert list(sig.receivers_for('xyz')) == []
    assert list(sig.receivers_for(o)) == []

    sig.connect(lambda sender: None, weak=False)
    assert sig.has_receivers_for('xyz')
    assert sig.has_receivers_for(o)
    assert sig.has_receivers_for(None)
    assert sig.has_receivers_for(blinker.ANY)
    assert sig.has_receivers_for('xyz')


def test_instance_doc():
    sig = blinker.Signal(doc='x')
    assert sig.__doc__ == 'x'

    sig = blinker.Signal('x')
    assert sig.__doc__ == 'x'


def test_named_blinker():
    sig = blinker.NamedSignal('squiznart')
    assert 'squiznart' in repr(sig)


def values_are_empty_sets_(dictionary):
    for val in dictionary.values():
        assert val == set()

if sys.version_info < (2, 5):
    def test_context_manager_warning():
        sig = blinker.Signal()
        receiver = lambda sender: None

        assert_raises(RuntimeError, sig.temporarily_connected_to, receiver)