Mercurial > ~astiob > upreckon > hgweb

comparison testcases.py @ 61:24f144e11b5e

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Accurate run-time reporting on Win32
author Oleg Oshmyan <chortos@inbox.lv>
date Tue, 21 Dec 2010 19:38:58 +0200
parents c95addbe0851
children 593ad09cd69b
comparison
equal deleted inserted replaced
60:7c6dba0b84f2 61:24f144e11b5e
92 from signal import SIGCHLD, signal, SIG_DFL 92 from signal import SIGCHLD, signal, SIG_DFL
93 from select import select, error as select_error 93 from select import select, error as select_error
94 from errno import EINTR 94 from errno import EINTR
95 except ImportError: 95 except ImportError:
96 try: 96 try:
97 from _subprocess import WAIT_OBJECT_0, STD_INPUT_HANDLE, INFINITE
98 except ImportError:
99 WAIT_OBJECT_0 = 0
100 STD_INPUT_HANDLE = -10
101 INFINITE = -1
102 try:
97 import ctypes 103 import ctypes
104 SetConsoleMode = ctypes.windll.kernel32.SetConsoleMode
105 FlushConsoleInputBuffer = ctypes.windll.kernel32.FlushConsoleInputBuffer
98 WaitForMultipleObjects = ctypes.windll.kernel32.WaitForMultipleObjects 106 WaitForMultipleObjects = ctypes.windll.kernel32.WaitForMultipleObjects
107 ReadConsoleInputA = ctypes.windll.kernel32.ReadConsoleInputA
108 try:
109 from _subprocess import GetStdHandle
110 except ImportError:
111 GetStdHandle = ctypes.windll.kernel32.GetStdHandle
99 except (ImportError, AttributeError): 112 except (ImportError, AttributeError):
100 pass 113 console_input = False
101 else: 114 else:
102 # TODO: implement Win32 call() 115 hStdin = GetStdHandle(STD_INPUT_HANDLE)
103 pass 116 console_input = bool(SetConsoleMode(hStdin, 1))
104 call = None 117 if console_input:
118 FlushConsoleInputBuffer(hStdin)
119 class KEY_EVENT_RECORD(ctypes.Structure):
120 _fields_ = (("bKeyDown", ctypes.c_int),
121 ("wRepeatCount", ctypes.c_ushort),
122 ("wVirtualKeyCode", ctypes.c_ushort),
123 ("wVirtualScanCode", ctypes.c_ushort),
124 ("UnicodeChar", ctypes.c_wchar),
125 ("dwControlKeyState", ctypes.c_uint))
126 class INPUT_RECORD(ctypes.Structure):
127 _fields_ = (("EventType", ctypes.c_int),
128 ("KeyEvent", KEY_EVENT_RECORD))
129 # Memory limits (currently) are not supported
130 def call(*args, **kwargs):
131 case = kwargs.pop('case')
132 try:
133 case.process = Popen(*args, **kwargs)
134 except OSError:
135 raise CannotStartTestee(sys.exc_info()[1])
136 case.time_started = clock()
137 if not console_input:
138 if case.maxtime:
139 if WaitForSingleObject(case.process._handle, int(case.maxtime * 1000)) != WAIT_OBJECT_0:
140 raise TimeLimitExceeded
141 else:
142 case.process.wait()
143 else:
144 ir = INPUT_RECORD()
145 n = ctypes.c_int()
146 lpHandles = (ctypes.c_int * 2)(hStdin, case.process._handle)
147 if case.maxtime:
148 time_end = clock() + case.maxtime
149 while case.process.poll() is None:
150 remaining = time_end - clock()
151 if remaining > 0:
152 if WaitForMultipleObjects(2, lpHandles, False, int(remaining * 1000)) == WAIT_OBJECT_0:
153 ReadConsoleInputA(hStdin, ctypes.byref(ir), 1, ctypes.byref(n))
154 if ir.EventType == 1 and ir.KeyEvent.bKeyDown and ir.KeyEvent.wVirtualKeyCode == 27:
155 raise CanceledByUser
156 else:
157 raise TimeLimitExceeded
158 else:
159 while case.process.poll() is None:
160 if WaitForMultipleObjects(2, lpHandles, False, INFINITE) == WAIT_OBJECT_0:
161 ReadConsoleInputA(hStdin, ctypes.byref(ir), 1, ctypes.byref(n))
162 if ir.EventType == 1 and ir.KeyEvent.bKeyDown and ir.KeyEvent.wVirtualKeyCode == 27:
163 raise CanceledByUser
164 case.time_stopped = clock()
165 if not console_input:
166 try:
167 try:
168 from _subprocess import WaitForSingleObject
169 except ImportError:
170 import ctypes
171 WaitForSingleObject = ctypes.windll.kernel32.WaitForSingleObject
172 except (ImportError, AttributeError):
173 # TODO: move the default implementation here
174 call = None
105 else: 175 else:
106 # Make SIGCHLD interrupt sleep() and select() 176 # Make SIGCHLD interrupt sleep() and select()
107 def bury_child(signum, frame): 177 def bury_child(signum, frame):
108 try: 178 try:
109 bury_child.case.time_stopped = clock() 179 bury_child.case.time_stopped = clock()
112 signal(SIGCHLD, bury_child) 182 signal(SIGCHLD, bury_child)
113 183
114 # If you want this to work, don't set any stdio argument to PIPE 184 # If you want this to work, don't set any stdio argument to PIPE
115 def call_real(*args, **kwargs): 185 def call_real(*args, **kwargs):
116 bury_child.case = case = kwargs.pop('case') 186 bury_child.case = case = kwargs.pop('case')
117 # FIXME: kill the process no matter what
118 try: 187 try:
119 case.process = Popen(*args, **kwargs) 188 case.process = Popen(*args, **kwargs)
120 except OSError: 189 except OSError:
121 raise CannotStartTestee(sys.exc_info()[1]) 190 raise CannotStartTestee(sys.exc_info()[1])
122 case.time_started = clock() 191 case.time_started = clock()