<gh_stars>0
import blinker
from nose.tools import assert_raises


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
    # weak by default, already out of scope
    assert 'def' not in ns
    del s1
    assert 'abc' not in ns


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

    sig = blinker.Signal()
    sig.connect(received, weak=True)
    del received

    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

    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
    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

    # 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_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
    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()

import blinker
from nose.tools import assert_raises


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
    # weak by default, already out of scope
    assert 'def' not in ns
    del s1
    assert 'abc' not in ns


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

    sig = blinker.Signal()
    sig.connect(received, weak=True)
    del received

    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

    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
    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

    # 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_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
    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()
# weak by default, already out of scope # general index isn't cleaned up # but receiver/sender pairs are