##// END OF EJS Templates
Assert status is OK.
Itamar Turner-Trauring -
Show More
@@ -1,210 +1,211 b''
1 1 # encoding: utf-8
2 2 """
3 3 Tests for platutils.py
4 4 """
5 5
6 6 #-----------------------------------------------------------------------------
7 7 # Copyright (C) 2008-2011 The IPython Development Team
8 8 #
9 9 # Distributed under the terms of the BSD License. The full license is in
10 10 # the file COPYING, distributed as part of this software.
11 11 #-----------------------------------------------------------------------------
12 12
13 13 #-----------------------------------------------------------------------------
14 14 # Imports
15 15 #-----------------------------------------------------------------------------
16 16
17 17 import sys
18 18 import os
19 19 import time
20 20 from _thread import interrupt_main # Py 3
21 21 import threading
22 22 from unittest import SkipTest
23 23
24 24 import nose.tools as nt
25 25
26 26 from IPython.utils.process import (find_cmd, FindCmdError, arg_split,
27 27 system, getoutput, getoutputerror,
28 28 get_output_error_code)
29 29 from IPython.utils.capture import capture_output
30 30 from IPython.testing import decorators as dec
31 31 from IPython.testing import tools as tt
32 32
33 33 python = os.path.basename(sys.executable)
34 34
35 35 #-----------------------------------------------------------------------------
36 36 # Tests
37 37 #-----------------------------------------------------------------------------
38 38
39 39
40 40 @dec.skip_win32
41 41 def test_find_cmd_ls():
42 42 """Make sure we can find the full path to ls."""
43 43 path = find_cmd('ls')
44 44 nt.assert_true(path.endswith('ls'))
45 45
46 46
47 47 def has_pywin32():
48 48 try:
49 49 import win32api
50 50 except ImportError:
51 51 return False
52 52 return True
53 53
54 54
55 55 @dec.onlyif(has_pywin32, "This test requires win32api to run")
56 56 def test_find_cmd_pythonw():
57 57 """Try to find pythonw on Windows."""
58 58 path = find_cmd('pythonw')
59 59 assert path.lower().endswith('pythonw.exe'), path
60 60
61 61
62 62 @dec.onlyif(lambda : sys.platform != 'win32' or has_pywin32(),
63 63 "This test runs on posix or in win32 with win32api installed")
64 64 def test_find_cmd_fail():
65 65 """Make sure that FindCmdError is raised if we can't find the cmd."""
66 66 nt.assert_raises(FindCmdError,find_cmd,'asdfasdf')
67 67
68 68
69 69 @dec.skip_win32
70 70 def test_arg_split():
71 71 """Ensure that argument lines are correctly split like in a shell."""
72 72 tests = [['hi', ['hi']],
73 73 [u'hi', [u'hi']],
74 74 ['hello there', ['hello', 'there']],
75 75 # \u01ce == \N{LATIN SMALL LETTER A WITH CARON}
76 76 # Do not use \N because the tests crash with syntax error in
77 77 # some cases, for example windows python2.6.
78 78 [u'h\u01cello', [u'h\u01cello']],
79 79 ['something "with quotes"', ['something', '"with quotes"']],
80 80 ]
81 81 for argstr, argv in tests:
82 82 nt.assert_equal(arg_split(argstr), argv)
83 83
84 84 @dec.skip_if_not_win32
85 85 def test_arg_split_win32():
86 86 """Ensure that argument lines are correctly split like in a shell."""
87 87 tests = [['hi', ['hi']],
88 88 [u'hi', [u'hi']],
89 89 ['hello there', ['hello', 'there']],
90 90 [u'h\u01cello', [u'h\u01cello']],
91 91 ['something "with quotes"', ['something', 'with quotes']],
92 92 ]
93 93 for argstr, argv in tests:
94 94 nt.assert_equal(arg_split(argstr), argv)
95 95
96 96
97 97 class SubProcessTestCase(tt.TempFileMixin):
98 98 def setUp(self):
99 99 """Make a valid python temp file."""
100 100 lines = [ "import sys",
101 101 "print('on stdout', end='', file=sys.stdout)",
102 102 "print('on stderr', end='', file=sys.stderr)",
103 103 "sys.stdout.flush()",
104 104 "sys.stderr.flush()"]
105 105 self.mktmp('\n'.join(lines))
106 106
107 107 def test_system(self):
108 108 status = system('%s "%s"' % (python, self.fname))
109 109 self.assertEqual(status, 0)
110 110
111 111 def test_system_quotes(self):
112 112 status = system('%s -c "import sys"' % python)
113 113 self.assertEqual(status, 0)
114 114
115 115 def assert_interrupts(self, command):
116 116 """
117 117 Interrupt a subprocess after a second.
118 118 """
119 119 if threading.main_thread() != threading.current_thread():
120 120 raise nt.SkipTest("Can't run this test if not in main thread.")
121 121
122 122 def interrupt():
123 123 # Wait for subprocess to start:
124 124 time.sleep(0.5)
125 125 interrupt_main()
126 126
127 127 threading.Thread(target=interrupt).start()
128 128 start = time.time()
129 129 try:
130 130 result = command()
131 131 except KeyboardInterrupt:
132 132 # Success!
133 133 return
134 134 end = time.time()
135 135 self.assertTrue(
136 136 end - start < 2, "Process didn't die quickly: %s" % (end - start)
137 137 )
138 138 return result
139 139
140 140 def test_system_interrupt(self):
141 141 """
142 142 When interrupted in the way ipykernel interrupts IPython, the
143 143 subprocess is interrupted.
144 144 """
145 145 def command():
146 146 return system('%s -c "import time; time.sleep(5)"' % python)
147 147
148 148 status = self.assert_interrupts(command)
149 149 self.assertNotEqual(
150 150 status, 0, "The process wasn't interrupted. Status: %s" % (status,)
151 151 )
152 152
153 153 def test_stderr_while_stdout_open(self):
154 154 """
155 155 If lots of data is written to stderr while stdout is still open, enough
156 156 data to fill the pipe buffer in fact, the process still exits (i.e.
157 157 there is no deadlock).
158 158 """
159 159 with capture_output(display=False):
160 system(("%s -c 'import sys\nfor i in range(20000): " +
161 "sys.stderr.write(\" \" * 100 + \"\\n\")'") % (python,))
160 status = system(("%s -c \"import sys\nfor i in range(20000): " +
161 "sys.stderr.write('b' * 100)\"") % (python,))
162 self.assertEqual(status, 0)
162 163
163 164 def test_getoutput(self):
164 165 out = getoutput('%s "%s"' % (python, self.fname))
165 166 # we can't rely on the order the line buffered streams are flushed
166 167 try:
167 168 self.assertEqual(out, 'on stderron stdout')
168 169 except AssertionError:
169 170 self.assertEqual(out, 'on stdouton stderr')
170 171
171 172 def test_getoutput_interrupt(self):
172 173 """
173 174 When interrupted in the way ipykernel interrupts IPython, the
174 175 subprocess is interrupted.
175 176 """
176 177 raise SkipTest("This fails on POSIX too, revisit in future.")
177 178 def command():
178 179 return getoutput('%s -c "import time; time.sleep(5)"' % (python, ))
179 180
180 181 self.assert_interrupts(command)
181 182
182 183 def test_getoutput_quoted(self):
183 184 out = getoutput('%s -c "print (1)"' % python)
184 185 self.assertEqual(out.strip(), '1')
185 186
186 187 #Invalid quoting on windows
187 188 @dec.skip_win32
188 189 def test_getoutput_quoted2(self):
189 190 out = getoutput("%s -c 'print (1)'" % python)
190 191 self.assertEqual(out.strip(), '1')
191 192 out = getoutput("%s -c 'print (\"1\")'" % python)
192 193 self.assertEqual(out.strip(), '1')
193 194
194 195 def test_getoutput_error(self):
195 196 out, err = getoutputerror('%s "%s"' % (python, self.fname))
196 197 self.assertEqual(out, 'on stdout')
197 198 self.assertEqual(err, 'on stderr')
198 199
199 200 def test_get_output_error_code(self):
200 201 quiet_exit = '%s -c "import sys; sys.exit(1)"' % python
201 202 out, err, code = get_output_error_code(quiet_exit)
202 203 self.assertEqual(out, '')
203 204 self.assertEqual(err, '')
204 205 self.assertEqual(code, 1)
205 206 out, err, code = get_output_error_code('%s "%s"' % (python, self.fname))
206 207 self.assertEqual(out, 'on stdout')
207 208 self.assertEqual(err, 'on stderr')
208 209 self.assertEqual(code, 0)
209 210
210 211
General Comments 0
You need to be logged in to leave comments. Login now