Initial import from flatland.util.signals

3 files changed, 376 insertions, 0 deletions

diff --git a/tests/test_saferef.py b/tests/test_saferef.py
new file mode 100644
index 0000000..3886a9e
--- /dev/null
+++ b/tests/test_saferef.py
@@ -0,0 +1,117 @@
+# extracted from Louie, http://pylouie.org/
+# updated for Python 3
+#
+# Copyright (c) 2006 Patrick K. O'Brien, Mike C. Fletcher,
+#                    Matthew R. Scott
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+#       notice, this list of conditions and the following disclaimer.
+#
+#     * Redistributions in binary form must reproduce the above
+#       copyright notice, this list of conditions and the following
+#       disclaimer in the documentation and/or other materials provided
+#       with the distribution.
+#
+#     * Neither the name of the <ORGANIZATION> nor the names of its
+#       contributors may be used to endorse or promote products derived
+#       from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+#
+import unittest
+
+from blinker._saferef import safe_ref
+
+
+class _Sample1(object):
+    def x(self):
+        pass
+
+
+def _sample2(obj):
+    pass
+
+
+class _Sample3(object):
+    def __call__(self, obj):
+        pass
+
+
+class TestSaferef(unittest.TestCase):
+
+    # XXX: The original tests had a test for closure, and it had an
+    # off-by-one problem, perhaps due to scope issues.  It has been
+    # removed from this test suite.
+
+    def setUp(self):
+        ts = []
+        ss = []
+        for x in range(100):
+            t = _Sample1()
+            ts.append(t)
+            s = safe_ref(t.x, self._closure)
+            ss.append(s)
+        ts.append(_sample2)
+        ss.append(safe_ref(_sample2, self._closure))
+        for x in range(30):
+            t = _Sample3()
+            ts.append(t)
+            s = safe_ref(t, self._closure)
+            ss.append(s)
+        self.ts = ts
+        self.ss = ss
+        self.closure_count = 0
+
+    def tearDown(self):
+        if hasattr(self, 'ts'):
+            del self.ts
+        if hasattr(self, 'ss'):
+            del self.ss
+
+    def test_In(self):
+        """Test the `in` operator for safe references (cmp)"""
+        for t in self.ts[:50]:
+            assert safe_ref(t.x) in self.ss
+
+    def test_Valid(self):
+        """Test that the references are valid (return instance methods)"""
+        for s in self.ss:
+            assert s()
+
+    def test_ShortCircuit(self):
+        """Test that creation short-circuits to reuse existing references"""
+        sd = {}
+        for s in self.ss:
+            sd[s] = 1
+        for t in self.ts:
+            if hasattr(t, 'x'):
+                assert sd.has_key(safe_ref(t.x))
+            else:
+                assert sd.has_key(safe_ref(t))
+
+    def test_Representation(self):
+        """Test that the reference object's representation works
+
+        XXX Doesn't currently check the results, just that no error
+            is raised
+        """
+        repr(self.ss[-1])
+
+    def _closure(self, ref):
+        """Dumb utility mechanism to increment deletion counter"""
+        self.closure_count += 1
+
diff --git a/tests/test_signals.py b/tests/test_signals.py
new file mode 100644
index 0000000..9715293
--- /dev/null
+++ b/tests/test_signals.py
@@ -0,0 +1,235 @@
+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()
diff --git a/tests/test_utilities.py b/tests/test_utilities.py
new file mode 100644
index 0000000..4f802f4
--- /dev/null
+++ b/tests/test_utilities.py
@@ -0,0 +1,24 @@
+import pickle
+
+from blinker._utilities import symbol
+
+
+def test_symbols():
+    foo = symbol('foo')
+    assert foo.name == 'foo'
+    assert foo is symbol('foo')
+
+    bar = symbol('bar')
+    assert foo is not bar
+    assert foo != bar
+    assert not foo == bar
+
+    assert repr(foo) == 'foo'
+
+
+def test_pickled_symbols():
+    foo = symbol('foo')
+
+    for protocol in 0, 1, 2:
+        roundtrip = pickle.loads(pickle.dumps(foo))
+        assert roundtrip is foo