def tearDown(self):
     # indicate that the client is finished
     self.client_evt.set()
     # wait for the server thread to terminate
     self.serv_evt.wait()
     self.thread.join()
     test_support.threading_cleanup(*self._threads)

def test_main():
    threads = []
    thread_info = threading_setup()

    print "Creating"
    for i in range(NUM_THREADS):
        t = TempFileGreedy()
        threads.append(t)
        t.start()

    print "Starting"
    startEvent.set()

    print "Reaping"
    ok = errors = 0
    for t in threads:
        t.join()
        ok += t.ok_count
        errors += t.error_count
        if t.error_count:
            print '%s errors:\n%s' % (t.getName(), t.errors.getvalue())

    msg = "Done: errors %d ok %d" % (errors, ok)
    print msg
    if errors:
        raise TestFailed(msg)

    threading_cleanup(*thread_info)

    def test_main(self):
        threads = []
        thread_info = threading_setup()

        for i in range(NUM_THREADS):
            t = TempFileGreedy()
            threads.append(t)
            t.start()

        startEvent.set()

        ok = 0
        errors = []
        for t in threads:
            t.join()
            ok += t.ok_count
            if t.error_count:
                errors.append(str(t.getName()) + str(t.errors.getvalue()))

        threading_cleanup(*thread_info)

        msg = "Errors: errors %d ok %d\n%s" % (len(errors), ok,
            '\n'.join(errors))
        self.assertEquals(errors, [], msg)
        self.assertEquals(ok, NUM_THREADS * FILES_PER_THREAD)

def test_main():
    tests = [TestPOP3Class, TestTimeouts, TestPOP3_SSLClass]
    thread_info = test_support.threading_setup()
    try:
        test_support.run_unittest(*tests)
    finally:
        test_support.threading_cleanup(*thread_info)

def test_main(verbose=False):
    if skip_expected:
        raise test_support.TestSkipped("No SSL support")

    global CERTFILE, SVN_PYTHON_ORG_ROOT_CERT
    CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "keycert.pem")
    SVN_PYTHON_ORG_ROOT_CERT = os.path.join(os.path.dirname(__file__) or os.curdir, "https_svn_python_org_root.pem")

    if not os.path.exists(CERTFILE) or not os.path.exists(SVN_PYTHON_ORG_ROOT_CERT):
        raise test_support.TestFailed("Can't read certificate files!")

    TESTPORT = test_support.find_unused_port()
    if not TESTPORT:
        raise test_support.TestFailed("Can't find open port to test servers on!")

    tests = [BasicTests]

    if test_support.is_resource_enabled("network"):
        tests.append(NetworkedTests)

    if _have_threads:
        thread_info = test_support.threading_setup()
        if thread_info and test_support.is_resource_enabled("network"):
            tests.append(ThreadedTests)

    test_support.run_unittest(*tests)

    if _have_threads:
        test_support.threading_cleanup(*thread_info)

def test_main():
    tests = [GeneralModuleTests, BasicTCPTest, TCPCloserTest, TCPTimeoutTest,
             TestExceptions, BufferIOTest, BasicTCPTest2]
    if sys.platform != 'mac':
        tests.extend([ BasicUDPTest, UDPTimeoutTest ])

    tests.extend([
        NonBlockingTCPTests,
        FileObjectClassTestCase,
        UnbufferedFileObjectClassTestCase,
        LineBufferedFileObjectClassTestCase,
        SmallBufferedFileObjectClassTestCase,
        Urllib2FileobjectTest,
        NetworkConnectionNoServer,
        NetworkConnectionAttributesTest,
        NetworkConnectionBehaviourTest,
    ])
    if hasattr(socket, "socketpair"):
        tests.append(BasicSocketPairTest)
    if sys.platform == 'linux2':
        tests.append(TestLinuxAbstractNamespace)
    if isTipcAvailable():
        tests.append(TIPCTest)
        tests.append(TIPCThreadableTest)

    thread_info = test_support.threading_setup()
    test_support.run_unittest(*tests)
    test_support.threading_cleanup(*thread_info)

    def tearDown(self):
        self.client.close()

        self.evt.wait()

        # Disable server feedback
        DocXMLRPCServer._send_traceback_header = False
        test_support.threading_cleanup(*self._threads)

