Show More
@@ -16,6 +16,9 b' Tests for platutils.py' | |||||
16 |
|
16 | |||
17 | import sys |
|
17 | import sys | |
18 | import os |
|
18 | import os | |
|
19 | import time | |||
|
20 | from _thread import interrupt_main # Py 3 | |||
|
21 | import threading | |||
19 |
|
22 | |||
20 | import nose.tools as nt |
|
23 | import nose.tools as nt | |
21 |
|
24 | |||
@@ -107,6 +110,29 b' class SubProcessTestCase(tt.TempFileMixin):' | |||||
107 | status = system('%s -c "import sys"' % python) |
|
110 | status = system('%s -c "import sys"' % python) | |
108 | self.assertEqual(status, 0) |
|
111 | self.assertEqual(status, 0) | |
109 |
|
112 | |||
|
113 | def test_system_interrupt(self): | |||
|
114 | """ | |||
|
115 | When interrupted in the way ipykernel interrupts IPython, the | |||
|
116 | subprocess is interrupted. | |||
|
117 | """ | |||
|
118 | if threading.main_thread() != threading.current_thread(): | |||
|
119 | raise nt.SkipTest("Can't run this test if not in main thread.") | |||
|
120 | ||||
|
121 | def interrupt(): | |||
|
122 | # Wait for subprocess to start: | |||
|
123 | time.sleep(0.5) | |||
|
124 | interrupt_main() | |||
|
125 | ||||
|
126 | threading.Thread(target=interrupt).start() | |||
|
127 | try: | |||
|
128 | status = system('%s -c "import time; time.sleep(5)"' % python) | |||
|
129 | except KeyboardInterrupt: | |||
|
130 | # Success! | |||
|
131 | return | |||
|
132 | self.assertNotEqual( | |||
|
133 | status, 0, "The process wasn't interrupted. Status: %s" % (status,) | |||
|
134 | ) | |||
|
135 | ||||
110 | def test_getoutput(self): |
|
136 | def test_getoutput(self): | |
111 | out = getoutput('%s "%s"' % (python, self.fname)) |
|
137 | out = getoutput('%s "%s"' % (python, self.fname)) | |
112 | # we can't rely on the order the line buffered streams are flushed |
|
138 | # we can't rely on the order the line buffered streams are flushed | |
@@ -131,7 +157,7 b' class SubProcessTestCase(tt.TempFileMixin):' | |||||
131 | out, err = getoutputerror('%s "%s"' % (python, self.fname)) |
|
157 | out, err = getoutputerror('%s "%s"' % (python, self.fname)) | |
132 | self.assertEqual(out, 'on stdout') |
|
158 | self.assertEqual(out, 'on stdout') | |
133 | self.assertEqual(err, 'on stderr') |
|
159 | self.assertEqual(err, 'on stderr') | |
134 |
|
160 | |||
135 | def test_get_output_error_code(self): |
|
161 | def test_get_output_error_code(self): | |
136 | quiet_exit = '%s -c "import sys; sys.exit(1)"' % python |
|
162 | quiet_exit = '%s -c "import sys; sys.exit(1)"' % python | |
137 | out, err, code = get_output_error_code(quiet_exit) |
|
163 | out, err, code = get_output_error_code(quiet_exit) | |
@@ -142,3 +168,5 b' class SubProcessTestCase(tt.TempFileMixin):' | |||||
142 | self.assertEqual(out, 'on stdout') |
|
168 | self.assertEqual(out, 'on stdout') | |
143 | self.assertEqual(err, 'on stderr') |
|
169 | self.assertEqual(err, 'on stderr') | |
144 | self.assertEqual(code, 0) |
|
170 | self.assertEqual(code, 0) | |
|
171 | ||||
|
172 |
General Comments 0
You need to be logged in to leave comments.
Login now