From: Zooko O'Whielacronx <zooko@zooko.com> Date: Thu, 19 Apr 2007 00:28:36 +0000 (-0700) Subject: copy testutil from pyutil X-Git-Tag: tahoe_v0.1.0-0-UNSTABLE~58 X-Git-Url: https://git.rkrishnan.org/specifications/components/com_hotproperty/flags?a=commitdiff_plain;h=47bdfdf4198c724f73dbd1d38d1248bd2607b0c2;p=tahoe-lafs%2Ftahoe-lafs.git copy testutil from pyutil --- diff --git a/src/allmydata/util/testutil.py b/src/allmydata/util/testutil.py new file mode 100644 index 00000000..72595c1f --- /dev/null +++ b/src/allmydata/util/testutil.py @@ -0,0 +1,98 @@ +import os, signal, sys, time + +from twisted.internet import reactor +from twisted.trial import unittest + +class SignalMixin(unittest.TestCase): + # This class is necessary for any code which wants to use Processes + # outside the usual reactor.run() environment. It is copied from + # Twisted's twisted.test.test_process + sigchldHandler = None + + def setUpClass(self): + # make sure SIGCHLD handler is installed, as it should be on + # reactor.run(). problem is reactor may not have been run when this + # test runs. + if hasattr(reactor, "_handleSigchld") and hasattr(signal, "SIGCHLD"): + self.sigchldHandler = signal.signal(signal.SIGCHLD, + reactor._handleSigchld) + + def tearDownClass(self): + if self.sigchldHandler: + signal.signal(signal.SIGCHLD, self.sigchldHandler) + +class TestMixin(SignalMixin): + def setUp(self, repeatable=False): + """ + @param repeatable: install the repeatable_randomness hacks to attempt + to without access to real randomness and real time.time from the + code under test + """ + self.repeatable = repeatable + if self.repeatable: + import repeatable_random + repeatable_random.force_repeatability() + if hasattr(time, 'realtime'): + self.teststarttime = time.realtime() + else: + self.teststarttime = time.time() + + def tearDown(self): + if self.repeatable: + repeatable_random.restore_non_repeatability() + self.clean_pending(required_to_quiesce=True) + + def clean_pending(self, dummy=None, required_to_quiesce=True): + """ + This handy method cleans all pending tasks from the reactor. + + When writing a unit test, consider the following question: + + Is the code that you are testing required to release control once it + has done its job, so that it is impossible for it to later come around + (with a delayed reactor task) and do anything further? + + If so, then trial will usefully test that for you -- if the code under + test leaves any pending tasks on the reactor then trial will fail it. + + On the other hand, some code is *not* required to release control -- some + code is allowed to continuously maintain control by rescheduling reactor + tasks in order to do ongoing work. Trial will incorrectly require that + code to clean up all its tasks from the reactor. + + Most people think that such code should be amended to have an optional + "shutdown" operation that releases all control, but on the contrary it is + good design for some code to *not* have a shutdown operation, but instead + to have a "crash-only" design in which it recovers from crash on startup. + + If the code under test is of the "long-running" kind, which is *not* + required to shutdown cleanly in order to pass tests, then you can simply + call testutil.clean_pending() at the end of the unit test, and trial will + be satisfied. + """ + pending = reactor.getDelayedCalls() + active = bool(pending) + for p in pending: + if p.active(): + p.cancel() + else: + print "WEIRNESS! pending timed call not active+!" + if required_to_quiesce and active: + self.fail("Reactor was still active when it was required to be quiescent.") + +if sys.platform == 'win32': + import win32file + import win32con + def make_readonly(path): + win32file.SetFileAttributes(path, win32con.FILE_ATTRIBUTE_READONLY) + def make_accessible(path): + win32file.SetFileAttributes(path, win32con.FILE_ATTRIBUTE_NORMAL) +else: + import stat + def make_readonly(path): + os.chmod(path, stat.S_IREAD) + os.chmod(os.path.dirname(path), stat.S_IREAD) + def make_accessible(path): + os.chmod(os.path.dirname(path), stat.S_IWRITE | stat.S_IEXEC | stat.S_IREAD) + os.chmod(path, stat.S_IWRITE | stat.S_IEXEC | stat.S_IREAD) +