import gc
import sys
import time

import blinker

from nose.tools import assert_raises


jython = sys.platform.startswith('java')
pypy = hasattr(sys, 'pypy_version_info')


def collect_acyclic_refs():
    # cpython releases these immediately without a collection
    if jython or pypy:
        gc.collect()
    if jython:
        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):
        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_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_acyclic_refs()

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

    # 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')

    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_weak_namespace():
    ns = blinker.WeakNamespace()
    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_acyclic_refs()

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

    assert 'abc' not in ns


def test_namespace():
    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

    del s1
    collect_acyclic_refs()

    assert 'def' in ns
    assert 'abc' 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_acyclic_refs()

    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_acyclic_refs()

    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_acyclic_refs()

    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_acyclic_refs()

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

    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.connected_to, receiver)