1+ # Python Tools for Visual Studio
2+ # Copyright(c) Microsoft Corporation
3+ # All rights reserved.
4+ #
5+ # Licensed under the Apache License, Version 2.0 (the License); you may not use
6+ # this file except in compliance with the License. You may obtain a copy of the
7+ # License at http://www.apache.org/licenses/LICENSE-2.0
8+ #
9+ # THIS CODE IS PROVIDED ON AN *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS
10+ # OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY
11+ # IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE,
12+ # MERCHANTABLITY OR NON-INFRINGEMENT.
13+ #
14+ # See the Apache Version 2.0 License for specific language governing
15+ # permissions and limitations under the License.
16+
17+ __author__ = "Microsoft Corporation <ptvshelp@microsoft.com>"
18+ __version__ = "3.0.0.0"
19+
20+ import os
21+ import sys
22+ import json
23+ import unittest
24+ import socket
25+ import traceback
26+ from types import CodeType , FunctionType
27+ import signal
28+ try :
29+ import thread
30+ except :
31+ import _thread as thread
32+
33+ class _TestOutput (object ):
34+ """file like object which redirects output to the repl window."""
35+ errors = 'strict'
36+
37+ def __init__ (self , old_out , is_stdout ):
38+ self .is_stdout = is_stdout
39+ self .old_out = old_out
40+ if sys .version >= '3.' and hasattr (old_out , 'buffer' ):
41+ self .buffer = _TestOutputBuffer (old_out .buffer , is_stdout )
42+
43+ def flush (self ):
44+ if self .old_out :
45+ self .old_out .flush ()
46+
47+ def writelines (self , lines ):
48+ for line in lines :
49+ self .write (line )
50+
51+ @property
52+ def encoding (self ):
53+ return 'utf8'
54+
55+ def write (self , value ):
56+ _channel .send_event ('stdout' if self .is_stdout else 'stderr' , content = value )
57+ if self .old_out :
58+ self .old_out .write (value )
59+
60+ def isatty (self ):
61+ return True
62+
63+ def next (self ):
64+ pass
65+
66+ @property
67+ def name (self ):
68+ if self .is_stdout :
69+ return "<stdout>"
70+ else :
71+ return "<stderr>"
72+
73+ def __getattr__ (self , name ):
74+ return getattr (self .old_out , name )
75+
76+ class _TestOutputBuffer (object ):
77+ def __init__ (self , old_buffer , is_stdout ):
78+ self .buffer = old_buffer
79+ self .is_stdout = is_stdout
80+
81+ def write (self , data ):
82+ _channel .send_event ('stdout' if self .is_stdout else 'stderr' , content = data )
83+ self .buffer .write (data )
84+
85+ def flush (self ):
86+ self .buffer .flush ()
87+
88+ def truncate (self , pos = None ):
89+ return self .buffer .truncate (pos )
90+
91+ def tell (self ):
92+ return self .buffer .tell ()
93+
94+ def seek (self , pos , whence = 0 ):
95+ return self .buffer .seek (pos , whence )
96+
97+ class _IpcChannel (object ):
98+ def __init__ (self , socket , callback ):
99+ self .socket = socket
100+ self .seq = 0
101+ self .callback = callback
102+ self .lock = thread .allocate_lock ()
103+ # start the testing reader thread loop
104+ self .test_thread_id = thread .start_new_thread (self .readSocket , ())
105+
106+ def readSocket (self ):
107+ data = self .socket .recv (1024 )
108+ self .callback ()
109+
110+ def receive (self ):
111+ pass
112+
113+ def send_event (self , name , ** args ):
114+ with self .lock :
115+ body = {'type' : 'event' , 'seq' : self .seq , 'event' :name , 'body' :args }
116+ self .seq += 1
117+ content = json .dumps (body ).encode ('utf8' )
118+ headers = ('Content-Length: %d\n \n ' % (len (content ), )).encode ('utf8' )
119+ self .socket .send (headers )
120+ self .socket .send (content )
121+
122+ _channel = None
123+
124+
125+ class VsTestResult (unittest .TextTestResult ):
126+ def startTest (self , test ):
127+ super (VsTestResult , self ).startTest (test )
128+ if _channel is not None :
129+ _channel .send_event (
130+ name = 'start' ,
131+ test = test .id ()
132+ )
133+
134+ def addError (self , test , err ):
135+ super (VsTestResult , self ).addError (test , err )
136+ self .sendResult (test , 'failed' , err )
137+
138+ def addFailure (self , test , err ):
139+ super (VsTestResult , self ).addFailure (test , err )
140+ self .sendResult (test , 'failed' , err )
141+
142+ def addSuccess (self , test ):
143+ super (VsTestResult , self ).addSuccess (test )
144+ self .sendResult (test , 'passed' )
145+
146+ def addSkip (self , test , reason ):
147+ super (VsTestResult , self ).addSkip (test , reason )
148+ self .sendResult (test , 'skipped' )
149+
150+ def addExpectedFailure (self , test , err ):
151+ super (VsTestResult , self ).addExpectedFailure (test , err )
152+ self .sendResult (test , 'failed' , err )
153+
154+ def addUnexpectedSuccess (self , test ):
155+ super (VsTestResult , self ).addUnexpectedSuccess (test )
156+ self .sendResult (test , 'passed' )
157+
158+ def sendResult (self , test , outcome , trace = None ):
159+ if _channel is not None :
160+ tb = None
161+ message = None
162+ if trace is not None :
163+ traceback .print_exc ()
164+ formatted = traceback .format_exception (* trace )
165+ # Remove the 'Traceback (most recent call last)'
166+ formatted = formatted [1 :]
167+ tb = '' .join (formatted )
168+ message = str (trace [1 ])
169+ _channel .send_event (
170+ name = 'result' ,
171+ outcome = outcome ,
172+ traceback = tb ,
173+ message = message ,
174+ test = test .id ()
175+ )
176+
177+ def stopTests ():
178+ os .kill (os .getpid (), signal .SIGUSR1 )
179+
180+ class ExitCommand (Exception ):
181+ pass
182+
183+ def signal_handler (signal , frame ):
184+ raise ExitCommand ()
185+
186+ def main ():
187+ import os
188+ import sys
189+ import unittest
190+ from optparse import OptionParser
191+ global _channel
192+
193+ parser = OptionParser (prog = 'visualstudio_py_testlauncher' , usage = 'Usage: %prog [<option>] <test names>... ' )
194+ parser .add_option ('-s' , '--secret' , metavar = '<secret>' , help = 'restrict server to only allow clients that specify <secret> when connecting' )
195+ parser .add_option ('-p' , '--port' , type = 'int' , metavar = '<port>' , help = 'listen for debugger connections on <port>' )
196+ parser .add_option ('-x' , '--mixed-mode' , action = 'store_true' , help = 'wait for mixed-mode debugger to attach' )
197+ parser .add_option ('-t' , '--test' , type = 'str' , dest = 'tests' , action = 'append' , help = 'specifies a test to run' )
198+ parser .add_option ('-c' , '--coverage' , type = 'str' , help = 'enable code coverage and specify filename' )
199+ parser .add_option ('-r' , '--result-port' , type = 'int' , help = 'connect to port on localhost and send test results' )
200+ parser .add_option ('--us' , type = 'str' , help = 'Directory to start discovery' )
201+ parser .add_option ('--up' , type = 'str' , help = 'Pattern to match test files (' 'test*.py' ' default)' )
202+ parser .add_option ('--ut' , type = 'str' , help = 'Top level directory of project (default to start directory)' )
203+ parser .add_option ('--uvInt' , '--verboseInt' , type = 'int' , help = 'Verbose output' )
204+ parser .add_option ('--uf' , '--failfast' , type = 'str' , help = 'Stop on first failure' )
205+ parser .add_option ('--uc' , '--catch' , type = 'str' , help = 'Catch control-C and display results' )
206+ (opts , _ ) = parser .parse_args ()
207+
208+ sys .path [0 ] = os .getcwd ()
209+ if opts .result_port :
210+ signal .signal (signal .SIGUSR1 , signal_handler )
211+ _channel = _IpcChannel (socket .create_connection (('127.0.0.1' , opts .result_port )), stopTests )
212+ sys .stdout = _TestOutput (sys .stdout , is_stdout = True )
213+ sys .stderr = _TestOutput (sys .stderr , is_stdout = False )
214+
215+ if opts .secret and opts .port :
216+ from ptvsd .visualstudio_py_debugger import DONT_DEBUG , DEBUG_ENTRYPOINTS , get_code
217+ from ptvsd .attach_server import DEFAULT_PORT , enable_attach , wait_for_attach
218+
219+ DONT_DEBUG .append (os .path .normcase (__file__ ))
220+ DEBUG_ENTRYPOINTS .add (get_code (main ))
221+
222+ enable_attach (opts .secret , ('127.0.0.1' , getattr (opts , 'port' , DEFAULT_PORT )), redirect_output = True )
223+ wait_for_attach ()
224+ elif opts .mixed_mode :
225+ # For mixed-mode attach, there's no ptvsd and hence no wait_for_attach(),
226+ # so we have to use Win32 API in a loop to do the same thing.
227+ from time import sleep
228+ from ctypes import windll , c_char
229+ while True :
230+ if windll .kernel32 .IsDebuggerPresent () != 0 :
231+ break
232+ sleep (0.1 )
233+ try :
234+ debugger_helper = windll ['Microsoft.PythonTools.Debugger.Helper.x86.dll' ]
235+ except WindowsError :
236+ debugger_helper = windll ['Microsoft.PythonTools.Debugger.Helper.x64.dll' ]
237+ isTracing = c_char .in_dll (debugger_helper , "isTracing" )
238+ while True :
239+ if isTracing .value != 0 :
240+ break
241+ sleep (0.1 )
242+
243+ cov = None
244+ try :
245+ if opts .coverage :
246+ try :
247+ import coverage
248+ cov = coverage .coverage (opts .coverage )
249+ cov .load ()
250+ cov .start ()
251+ except :
252+ pass
253+ if opts .tests is None :
254+ if opts .us is None :
255+ opts .us = '.'
256+ if opts .up is None :
257+ opts .up = 'test*.py'
258+ tests = unittest .defaultTestLoader .discover (opts .us , opts .up )
259+ else :
260+ tests = unittest .defaultTestLoader .loadTestsFromNames (opts .tests )
261+ if opts .uvInt is None :
262+ opts .uvInt = 0
263+ runner = unittest .TextTestRunner (verbosity = opts .uvInt , resultclass = VsTestResult )
264+ if opts .uf is not None :
265+ runner .failfast = 1
266+
267+ result = runner .run (tests )
268+
269+ sys .exit (not result .wasSuccessful ())
270+ finally :
271+ if cov is not None :
272+ cov .stop ()
273+ cov .save ()
274+ cov .xml_report (outfile = opts .coverage + '.xml' , omit = __file__ )
275+ if _channel is not None :
276+ _channel .send_event (
277+ name = 'done'
278+ )
279+ _channel .socket .close ()
280+
281+ if __name__ == '__main__' :
282+ main ()
0 commit comments