##// END OF EJS Templates
more lints
M Bussonnier -
Show More
@@ -1,220 +1,211
1 1 """Windows-specific implementation of process utilities.
2 2
3 3 This file is only meant to be imported by process.py, not by end-users.
4 4 """
5
6 #-----------------------------------------------------------------------------
7 # Copyright (C) 2010-2011 The IPython Development Team
8 #
9 # Distributed under the terms of the BSD License. The full license is in
10 # the file COPYING, distributed as part of this software.
11 #-----------------------------------------------------------------------------
12
13 #-----------------------------------------------------------------------------
14 # Imports
15 #-----------------------------------------------------------------------------
16
17 # stdlib
18 5 import ctypes
19 6 import os
20 7 import subprocess
21 8 import sys
22 9 import time
23 10 from ctypes import POINTER, c_int
24 11 from ctypes.wintypes import HLOCAL, LPCWSTR
25 12 from subprocess import STDOUT
26 13 from threading import Thread
27 14 from types import TracebackType
28 from typing import IO, Any, List, Optional
15 from typing import List, Optional
29 16
30 17 from . import py3compat
31 18 from ._process_common import arg_split as py_arg_split
32 19
33 # our own imports
34 20 from ._process_common import process_handler, read_no_interrupt
35 21 from .encoding import DEFAULT_ENCODING
36 22
37 #-----------------------------------------------------------------------------
38 # Function definitions
39 #-----------------------------------------------------------------------------
23
40 24
41 25 class AvoidUNCPath:
42 26 """A context manager to protect command execution from UNC paths.
43 27
44 28 In the Win32 API, commands can't be invoked with the cwd being a UNC path.
45 29 This context manager temporarily changes directory to the 'C:' drive on
46 30 entering, and restores the original working directory on exit.
47 31
48 32 The context manager returns the starting working directory *if* it made a
49 33 change and None otherwise, so that users can apply the necessary adjustment
50 34 to their system calls in the event of a change.
51 35
52 36 Examples
53 37 --------
54 38 ::
55 39 cmd = 'dir'
56 40 with AvoidUNCPath() as path:
57 41 if path is not None:
58 42 cmd = '"pushd %s &&"%s' % (path, cmd)
59 43 os.system(cmd)
60 44 """
61 45
62 46 def __enter__(self) -> Optional[str]:
63 47 self.path = os.getcwd()
64 48 self.is_unc_path = self.path.startswith(r"\\")
65 49 if self.is_unc_path:
66 50 # change to c drive (as cmd.exe cannot handle UNC addresses)
67 51 os.chdir("C:")
68 52 return self.path
69 53 else:
70 54 # We return None to signal that there was no change in the working
71 55 # directory
72 56 return None
73 57
74 58 def __exit__(
75 self, exc_type: Optional[type[BaseException]], exc_value: Optional[BaseException], traceback:TracebackType
59 self,
60 exc_type: Optional[type[BaseException]],
61 exc_value: Optional[BaseException],
62 traceback: TracebackType,
76 63 ) -> None:
77 64 if self.is_unc_path:
78 65 os.chdir(self.path)
79 66
80 67
81 68 def _system_body(p: subprocess.Popen) -> int:
82 69 """Callback for _system."""
83 70 enc = DEFAULT_ENCODING
84 71
85 72 # Dec 2024: in both of these functions, I'm not sure why we .splitlines()
86 73 # the bytes and then decode each line individually instead of just decoding
87 74 # the whole thing at once.
88 75 def stdout_read() -> None:
89 76 try:
90 77 assert p.stdout is not None
91 78 for byte_line in read_no_interrupt(p.stdout).splitlines():
92 79 line = byte_line.decode(enc, "replace")
93 80 print(line, file=sys.stdout)
94 81 except Exception as e:
95 82 print(f"Error reading stdout: {e}", file=sys.stderr)
96 83
97 84 def stderr_read() -> None:
98 85 try:
99 86 assert p.stderr is not None
100 87 for byte_line in read_no_interrupt(p.stderr).splitlines():
101 88 line = byte_line.decode(enc, "replace")
102 89 print(line, file=sys.stderr)
103 90 except Exception as e:
104 91 print(f"Error reading stderr: {e}", file=sys.stderr)
105 92
106 93 stdout_thread = Thread(target=stdout_read)
107 94 stderr_thread = Thread(target=stderr_read)
108 95
109 96 stdout_thread.start()
110 97 stderr_thread.start()
111 98
112 99 # Wait to finish for returncode. Unfortunately, Python has a bug where
113 100 # wait() isn't interruptible (https://bugs.python.org/issue28168) so poll in
114 101 # a loop instead of just doing `return p.wait()`
115 102 while True:
116 103 result = p.poll()
117 104 if result is None:
118 105 time.sleep(0.01)
119 106 else:
120 107 break
121 108
122 109 # Join the threads to ensure they complete before returning
123 110 stdout_thread.join()
124 111 stderr_thread.join()
125 112
126 113 return result
127 114
128 115
129 116 def system(cmd: str) -> Optional[int]:
130 117 """Win32 version of os.system() that works with network shares.
131 118
132 119 Note that this implementation returns None, as meant for use in IPython.
133 120
134 121 Parameters
135 122 ----------
136 123 cmd : str or list
137 124 A command to be executed in the system shell.
138 125
139 126 Returns
140 127 -------
141 128 int : child process' exit code.
142 129 """
143 130 # The controller provides interactivity with both
144 131 # stdin and stdout
145 #import _process_win32_controller
146 #_process_win32_controller.system(cmd)
132 # import _process_win32_controller
133 # _process_win32_controller.system(cmd)
147 134
148 135 with AvoidUNCPath() as path:
149 136 if path is not None:
150 137 cmd = '"pushd %s &&"%s' % (path, cmd)
151 138 return process_handler(cmd, _system_body)
152 139
140
153 141 def getoutput(cmd: str) -> str:
154 142 """Return standard output of executing cmd in a shell.
155 143
156 144 Accepts the same arguments as os.system().
157 145
158 146 Parameters
159 147 ----------
160 148 cmd : str or list
161 149 A command to be executed in the system shell.
162 150
163 151 Returns
164 152 -------
165 153 stdout : str
166 154 """
167 155
168 156 with AvoidUNCPath() as path:
169 157 if path is not None:
170 158 cmd = '"pushd %s &&"%s' % (path, cmd)
171 159 out = process_handler(cmd, lambda p: p.communicate()[0], STDOUT)
172 160
173 161 if out is None:
174 out = b''
162 out = b""
175 163 return py3compat.decode(out)
176 164
165
177 166 try:
178 167 windll = ctypes.windll # type: ignore [attr-defined]
179 168 CommandLineToArgvW = windll.shell32.CommandLineToArgvW
180 169 CommandLineToArgvW.arg_types = [LPCWSTR, POINTER(c_int)]
181 170 CommandLineToArgvW.restype = POINTER(LPCWSTR)
182 171 LocalFree = windll.kernel32.LocalFree
183 172 LocalFree.res_type = HLOCAL
184 173 LocalFree.arg_types = [HLOCAL]
185 174
186 175 def arg_split(
187 176 commandline: str, posix: bool = False, strict: bool = True
188 177 ) -> List[str]:
189 178 """Split a command line's arguments in a shell-like manner.
190 179
191 180 This is a special version for windows that use a ctypes call to CommandLineToArgvW
192 181 to do the argv splitting. The posix parameter is ignored.
193 182
194 183 If strict=False, process_common.arg_split(...strict=False) is used instead.
195 184 """
196 #CommandLineToArgvW returns path to executable if called with empty string.
185 # CommandLineToArgvW returns path to executable if called with empty string.
197 186 if commandline.strip() == "":
198 187 return []
199 188 if not strict:
200 189 # not really a cl-arg, fallback on _process_common
201 190 return py_arg_split(commandline, posix=posix, strict=strict)
202 191 argvn = c_int()
203 192 result_pointer = CommandLineToArgvW(commandline.lstrip(), ctypes.byref(argvn))
204 193 result_array_type = LPCWSTR * argvn.value
205 194 result = [
206 195 arg
207 196 for arg in result_array_type.from_address(
208 197 ctypes.addressof(result_pointer.contents)
209 198 )
210 199 if arg is not None
211 200 ]
212 LocalFree(result_pointer)
201 # for side effects
202 _ = LocalFree(result_pointer)
213 203 return result
214 204 except AttributeError:
215 205 arg_split = py_arg_split
216 206
207
217 208 def check_pid(pid: int) -> bool:
218 209 # OpenProcess returns 0 if no such process (of ours) exists
219 210 # positive int otherwise
220 211 return bool(windll.kernel32.OpenProcess(1, 0, pid))
General Comments 0
You need to be logged in to leave comments. Login now