本文整理汇总了Python中test.test_support.threading_setup函数的典型用法代码示例。如果您正苦于以下问题:Python threading_setup函数的具体用法?Python threading_setup怎么用?Python threading_setup使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了threading_setup函数的18个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_main
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)
开发者ID:Alex-CS,项目名称:sonify,代码行数:28,代码来源:test_threadedtempfile.py
示例2: test_main
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)
开发者ID:cimarieta,项目名称:usp,代码行数:7,代码来源:test_poplib.py
示例3: test_main
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)
开发者ID:Cinnz,项目名称:python,代码行数:29,代码来源:test_ssl.py
示例4: test_main
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)
开发者ID:1310701102,项目名称:sl4a,代码行数:25,代码来源:test_threadedtempfile.py
示例5: test_main
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)
开发者ID:1310701102,项目名称:sl4a,代码行数:28,代码来源:test_socket.py
示例6: setUp
def setUp(self):
self._threads = test_support.threading_setup()
os.environ = test_support.EnvironmentVarGuard()
self.server_started = threading.Event()
self.thread = TestServerThread(self, self.request_handler)
self.thread.start()
self.server_started.wait()
开发者ID:Darriall,项目名称:pypy,代码行数:7,代码来源:test_httpservers.py
示例7: test_main
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)
开发者ID:cimarieta,项目名称:usp,代码行数:8,代码来源:test_ftplib.py
示例8: test_main
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)
开发者ID:49476291,项目名称:android-plus-plus,代码行数:9,代码来源:test_poplib.py
示例9: test_main
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)
开发者ID:BillyboyD,项目名称:main,代码行数:9,代码来源:test_poplib.py
示例10: setUp
def setUp(self):
self._threads = test_support.threading_setup()
self.f = None
self.filename = TESTFN
with open(self.filename, "w") as f:
f.write("\n".join("0123456789"))
self._count_lock = threading.Lock()
self.close_count = 0
self.close_success_count = 0
self.use_buffering = False
开发者ID:davidtrem,项目名称:activepapers-python,代码行数:10,代码来源:test_internal_files.py
示例11: setUp
def setUp(self):
self._threads = test_support.threading_setup()
self.evt = threading.Event()
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(15)
self.port = test_support.bind_port(self.sock)
servargs = (self.evt, "220 Hola mundo\n", self.sock)
self.thread = threading.Thread(target=server, args=servargs)
self.thread.start()
self.evt.wait()
self.evt.clear()
开发者ID:billtsay,项目名称:win-demo-opcua,代码行数:11,代码来源:test_smtplib.py
示例12: setUp
def setUp(self):
self.done_mutex = thread.allocate_lock()
self.done_mutex.acquire()
self.running_mutex = thread.allocate_lock()
self.random_mutex = thread.allocate_lock()
self.created = 0
self.running = 0
self.next_ident = 0
key = test_support.threading_setup()
self.addCleanup(test_support.threading_cleanup, *key)
开发者ID:IronLanguages,项目名称:ironpython2,代码行数:11,代码来源:test_thread.py
示例13: setUp
def setUp(self):
self._threads = test_support.threading_setup()
self.f = None
self.filename = TESTFN
with open(self.filename, "w") as f:
f.write("\n".join("0123456789"))
self._count_lock = threading.Lock()
self.close_count = 0
self.close_success_count = 0
self.use_buffering = False
# to prevent running out of file descriptors on PyPy,
# we only keep the 50 most recent files open
self.all_files = [None] * 50
开发者ID:Alkalit,项目名称:pypyjs,代码行数:13,代码来源:test_file2k.py
示例14: setUp
def setUp(self):
self._threads = test_support.threading_setup()
# Enable server feedback
DocXMLRPCServer._send_traceback_header = True
self.evt = threading.Event()
threading.Thread(target=server, args=(self.evt, 1)).start()
# wait for port to be assigned
n = 1000
while n > 0 and PORT is None:
time.sleep(0.001)
n -= 1
self.client = httplib.HTTPConnection("localhost:%d" % PORT)
开发者ID:billtsay,项目名称:win-demo-opcua,代码行数:15,代码来源:test_docxmlrpc.py
示例15: test_main
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)
开发者ID:AojiaoZero,项目名称:CrossApp,代码行数:18,代码来源:test_imaplib.py
示例16: test_main
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)
开发者ID:avenrainbow,项目名称:aven_su,代码行数:21,代码来源:test_socket.py
示例17: setUp
def setUp(self):
self._threads = support.threading_setup()
开发者ID:Jesus00792,项目名称:Maganews,代码行数:2,代码来源:lock_tests.py
示例18: decorator
def decorator(*args):
key = test_support.threading_setup()
try:
return func(*args)
finally:
test_support.threading_cleanup(*key)
开发者ID:wujuguang,项目名称:pythonfutures,代码行数:6,代码来源:test_futures.py
注:本文中的test.test_support.threading_setup函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论