twistedでunittestをするときは、twisted.trialモジュールを利用する。 基本的にはbuilt-inのunittestと同じなのだが、deferredが存在するための仕組みが備わっている。 ここ にドキュメントがあるのだが若干分かりにくい。というか、結論を先に書いてくれ！
3つ大事なコトがある。
- deferredを返す関数をテストするtest methodはdeferredを返せ。
- timeoutをつかえ
- reactor.run()、reactor.stop()、reactor.crash()、reactor.iterateを呼ぶな
1)の原文は
The golden rule is: If your tests call a function which returns a Deferred, your test should return a Deferred.
である。callbackを勝手によんでくれます。もしかしたらcallLaterを使わないで値を渡す賢いやり方があるのかもしれません。
2)は、timeoutを設定することになる。これにはメソッドのアトリビュートを使用する。 これは下記のサンプルコードをみれば了解できると思われる。というか、ドキュメントは3行も書いているが、 なんでサンプルコードを載せないのか理解に苦しむ。
ここに書いたサンプルのコードを実行すると、2つのテストが成功し、最後の物がtimeoutする。
3)は守れ。理由はreactorのインスタンスの数を考えれば自明（trial自身がreactorを持っている)
#!/usr/bin/python
# -*- coding=utf8 -*-
'''
http://twistedmatrix.com/documents/current/core/howto/testing.html
how to run:
$ trial twisted
'''
from twisted.internet import defer, reactor
from twisted.trial.unittest import TestCase
# c.f. from unittest import TestCase
class MyTestCase(TestCase):
def test_with_deferred_ok(self):
d = defer.Deferred()
def fire(ignore):
return 'deadbeaf'
d.addCallback(fire)
def ok(s):
self.assertEquals(s, 'deadbeaf')
d.addCallback(ok)
def fail(s):
self.assertNotEquals(s, 'deadbeaf')
d.addErrback(fail)
#reactor.callLater(0, d.callback, 'deadbeaf')
return d
'''
The golden rule is: If your tests call a function which returns a Deferred, your test should return a Deferred.
'''
test_with_deferred_ok.timeout = 1
'''
The way to do this in Trial is to set the .timeout attribute on your unit test method.
Set the attribute to the number of seconds you wish to elapse before the test raises a
timeout error. Trial has a default timeout which will be applied even if the timeout
attribute is not set. The Trial default timeout is usually sufficient and
should be overridden only in unusual cases.
'''
def test_with_deferred_fail(self):
d = defer.Deferred()
def fire(ignore):
return 'moomoo'
d.addErrback(fire)
def ok(s):
self.assert_(False)
d.addCallback(ok)
def fail(s):
self.assertNotEquals(s, 'deadbeaf')
d.addErrback(fail)
return d
test_with_deferred_fail.timeout = 1
def test_with_deferred_timeout(self):
d = defer.Deferred()
def ok(s):
self.assert_(False)
d.addCallback(ok)
def fail(s):
self.assertNotEquals(s, 'deadbeaf')
d.addErrback(fail)
reactor.callLater(2, d.errback, 'moomoo')
return d
test_with_deferred_timeout.timeout = 1
投稿
