Added ``Signal.temporarily_connected_to`` context manager

2 files changed, 64 insertions, 0 deletions

diff --git a/tests/test_context.py b/tests/test_context.py
new file mode 100644
index 0000000..b5de372
--- /dev/null
+++ b/tests/test_context.py
@@ -0,0 +1,56 @@
+from __future__ import with_statement
+
+from blinker import Signal
+
+
+def test_temp_connection():
+    sig = Signal()
+
+    canary = []
+    receiver = lambda sender: canary.append(sender)
+
+    sig.send(1)
+    with sig.temporarily_connected_to(receiver):
+        sig.send(2)
+    sig.send(3)
+
+    assert canary == [2]
+    assert not sig.receivers
+
+
+def test_temp_connection_for_sender():
+    sig = Signal()
+
+    canary = []
+    receiver = lambda sender: canary.append(sender)
+
+    with sig.temporarily_connected_to(receiver, sender=2):
+        sig.send(1)
+        sig.send(2)
+
+    assert canary == [2]
+    assert not sig.receivers
+
+
+def test_temp_connection_failure():
+    sig = Signal()
+
+    canary = []
+    receiver = lambda sender: canary.append(sender)
+
+    class Failure(Exception):
+        pass
+
+    try:
+        sig.send(1)
+        with sig.temporarily_connected_to(receiver):
+            sig.send(2)
+            raise Failure
+        sig.send(3)
+    except Failure:
+        pass
+    else:
+        raise AssertionError("Context manager did not propagate.")
+
+    assert canary == [2]
+    assert not sig.receivers
diff --git a/tests/test_signals.py b/tests/test_signals.py
index 9715293..5328872 100644
--- a/tests/test_signals.py
+++ b/tests/test_signals.py
@@ -1,3 +1,4 @@
+import sys
 import blinker
 from nose.tools import assert_raises
 
@@ -233,3 +234,10 @@ def test_named_blinker():
 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)