1 """ 2 Tests for platutils.py 3 """ 4 5 #----------------------------------------------------------------------------- 6 # Copyright (C) 2008-2011 The IPython Development Team 7 # 8 # Distributed under the terms of the BSD License. The full license is in 9 # the file COPYING, distributed as part of this software. 10 #----------------------------------------------------------------------------- 11 12 #----------------------------------------------------------------------------- 13 # Imports 14 #----------------------------------------------------------------------------- 15 16 import sys 17 import signal 18 import os 19 import time 20 from _thread import interrupt_main # Py 3 21 import threading 22 23 import pytest 24 25 from IPython.utils.process import (find_cmd, FindCmdError, arg_split, 26 system, getoutput, getoutputerror, 27 get_output_error_code) 28 from IPython.utils.capture import capture_output 29 from IPython.testing import decorators as dec 30 from IPython.testing import tools as tt 31 32 python = os.path.basename(sys.executable) 33 34 #----------------------------------------------------------------------------- 35 # Tests 36 #----------------------------------------------------------------------------- 37 38 39 @dec.skip_win32 40 def test_find_cmd_ls(): 41 """Make sure we can find the full path to ls.""" 42 path = find_cmd("ls") 43 assert path.endswith("ls") 44 45 46 @dec.skip_if_not_win32 47 def test_find_cmd_pythonw(): 48 """Try to find pythonw on Windows.""" 49 path = find_cmd('pythonw') 50 assert path.lower().endswith('pythonw.exe'), path 51 52 53 def test_find_cmd_fail(): 54 """Make sure that FindCmdError is raised if we can't find the cmd.""" 55 pytest.raises(FindCmdError, find_cmd, "asdfasdf") 56 57 58 @dec.skip_win32 59 @pytest.mark.parametrize( 60 "argstr, argv", 61 [ 62 ("hi", ["hi"]), 63 ("hello there", ["hello", "there"]), 64 # \u01ce == \N{LATIN SMALL LETTER A WITH CARON} 65 # Do not use \N because the tests crash with syntax error in 66 # some cases, for example windows python2.6. 67 ("h\u01cello", ["h\u01cello"]), 68 ('something "with quotes"', ["something", '"with quotes"']), 69 ], 70 ) 71 def test_arg_split(argstr, argv): 72 """Ensure that argument lines are correctly split like in a shell.""" 73 assert arg_split(argstr) == argv 74 75 76 @dec.skip_if_not_win32 77 @pytest.mark.parametrize( 78 "argstr,argv", 79 [ 80 ("hi", ["hi"]), 81 ("hello there", ["hello", "there"]), 82 ("h\u01cello", ["h\u01cello"]), 83 ('something "with quotes"', ["something", "with quotes"]), 84 ], 85 ) 86 def test_arg_split_win32(argstr, argv): 87 """Ensure that argument lines are correctly split like in a shell.""" 88 assert arg_split(argstr) == argv 89 90 91 class SubProcessTestCase(tt.TempFileMixin): 92 def setUp(self): 93 """Make a valid python temp file.""" 94 lines = [ "import sys", 95 "print('on stdout', end='', file=sys.stdout)", 96 "print('on stderr', end='', file=sys.stderr)", 97 "sys.stdout.flush()", 98 "sys.stderr.flush()"] 99 self.mktmp('\n'.join(lines)) 100 101 def test_system(self): 102 status = system(f'{python} "{self.fname}"') 103 self.assertEqual(status, 0) 104 105 def test_system_quotes(self): 106 status = system('%s -c "import sys"' % python) 107 self.assertEqual(status, 0) 108 109 def assert_interrupts(self, command): 110 """ 111 Interrupt a subprocess after a second. 112 """ 113 if threading.main_thread() != threading.current_thread(): 114 raise pytest.skip("Can't run this test if not in main thread.") 115 116 # Some tests can overwrite SIGINT handler (by using pdb for example), 117 # which then breaks this test, so just make sure it's operating 118 # normally. 119 signal.signal(signal.SIGINT, signal.default_int_handler) 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 start = time.time() 128 try: 129 result = command() 130 except KeyboardInterrupt: 131 # Success! 132 pass 133 end = time.time() 134 self.assertTrue( 135 end - start < 2, "Process didn't die quickly: %s" % (end - start) 136 ) 137 return result 138 139 def test_system_interrupt(self): 140 """ 141 When interrupted in the way ipykernel interrupts IPython, the 142 subprocess is interrupted. 143 """ 144 def command(): 145 return system('%s -c "import time; time.sleep(5)"' % python) 146 147 status = self.assert_interrupts(command) 148 self.assertNotEqual( 149 status, 0, f"The process wasn't interrupted. Status: {status}" 150 ) 151 152 def test_getoutput(self): 153 out = getoutput(f'{python} "{self.fname}"') 154 # we can't rely on the order the line buffered streams are flushed 155 try: 156 self.assertEqual(out, 'on stderron stdout') 157 except AssertionError: 158 self.assertEqual(out, 'on stdouton stderr') 159 160 def test_getoutput_quoted(self): 161 out = getoutput('%s -c "print (1)"' % python) 162 self.assertEqual(out.strip(), '1') 163 164 #Invalid quoting on windows 165 @dec.skip_win32 166 def test_getoutput_quoted2(self): 167 out = getoutput("%s -c 'print (1)'" % python) 168 self.assertEqual(out.strip(), '1') 169 out = getoutput("%s -c 'print (\"1\")'" % python) 170 self.assertEqual(out.strip(), '1') 171 172 def test_getoutput_error(self): 173 out, err = getoutputerror(f'{python} "{self.fname}"') 174 self.assertEqual(out, 'on stdout') 175 self.assertEqual(err, 'on stderr') 176 177 def test_get_output_error_code(self): 178 quiet_exit = '%s -c "import sys; sys.exit(1)"' % python 179 out, err, code = get_output_error_code(quiet_exit) 180 self.assertEqual(out, '') 181 self.assertEqual(err, '') 182 self.assertEqual(code, 1) 183 out, err, code = get_output_error_code(f'{python} "{self.fname}"') 184 self.assertEqual(out, 'on stdout') 185 self.assertEqual(err, 'on stderr') 186 self.assertEqual(code, 0) 187 188 189