1 import os, signal, sys, time
2 from random import randrange
4 from twisted.internet import reactor, defer
5 from twisted.python import failure
7 from allmydata.util import fileutil, log
8 from allmydata.util.encodingutil import unicode_platform, get_filesystem_encoding
11 def insecurerandstr(n):
12 return ''.join(map(chr, map(randrange, [0]*n, [256]*n)))
14 def flip_bit(good, which):
15 # flip the low-order bit of good[which]
17 pieces = good[:which], good[-1:], ""
19 pieces = good[:which], good[which:which+1], good[which+1:]
20 return pieces[0] + chr(ord(pieces[1]) ^ 0x01) + pieces[2]
22 def flip_one_bit(s, offset=0, size=None):
23 """ flip one random bit of the string s, in a byte greater than or equal to offset and less
27 i = randrange(offset, offset+size)
28 result = s[:i] + chr(ord(s[i])^(0x01<<randrange(0, 8))) + s[i+1:]
29 assert result != s, "Internal error -- flip_one_bit() produced the same string as its input: %s == %s" % (result, s)
33 class ReallyEqualMixin:
34 def failUnlessReallyEqual(self, a, b, msg=None):
35 self.failUnlessEqual(a, b, msg=msg)
36 self.failUnlessEqual(type(a), type(b), msg="a :: %r, b :: %r, %r" % (a, b, msg))
39 class NonASCIIPathMixin:
40 def mkdir_nonascii(self, dirpath):
41 # Kludge to work around the fact that buildbot can't remove a directory tree that has
42 # any non-ASCII directory names on Windows. (#1472)
43 if sys.platform == "win32":
46 fileutil.rm_dir(dirpath)
48 log.err("We were unable to delete a non-ASCII directory %r created by the test. "
49 "This is liable to cause failures on future builds." % (dirpath,))
50 self.addCleanup(self._cleanup, dirpath)
53 def unicode_or_fallback(self, unicode_name, fallback_name):
54 if unicode_platform():
57 unicode_name.encode(get_filesystem_encoding())
59 except UnicodeEncodeError:
64 # This class is necessary for any code which wants to use Processes
65 # outside the usual reactor.run() environment. It is copied from
66 # Twisted's twisted.test.test_process . Note that Twisted-8.2.0 uses
67 # something rather different.
71 # make sure SIGCHLD handler is installed, as it should be on
72 # reactor.run(). problem is reactor may not have been run when this
74 if hasattr(reactor, "_handleSigchld") and hasattr(signal, "SIGCHLD"):
75 self.sigchldHandler = signal.signal(signal.SIGCHLD,
76 reactor._handleSigchld)
79 if self.sigchldHandler:
80 signal.signal(signal.SIGCHLD, self.sigchldHandler)
83 def stall(self, res=None, delay=1):
85 reactor.callLater(delay, d.callback, res)
88 class ShouldFailMixin:
90 def shouldFail(self, expected_failure, which, substring,
91 callable, *args, **kwargs):
92 assert substring is None or isinstance(substring, str)
93 d = defer.maybeDeferred(callable, *args, **kwargs)
95 if isinstance(res, failure.Failure):
96 res.trap(expected_failure)
98 self.failUnless(substring in str(res),
99 "%s: substring '%s' not in '%s'"
100 % (which, substring, str(res)))
101 # return the Failure for further analysis, but in a form that
102 # doesn't make the Deferred chain think that we failed.
105 self.fail("%s was supposed to raise %s, not get '%s'" %
106 (which, expected_failure, res))
111 class TestMixin(SignalMixin):
112 def setUp(self, repeatable=False):
114 @param repeatable: install the repeatable_randomness hacks to attempt
115 to without access to real randomness and real time.time from the
118 SignalMixin.setUp(self)
119 self.repeatable = repeatable
121 import repeatable_random
122 repeatable_random.force_repeatability()
123 if hasattr(time, 'realtime'):
124 self.teststarttime = time.realtime()
126 self.teststarttime = time.time()
129 SignalMixin.tearDown(self)
131 import repeatable_random
132 repeatable_random.restore_non_repeatability()
133 self.clean_pending(required_to_quiesce=True)
135 def clean_pending(self, dummy=None, required_to_quiesce=True):
137 This handy method cleans all pending tasks from the reactor.
139 When writing a unit test, consider the following question:
141 Is the code that you are testing required to release control once it
142 has done its job, so that it is impossible for it to later come around
143 (with a delayed reactor task) and do anything further?
145 If so, then trial will usefully test that for you -- if the code under
146 test leaves any pending tasks on the reactor then trial will fail it.
148 On the other hand, some code is *not* required to release control -- some
149 code is allowed to continuously maintain control by rescheduling reactor
150 tasks in order to do ongoing work. Trial will incorrectly require that
151 code to clean up all its tasks from the reactor.
153 Most people think that such code should be amended to have an optional
154 "shutdown" operation that releases all control, but on the contrary it is
155 good design for some code to *not* have a shutdown operation, but instead
156 to have a "crash-only" design in which it recovers from crash on startup.
158 If the code under test is of the "long-running" kind, which is *not*
159 required to shutdown cleanly in order to pass tests, then you can simply
160 call testutil.clean_pending() at the end of the unit test, and trial will
163 pending = reactor.getDelayedCalls()
164 active = bool(pending)
169 print "WEIRDNESS! pending timed call not active!"
170 if required_to_quiesce and active:
171 self.fail("Reactor was still active when it was required to be quiescent.")
176 def make_readonly(path):
177 win32file.SetFileAttributes(path, win32con.FILE_ATTRIBUTE_READONLY)
178 def make_accessible(path):
179 win32file.SetFileAttributes(path, win32con.FILE_ATTRIBUTE_NORMAL)
182 def _make_readonly(path):
183 os.chmod(path, stat.S_IREAD)
184 os.chmod(os.path.dirname(path), stat.S_IREAD)
185 def _make_accessible(path):
186 os.chmod(os.path.dirname(path), stat.S_IWRITE | stat.S_IEXEC | stat.S_IREAD)
187 os.chmod(path, stat.S_IWRITE | stat.S_IEXEC | stat.S_IREAD)
188 make_readonly = _make_readonly
189 make_accessible = _make_accessible