Hello,
I have similar needs and did not found usable answers, so I wrote some
q&d hacks for doctests and also for UI interactions, where the user should
be blocked until something in non-blocking deferred finish or "time-outed".
Below hacks are working with twisted 10.0.0, no idea about more actual versions.
1. Multiple start/stop of reactor during doctests works when using
"reactor.crash"
Example - 2 tested class methods in one class:
# ---------------------
def method_a(self, parm):
'''
>>> g = GateWay('TEST')
>>> s1 = reactor.callWhenRunning(lambda: print(g.method_a('X') ) )
>>> s2 = reactor.callWhenRunning(lambda: print(g.method_a('Y') ) )
>>> s3 = reactor.callWhenRunning(lambda: print(g.method_a('Z') ) )
>>> s0 = reactor.callLater(11.1, reactor.crash); reactor.run() #
doctest:+ELLIPSIS
{u'result of X...}
{u'result of Y...}
{u'result of Z...}
'''
[code]
def method_b(self):
>>> g = GateWay('TEST')
>>> s1 = reactor.callWhenRunning(lambda: print(g.method_b() ) )
>>> s0 = reactor.callLater(5.1, reactor.crash); reactor.run() #
doctest:+ELLIPSIS
{u'bytes': ..., u'packets': ..., u'duration': ...}
'''
[code]
# ---------------------
This way, it is possible to use doctests with reactor start/stop in several
methods/functions in a class/module and also in external doctests files.
2. Sleep w/o blocking reactor:
# ---------------------
def nonBlockingSleep(delay):
'''Hack to allow sleep for a period of time w/o blocking the reactor
Do *block* the caller for `delay` seconds.
'''
if delay > 90:
print('WARNING nonBlockingSleep delay={0} sec. dangerous, reset to
90s'.format(delay) )
delay = 90.0
toTime = time.time() + delay
while toTime >= time.time():
try:
reactor.iterate(0.055)
except Exception as err:
pass
time.sleep(0.001)
# ---------------------
The values of .iterate() and .sleep() are the best, I found.
time.sleep(0.001) is important, it is not reliable w/o + CPU load is high.
3. Wait until deferred return a result or timeout expire
# ---------------------
def sleepUntilDFRDresult(dfrd, timeout=60):
'''Hack to make non-blocking deferred waiting for its result or error for
in `timeout` specified number of seconds, but not to block reactor loop.
This wrapper *block* the caller until deferred.result or timeout appears.
>>> dfrd = defer.Deferred() #.addBoth(cbReactorStop)
>>> sleepUntilDFRDresult(dfrd) # doctest:+ELLIPSIS
Exception(u'ERROR sleepUntilDFRDresult server not running for: <Deferred at
0x...> from <doctest
__main__/<module> #L1',)
>>> startTime = time.time()
>>> f = lambda: print(sleepUntilDFRDresult(dfrd, timeout=999), 'in:',
time.time()-startTime,
'sec.' )
>>> s1 = reactor.callWhenRunning( f );
>>> dc2 = reactor.callLater(2, dfrd.callback, 'CALLED')
>>> dc0 = reactor.callLater(3, reactor.crash); reactor.run() #
doctest:+ELLIPSIS
WARNING sleepUntilDFRDresult timeout=999 sec. dangerous
CALLED in: 2.0... sec.
'''
if timeout > 600:
print('WARNING sleepUntilDFRDresult timeout={0} sec.
dangerous'.format(timeout) )
err = None
if not isinstance(dfrd, defer.Deferred):
err = 'ERROR sleepUntilDFRDresult dfrd "{0}" not Deferred:
'.format(type(dfrd))
elif not reactor.running:
err = 'ERROR sleepUntilDFRDresult server not running for: '
else:
toTime = time.time() + timeout
while not dfrd.called and toTime >= time.time():
try:
reactor.iterate(0.055)
except Exception as err:
pass
Htime.sleep(0.001)
if dfrd.called and hasattr(dfrd, 'result'):
return dfrd.result
func_name, module_name, line_no = tools.getCallerNameLocation()
repr_dfrd = '{0} from {2}/{1} #L{3}'.format(dfrd, func_name, module_name,
line_no)
err = Exception( (err or 'ERROR TIMEOUT sleepUntilDFRDresult for: ') +
repr_dfrd )
## print(err)
return err
# ---------------------
This hack is not 100% reliable, but I did not found a better solution.
There is sometime an "Already called..." error in the logs, when server is under
high load with lot of connections, but since around 2 years I did not observed
a data loss or other severer problem.
I hope it helps.
If somebody have a better (less "hacky") solution for above 3 tasks,
please send it to this list.
Regards
Paul
Laurens Van Houtven wrote:
> The thing I'm documenting is a server, I'm documenting it by interacting with
> it as a client. I realize that doesn't entirely detract from your point --
> you might still be introducing problems that would not affect a "real" client.
>
> My intention is to write BDD-ish stuff (except not with the usual
> cucumber-style scenario text).
>
> cheers
> lvh
>
>
> On 25 Mar 2012, at 23:44, Itamar Turner-Trauring wrote:
>
>> On 03/25/2012 05:02 PM, Laurens Van Houtven wrote:
>>> Hi,
>>>
>>>
>>> I'm trying to find out if there's a reasonable way to mimic a blocking
>>> API with an existing non-blocking API. I want to do this so I can
>>> write doctests.
>>>
>>> For example, I want to make a remote AMP call. It returns a deferred.
>>> Instead of returning a deferred, I want it to block until the deferred
>>> fires.
>> Doctests are supposed to document and test, or perhaps have tested
>> documentation. If the API you're presenting doesn't match actual usage,
>> that's bad documentation...
_______________________________________________
Twisted-Python mailing list
Twisted-Python@twistedmatrix.com
http://twistedmatrix.com/cgi-bin/mailman/listinfo/twisted-python
