From f70d357b512a155b9e5b8807755f17805d32f839 Mon Sep 17 00:00:00 2001
From: Jeremy Hylton <jeremy@svn.zope.org>
Date: Fri, 9 Aug 2002 20:31:45 +0000
Subject: [PATCH] Add TestThread class that helps manage threads started by a
unittest. Use TestThread as base class for WorkerThread in CommitLockTests.
---
src/ZEO/tests/CommitLockTests.py | 18 ++++++-------
src/ZEO/tests/TestThread.py | 43 ++++++++++++++++++++++++++++++++
2 files changed, 50 insertions(+), 11 deletions(-)
create mode 100644 src/ZEO/tests/TestThread.py
diff --git a/src/ZEO/tests/CommitLockTests.py b/src/ZEO/tests/CommitLockTests.py
index abc23a40..0ea3fb39 100644
--- a/src/ZEO/tests/CommitLockTests.py
+++ b/src/ZEO/tests/CommitLockTests.py
@@ -13,13 +13,12 @@
##############################################################################
"""Tests of the distributed commit lock."""
-import threading
-
from ZODB.Transaction import Transaction
from ZODB.tests.StorageTestBase import zodb_pickle, MinPO
import ZEO.ClientStorage
from ZEO.Exceptions import Disconnected
+from ZEO.tests.TestThread import TestThread
ZERO = '\0'*8
@@ -27,19 +26,18 @@ class DummyDB:
def invalidate(self, *args):
pass
-class WorkerThread(threading.Thread):
+class WorkerThread(TestThread):
# run the entire test in a thread so that the blocking call for
# tpc_vote() doesn't hang the test suite.
- def __init__(self, storage, trans, method="tpc_finish"):
+ def __init__(self, testcase, storage, trans, method="tpc_finish"):
self.storage = storage
self.trans = trans
self.method = method
- threading.Thread.__init__(self)
- self.setDaemon(1)
+ TestThread.__init__(self, testcase)
- def run(self):
+ def testrun(self):
try:
self.storage.tpc_begin(self.trans)
oid = self.storage.new_oid()
@@ -151,15 +149,13 @@ class CommitLockTests:
def _dosetup2(self, storage, trans, tid):
self._threads = []
- t = WorkerThread(storage, trans)
+ t = WorkerThread(self, storage, trans)
self._threads.append(t)
t.start()
def _dowork2(self, method_name):
for t in self._threads:
- t.join(10)
- for t in self._threads:
- self.failIf(t.isAlive())
+ t.cleanup()
def _duplicate_client(self):
"Open another ClientStorage to the same server."
diff --git a/src/ZEO/tests/TestThread.py b/src/ZEO/tests/TestThread.py
new file mode 100644
index 00000000..4eaae2a1
--- /dev/null
+++ b/src/ZEO/tests/TestThread.py
@@ -0,0 +1,43 @@
+##############################################################################
+#
+# Copyright (c) 2002 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE
+#
+##############################################################################
+"""A Thread base class for use with unittest."""
+
+from cStringIO import StringIO
+import threading
+import traceback
+
+class TestThread(threading.Thread):
+ __super_init = threading.Thread.__init__
+ __super_run = threading.Thread.run
+
+ def __init__(self, testcase, group=None, target=None, name=None,
+ args=(), kwargs={}, verbose=None):
+ self.__super_init(group, target, name, args, kwargs, verbose)
+ self.setDaemon(1)
+ self._testcase = testcase
+
+ def run(self):
+ try:
+ self.testrun()
+ except Exception, err:
+ s = StringIO()
+ traceback.print_exc(file=s)
+ self._testcase.fail("Exception in thread %s:\n%s\n" %
+ (self, s.getvalue()))
+
+ def cleanup(self, timeout=15):
+ self.join(timeout)
+ if self.isAlive():
+ self._testcase.fail("Thread did not finish: %s" % self)
+
--
2.30.9