def test_main():
    tests = [TestFTPClass, TestTimeouts, TestIPv6Environment, TestTLS_FTPClassMixin, TestTLS_FTPClass]

    thread_info = test_support.threading_setup()
    try:
        test_support.run_unittest(*tests)
    finally:
        test_support.threading_cleanup(*thread_info)

def test_main():
    tests = [TestPOP3Class, TestTimeouts]
    if SUPPORTS_SSL and not test_support.due_to_ironpython_bug("http://www.codeplex.com/IronPython/WorkItem/View.aspx?WorkItemId=28750"):
        tests.append(TestPOP3_SSLClass)
    thread_info = test_support.threading_setup()
    try:
        test_support.run_unittest(*tests)
    finally:
        test_support.threading_cleanup(*thread_info)

def test_main():
    tests = [TestPOP3Class, TestTimeouts]
    if SUPPORTS_SSL:
        tests.append(TestPOP3_SSLClass)
    thread_info = test_support.threading_setup()
    try:
        test_support.run_unittest(*tests)
    finally:
        test_support.threading_cleanup(*thread_info)

 def tearDown(self):
     try:
         os.remove(self.filename)
     except EnvironmentError as ee:
         # (Jython addition) detect failure common on Windows, on missing
         # close, that creates spurious errors in subsequent tests.
         if ee.errno != errno.ENOENT:
             raise ee
     test_support.threading_cleanup(*self._threads)

 def tearDown(self):
     if self.f:
         try:
             self.f.close()
         except (EnvironmentError, ValueError):
             pass
     try:
         os.remove(self.filename)
     except EnvironmentError:
         pass
     test_support.threading_cleanup(*self._threads)

def test_main():
    tests = [TestImaplib]

    if support.is_resource_enabled('network'):
        if ssl:
            global CERTFILE
            CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,
                                    "keycert.pem")
            if not os.path.exists(CERTFILE):
                raise support.TestFailed("Can't read certificate files!")
            tests.append(ThreadedNetworkedTestsSSL)
        tests.append(ThreadedNetworkedTests)

    threadinfo = support.threading_setup()

    support.run_unittest(*tests)

    support.threading_cleanup(*threadinfo)

def test_main():
    tests = [GeneralModuleTests, BasicTCPTest, TCPCloserTest, TCPTimeoutTest,
             TestExceptions, BufferIOTest]
    if sys.platform != 'mac':
        tests.extend([ BasicUDPTest, UDPTimeoutTest ])

    tests.extend([
        NonBlockingTCPTests,
        FileObjectClassTestCase,
        UnbufferedFileObjectClassTestCase,
        LineBufferedFileObjectClassTestCase,
        SmallBufferedFileObjectClassTestCase
    ])
    if hasattr(socket, "socketpair"):
        tests.append(BasicSocketPairTest)
    if sys.platform == 'linux2':
        tests.append(TestLinuxAbstractNamespace)

    thread_info = test_support.threading_setup()
    test_support.run_unittest(*tests)
    test_support.threading_cleanup(*thread_info)

 def decorator(*args):
     key = test_support.threading_setup()
     try:
         return func(*args)
     finally:
         test_support.threading_cleanup(*key)

 def tearDown(self):
     self.thread.stop()
     os.environ.__exit__()
     test_support.threading_cleanup(*self._threads)

 def tearDown(self):
     test_support.threading_cleanup(*self._threads)

 def tearDown(self):
     self.doCleanups()
     test_support.threading_cleanup(*self._threads)

 def tearDown(self):
     support.threading_cleanup(*self._threads)
     support.reap_children()