Add tests

Author: sahil1105

ipyparallel/tests/test_magics.py

@@ -1,5 +1,6 @@
 """Test Parallel magics"""
 import re
+import signal
 import sys
 import time
 
@@ -480,3 +481,45 @@ def test_cellpx_block(self):
                     pass
             self.assertNotIn('Async', io.stdout)
             self.assertEqual(view.block, False)
+
+    def cellpx_keyboard_interrupt_test_helper(self, sig=None):
+        """%%px with Keyboard Interrupt on blocking execution"""
+
+        ip = get_ipython()
+        v = self.client[:]
+        v.block = True
+        v.activate()
+
+        def _sigalarm(sig, frame):
+            raise KeyboardInterrupt
+
+        signal.signal(signal.SIGALRM, _sigalarm)
+        signal.alarm(2)
+        with capture_output(display=False) as io:
+            ip.run_cell_magic(
+                "px",
+                "" if sig is None else f"--signal-on-interrupt {sig}",
+                "print('Entering...'); import time; time.sleep(5); print('Exiting...');",
+            )
+
+        print(io.stdout)
+        print(io.stderr, file=sys.stderr)
+        assert (
+            'Received Keyboard Interrupt. Sending signal {} to engines...'.format(
+                "SIGINT" if sig is None else sig
+            )
+            in io.stderr
+        )
+        assert 'Exiting...' not in io.stdout
+
+    def test_cellpx_keyboard_interrupt_default(self):
+        self.cellpx_keyboard_interrupt_test_helper()
+
+    def test_cellpx_keyboard_interrupt_SIGINT(self):
+        self.cellpx_keyboard_interrupt_test_helper("SIGINT")
+
+    def test_cellpx_keyboard_interrupt_signal_2(self):
+        self.cellpx_keyboard_interrupt_test_helper("2")
+
+    def test_cellpx_keyboard_interrupt_signal_0(self):
+        self.cellpx_keyboard_interrupt_test_helper("0")