+++ /dev/null
-# Python Tools for Visual Studio
-# Copyright(c) Microsoft Corporation
-# All rights reserved.
-#
-# Licensed under the Apache License, Version 2.0 (the License); you may not use
-# this file except in compliance with the License. You may obtain a copy of the
-# License at http://www.apache.org/licenses/LICENSE-2.0
-#
-# THIS CODE IS PROVIDED ON AN *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS
-# OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY
-# IMPLIED WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE,
-# MERCHANTABLITY OR NON-INFRINGEMENT.
-#
-# See the Apache Version 2.0 License for specific language governing
-# permissions and limitations under the License.
-
-__author__ = "Microsoft Corporation <ptvshelp@microsoft.com>"
-__version__ = "3.0.0.0"
-
-import os
-import sys
-import json
-import unittest
-import socket
-import traceback
-from types import CodeType, FunctionType
-import signal
-try:
- import thread
-except:
- import _thread as thread
-
-class _TestOutput(object):
- """file like object which redirects output to the repl window."""
- errors = 'strict'
-
- def __init__(self, old_out, is_stdout):
- self.is_stdout = is_stdout
- self.old_out = old_out
- if sys.version >= '3.' and hasattr(old_out, 'buffer'):
- self.buffer = _TestOutputBuffer(old_out.buffer, is_stdout)
-
- def flush(self):
- if self.old_out:
- self.old_out.flush()
-
- def writelines(self, lines):
- for line in lines:
- self.write(line)
-
- @property
- def encoding(self):
- return 'utf8'
-
- def write(self, value):
- _channel.send_event('stdout' if self.is_stdout else 'stderr', content=value)
- if self.old_out:
- self.old_out.write(value)
- # flush immediately, else things go wonky and out of order
- self.flush()
-
- def isatty(self):
- return True
-
- def next(self):
- pass
-
- @property
- def name(self):
- if self.is_stdout:
- return "<stdout>"
- else:
- return "<stderr>"
-
- def __getattr__(self, name):
- return getattr(self.old_out, name)
-
-class _TestOutputBuffer(object):
- def __init__(self, old_buffer, is_stdout):
- self.buffer = old_buffer
- self.is_stdout = is_stdout
-
- def write(self, data):
- _channel.send_event('stdout' if self.is_stdout else 'stderr', content=data)
- self.buffer.write(data)
-
- def flush(self):
- self.buffer.flush()
-
- def truncate(self, pos = None):
- return self.buffer.truncate(pos)
-
- def tell(self):
- return self.buffer.tell()
-
- def seek(self, pos, whence = 0):
- return self.buffer.seek(pos, whence)
-
-class _IpcChannel(object):
- def __init__(self, socket, callback):
- self.socket = socket
- self.seq = 0
- self.callback = callback
- self.lock = thread.allocate_lock()
- self._closed = False
- # start the testing reader thread loop
- self.test_thread_id = thread.start_new_thread(self.readSocket, ())
-
- def close(self):
- self._closed = True
-
- def readSocket(self):
- try:
- data = self.socket.recv(1024)
- self.callback()
- except OSError:
- if not self._closed:
- raise
-
- def receive(self):
- pass
-
- def send_event(self, name, **args):
- with self.lock:
- body = {'type': 'event', 'seq': self.seq, 'event':name, 'body':args}
- self.seq += 1
- content = json.dumps(body).encode('utf8')
- headers = ('Content-Length: %d\n\n' % (len(content), )).encode('utf8')
- self.socket.send(headers)
- self.socket.send(content)
-
-_channel = None
-
-
-class VsTestResult(unittest.TextTestResult):
- def startTest(self, test):
- super(VsTestResult, self).startTest(test)
- if _channel is not None:
- _channel.send_event(
- name='start',
- test = test.id()
- )
-
- def addError(self, test, err):
- super(VsTestResult, self).addError(test, err)
- self.sendResult(test, 'error', err)
-
- def addFailure(self, test, err):
- super(VsTestResult, self).addFailure(test, err)
- self.sendResult(test, 'failed', err)
-
- def addSuccess(self, test):
- super(VsTestResult, self).addSuccess(test)
- self.sendResult(test, 'passed')
-
- def addSkip(self, test, reason):
- super(VsTestResult, self).addSkip(test, reason)
- self.sendResult(test, 'skipped')
-
- def addExpectedFailure(self, test, err):
- super(VsTestResult, self).addExpectedFailure(test, err)
- self.sendResult(test, 'failed', err)
-
- def addUnexpectedSuccess(self, test):
- super(VsTestResult, self).addUnexpectedSuccess(test)
- self.sendResult(test, 'passed')
-
- def sendResult(self, test, outcome, trace = None):
- if _channel is not None:
- tb = None
- message = None
- if trace is not None:
- traceback.print_exc()
- formatted = traceback.format_exception(*trace)
- # Remove the 'Traceback (most recent call last)'
- formatted = formatted[1:]
- tb = ''.join(formatted)
- message = str(trace[1])
- _channel.send_event(
- name='result',
- outcome=outcome,
- traceback = tb,
- message = message,
- test = test.id()
- )
-
-def stopTests():
- try:
- os.kill(os.getpid(), signal.SIGUSR1)
- except:
- try:
- os.kill(os.getpid(), signal.SIGTERM)
- except:
- pass
-
-class ExitCommand(Exception):
- pass
-
-def signal_handler(signal, frame):
- raise ExitCommand()
-
-def main():
- import os
- import sys
- import unittest
- from optparse import OptionParser
- global _channel
-
- parser = OptionParser(prog = 'visualstudio_py_testlauncher', usage = 'Usage: %prog [<option>] <test names>... ')
- parser.add_option('--debug', action='store_true', help='Whether debugging the unit tests')
- parser.add_option('-x', '--mixed-mode', action='store_true', help='wait for mixed-mode debugger to attach')
- parser.add_option('-t', '--test', type='str', dest='tests', action='append', help='specifies a test to run')
- parser.add_option('--testFile', type='str', help='Fully qualitified path to file name')
- parser.add_option('-c', '--coverage', type='str', help='enable code coverage and specify filename')
- parser.add_option('-r', '--result-port', type='int', help='connect to port on localhost and send test results')
- parser.add_option('--us', type='str', help='Directory to start discovery')
- parser.add_option('--up', type='str', help='Pattern to match test files (''test*.py'' default)')
- parser.add_option('--ut', type='str', help='Top level directory of project (default to start directory)')
- parser.add_option('--uvInt', '--verboseInt', type='int', help='Verbose output (0 none, 1 (no -v) simple, 2 (-v) full)')
- parser.add_option('--uf', '--failfast', type='str', help='Stop on first failure')
- parser.add_option('--uc', '--catch', type='str', help='Catch control-C and display results')
- (opts, _) = parser.parse_args()
-
- if opts.debug:
- from ptvsd.visualstudio_py_debugger import DONT_DEBUG, DEBUG_ENTRYPOINTS, get_code
-
- sys.path[0] = os.getcwd()
- if opts.result_port:
- try:
- signal.signal(signal.SIGUSR1, signal_handler)
- except:
- try:
- signal.signal(signal.SIGTERM, signal_handler)
- except:
- pass
- _channel = _IpcChannel(socket.create_connection(('127.0.0.1', opts.result_port)), stopTests)
- sys.stdout = _TestOutput(sys.stdout, is_stdout = True)
- sys.stderr = _TestOutput(sys.stderr, is_stdout = False)
-
- if opts.debug:
- DONT_DEBUG.append(os.path.normcase(__file__))
- DEBUG_ENTRYPOINTS.add(get_code(main))
-
- pass
- elif opts.mixed_mode:
- # For mixed-mode attach, there's no ptvsd and hence no wait_for_attach(),
- # so we have to use Win32 API in a loop to do the same thing.
- from time import sleep
- from ctypes import windll, c_char
- while True:
- if windll.kernel32.IsDebuggerPresent() != 0:
- break
- sleep(0.1)
- try:
- debugger_helper = windll['Microsoft.PythonTools.Debugger.Helper.x86.dll']
- except WindowsError:
- debugger_helper = windll['Microsoft.PythonTools.Debugger.Helper.x64.dll']
- isTracing = c_char.in_dll(debugger_helper, "isTracing")
- while True:
- if isTracing.value != 0:
- break
- sleep(0.1)
-
- cov = None
- try:
- if opts.coverage:
- try:
- import coverage
- cov = coverage.coverage(opts.coverage)
- cov.load()
- cov.start()
- except:
- pass
- if opts.tests is None and opts.testFile is None:
- if opts.us is None:
- opts.us = '.'
- if opts.up is None:
- opts.up = 'test*.py'
- tests = unittest.defaultTestLoader.discover(opts.us, opts.up)
- else:
- # loadTestsFromNames doesn't work well (with duplicate file names or class names)
- # Easier approach is find the test suite and use that for running
- loader = unittest.TestLoader()
- # opts.us will be passed in
- suites = loader.discover(opts.us, pattern=os.path.basename(opts.testFile))
- suite = None
- tests = None
- if opts.tests is None:
- # Run everything in the test file
- tests = suites
- else:
- # Run a specific test class or test method
- for test_suite in suites._tests:
- for cls in test_suite._tests:
- try:
- for m in cls._tests:
- testId = m.id()
- if testId.startswith(opts.tests[0]):
- suite = cls
- if testId == opts.tests[0]:
- tests = unittest.TestSuite([m])
- break
- except Exception as err:
- errorMessage = traceback.format_exception()
- pass
- if tests is None:
- tests = suite
- if tests is None and suite is None:
- _channel.send_event(
- name='error',
- outcome='',
- traceback = '',
- message = 'Failed to identify the test',
- test = ''
- )
- if opts.uvInt is None:
- opts.uvInt = 0
- if opts.uf is not None:
- runner = unittest.TextTestRunner(verbosity=opts.uvInt, resultclass=VsTestResult, failfast=True)
- else:
- runner = unittest.TextTestRunner(verbosity=opts.uvInt, resultclass=VsTestResult)
- result = runner.run(tests)
- if _channel is not None:
- _channel.close()
- sys.exit(not result.wasSuccessful())
- finally:
- if cov is not None:
- cov.stop()
- cov.save()
- cov.xml_report(outfile = opts.coverage + '.xml', omit=__file__)
- if _channel is not None:
- _channel.send_event(
- name='done'
- )
- _channel.socket.close()
- # prevent generation of the error 'Error in sys.exitfunc:'
- try:
- sys.stdout.close()
- except:
- pass
- try:
- sys.stderr.close()
- except:
- pass
-
-if __name__ == '__main__':
- main()
projects / dotfiles / .git / blobdiff