我是刚接触扭曲的人,我正尝试使用试用测试框架编写一些单元测试。我的测试按预期运行并通过,但是由于某些原因,测试之间挂起了试验。每次测试后,我必须按CTRL + C才能继续进行下一个测试。我猜我配置不正确,或者我没有打电话告诉我测试已经完成的某种方法。

这是要测试的课程:

from twisted.internet import reactor, defer
import threading
import time


class SomeClass:
   def doSomething(self):
      return self.asyncMethod()

   def asyncMethod(self):
       d = defer.Deferred()
       t = SomeThread(d)
       t.start()
       return d


class SomeThread(threading.Thread):
    def __init__(self, d):
        super(SomeThread, self).__init__()
        self.d = d

    def run(self):
        time.sleep(2) # pretend to do something
        retVal = 123
        self.d.callback(retVal)


这是单元测试类:

from twisted.trial import unittest
import tested


class SomeTest(unittest.TestCase):
    def testOne(self):
        sc = tested.SomeClass()
        d = sc.doSomething()
        return d.addCallback(self.allDone)

    def allDone(self, retVal):
        self.assertEquals(retVal, 123)

    def testTwo(self):
        sc = tested.SomeClass()
        d = sc.doSomething()
        return d.addCallback(self.allDone2)

    def allDone2(self, retVal):
        self.assertEquals(retVal, 123)


命令行输出如下所示:

me$ trial test.py
test
  SomeTest
    testOne ... ^C                                                           [OK]
    testTwo ... ^C                                                           [OK]

-------------------------------------------------------------------------------
Ran 2 tests in 8.499s

PASSED (successes=2)

最佳答案

我想您的问题与您的线程有关。 Twisted不是线程安全的,如果需要与线程接口,则应让反应堆通过使用deferToThreadcallInThreadcallFromThread处理事情。
有关如何使用Twisted进行线程安全的信息,请参见here

10-07 12:17