def main()

in pythonFiles/visualstudio_py_testlauncher.py [0:0]

• 167 lines of code
• 23 McCabe index (conditional complexity)

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()

    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.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 in opts.tests:
                                    if tests is None:
                                        tests = unittest.TestSuite([m])
                                    else:
                                        tests.addTest(m)
                        except Exception as err:
                            errorMessage = traceback.format_exc()
                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