1 import os, signal, time
2 from random import randrange
4 from twisted.internet import reactor, defer
5 from twisted.python import failure
7 def insecurerandstr(n):
8 return ''.join(map(chr, map(randrange, [0]*n, [256]*n)))
10 def flip_bit(good, which):
11 # flip the low-order bit of good[which]
13 pieces = good[:which], good[-1:], ""
15 pieces = good[:which], good[which:which+1], good[which+1:]
16 return pieces[0] + chr(ord(pieces[1]) ^ 0x01) + pieces[2]
18 def flip_one_bit(s, offset=0, size=None):
19 """ flip one random bit of the string s, in a byte greater than or equal to offset and less
23 i = randrange(offset, offset+size)
24 result = s[:i] + chr(ord(s[i])^(0x01<<randrange(0, 8))) + s[i+1:]
25 assert result != s, "Internal error -- flip_one_bit() produced the same string as its input: %s == %s" % (result, s)
29 # This class is necessary for any code which wants to use Processes
30 # outside the usual reactor.run() environment. It is copied from
31 # Twisted's twisted.test.test_process
35 # make sure SIGCHLD handler is installed, as it should be on
36 # reactor.run(). problem is reactor may not have been run when this
38 if hasattr(reactor, "_handleSigchld") and hasattr(signal, "SIGCHLD"):
39 self.sigchldHandler = signal.signal(signal.SIGCHLD,
40 reactor._handleSigchld)
42 def tearDownClass(self):
43 if self.sigchldHandler:
44 signal.signal(signal.SIGCHLD, self.sigchldHandler)
47 def stall(self, res=None, delay=1):
49 reactor.callLater(delay, d.callback, res)
52 class ShouldFailMixin:
54 def shouldFail(self, expected_failure, which, substring,
55 callable, *args, **kwargs):
56 assert substring is None or isinstance(substring, str)
57 d = defer.maybeDeferred(callable, *args, **kwargs)
59 if isinstance(res, failure.Failure):
60 res.trap(expected_failure)
62 self.failUnless(substring in str(res),
63 "%s: substring '%s' not in '%s'"
64 % (which, substring, str(res)))
65 # return the Failure for further analysis, but in a form that
66 # doesn't make the Deferred chain think that we failed.
69 self.fail("%s was supposed to raise %s, not get '%s'" %
70 (which, expected_failure, res))
75 class TestMixin(SignalMixin):
76 def setUp(self, repeatable=False):
78 @param repeatable: install the repeatable_randomness hacks to attempt
79 to without access to real randomness and real time.time from the
82 self.repeatable = repeatable
84 import repeatable_random
85 repeatable_random.force_repeatability()
86 if hasattr(time, 'realtime'):
87 self.teststarttime = time.realtime()
89 self.teststarttime = time.time()
93 import repeatable_random
94 repeatable_random.restore_non_repeatability()
95 self.clean_pending(required_to_quiesce=True)
97 def clean_pending(self, dummy=None, required_to_quiesce=True):
99 This handy method cleans all pending tasks from the reactor.
101 When writing a unit test, consider the following question:
103 Is the code that you are testing required to release control once it
104 has done its job, so that it is impossible for it to later come around
105 (with a delayed reactor task) and do anything further?
107 If so, then trial will usefully test that for you -- if the code under
108 test leaves any pending tasks on the reactor then trial will fail it.
110 On the other hand, some code is *not* required to release control -- some
111 code is allowed to continuously maintain control by rescheduling reactor
112 tasks in order to do ongoing work. Trial will incorrectly require that
113 code to clean up all its tasks from the reactor.
115 Most people think that such code should be amended to have an optional
116 "shutdown" operation that releases all control, but on the contrary it is
117 good design for some code to *not* have a shutdown operation, but instead
118 to have a "crash-only" design in which it recovers from crash on startup.
120 If the code under test is of the "long-running" kind, which is *not*
121 required to shutdown cleanly in order to pass tests, then you can simply
122 call testutil.clean_pending() at the end of the unit test, and trial will
125 pending = reactor.getDelayedCalls()
126 active = bool(pending)
131 print "WEIRNESS! pending timed call not active+!"
132 if required_to_quiesce and active:
133 self.fail("Reactor was still active when it was required to be quiescent.")
138 def make_readonly(path):
139 win32file.SetFileAttributes(path, win32con.FILE_ATTRIBUTE_READONLY)
140 def make_accessible(path):
141 win32file.SetFileAttributes(path, win32con.FILE_ATTRIBUTE_NORMAL)
144 def _make_readonly(path):
145 os.chmod(path, stat.S_IREAD)
146 os.chmod(os.path.dirname(path), stat.S_IREAD)
147 def _make_accessible(path):
148 os.chmod(os.path.dirname(path), stat.S_IWRITE | stat.S_IEXEC | stat.S_IREAD)
149 os.chmod(path, stat.S_IWRITE | stat.S_IEXEC | stat.S_IREAD)
150 make_readonly = _make_readonly
151 make_accessible = _make_accessible