Move test

Author: glentakahashi

ipykernel/tests/test_kernel.py

@@ -391,29 +391,3 @@ def test_interrupt_during_pdb_set_trace():
         validate_message(reply, 'execute_reply', msg_id)
         reply = kc.get_shell_msg(timeout=TIMEOUT)
         validate_message(reply, 'execute_reply', msg_id2)
-
-def test_abort_execute_requests():
-    """test that execute_request's are aborted after an error"""
-    with kernel() as kc:
-        msg_id1 = kc.execute(code="assert False")
-        msg_id2 = kc.execute(code="assert True")
-        reply1 = kc.get_shell_msg(timeout=TIMEOUT)
-        reply2 = kc.get_shell_msg(timeout=TIMEOUT)
-        assert reply1['content']['status'] == 'error'
-        assert reply2['content']['status'] == 'aborted'
-
-def test_dont_abort_non_execute_requests():
-    """test that kernel_info, comm_info and inspect requests are not aborted after an error"""
-    with kernel() as kc:
-        msg_id1 = kc.execute(code="assert False")
-        msg_id2 = kc.kernel_info()
-        msg_id3 = kc.comm_info()
-        msg_id4 = kc.inspect(code="pri")
-        reply1 = kc.get_shell_msg(timeout=TIMEOUT) # execute
-        reply2 = kc.get_shell_msg(timeout=TIMEOUT) # kernel_info
-        reply3 = kc.get_shell_msg(timeout=TIMEOUT) # comm_info
-        reply4 = kc.get_shell_msg(timeout=TIMEOUT) # inspect
-        assert reply1['content']['status'] == 'error'
-        assert reply2['content']['status'] == 'ok'
-        assert reply3['content']['status'] == 'ok'
-        assert reply4['content']['status'] == 'ok'

ipykernel/tests/test_message_spec.py

@@ -355,6 +355,30 @@ def test_execute_stop_on_error():
     assert reply['content']['status'] == 'ok'
 
 
+def test_non_execute_stop_on_error():
+    """test that non-execute_request's are not aborted after an error"""
+    flush_channels()
+
+    fail = '\n'.join([
+        # sleep to ensure subsequent message is waiting in the queue to be aborted
+        'import time',
+        'time.sleep(0.5)',
+        'raise ValueError',
+    ])
+    KC.execute(code=fail)
+    KC.kernel_info()
+    KC.comm_info()
+    KC.inspect(code="print")
+    reply = KC.get_shell_msg(timeout=TIMEOUT) # execute
+    assert reply['content']['status'] == 'error'
+    reply = KC.get_shell_msg(timeout=TIMEOUT) # kernel_info
+    assert reply['content']['status'] == 'ok'
+    reply = KC.get_shell_msg(timeout=TIMEOUT) # comm_info
+    assert reply['content']['status'] == 'ok'
+    reply = KC.get_shell_msg(timeout=TIMEOUT) # inspect
+    assert reply['content']['status'] == 'ok'
+
+
 def test_user_expressions():
     flush_channels()