当使用twisted.trial.unittest时,反应器在两次测试之间停止



有没有一种方法可以在测试之间保持在线连接?

我试图在运行多个测试集时记录用户,但注意到了意外的行为:每次测试后反应器都会停止(所以一旦建立的连接就无法使用)。

为了探索这一点,我准备了示例测试文件(如下)。

这将创建简单的侦听echo服务器,并将其存储在类字段中。进行了3次测试。服务器应在所有测试期间正常运行(实际上它只运行到第一次测试结束)。

示例:

#! /usr/bin/python
# -*- coding: utf-8 -*-
import logging
from twisted.python import log
from twisted.internet import defer, base, reactor
from twisted.trial import unittest
from twisted.internet.protocol import Protocol, Factory
from twisted.internet.endpoints import TCP4ServerEndpoint
observer = log.PythonLoggingObserver()
observer.start()
logging.basicConfig(level=logging.DEBUG)
class Echo(Protocol):
    '''Protocol from twistedmatrix examples'''
    def connectionMade(self):
        log.msg('Got incomming connection')
        self.transport.write("An apple a day keeps the doctor awayrn")
    def connectionLost(self, reason):
        log.msg('Connection lost due to: %s' % reason)
    def dataReceived(self, data):
        self.transport.write(data)
        log.msg('Got some data: %s' % data)

class EchoFactory(Factory):
    '''Factory from twistedmatrix examples'''
    def buildProtocol(self, addr):
        return Echo()

class SampleTest(unittest.TestCase):
    '''Sample test case class derived straight from twisteds TestCase'''
    is_a_first_test = True
    endppoint = None
    def logLater(self, msgg = None):
        log.msg('called later message')
    @defer.inlineCallbacks
    def setUp(self):
        if self.__class__.is_a_first_test:
            self.__class__.endpoint = TCP4ServerEndpoint(reactor, 8007)
            self.__class__.endpoint.listen(EchoFactory())
            self.__class__.is_a_first_test = False
        log.msg('setting Up ... You may try (re)connecting now!!!')
        log.msg('We have endpoint: %s' % self.endpoint)
        yield reactor.callLater(5, self.logLater)
        log.msg('setUp done')
    def tearDown(self):
        log.msg('tearDown started')
        result = defer.Deferred()
        result.addCallback(self.logLater)
        reactor.callLater(5, result.callback, 'tearDown msg')
        log.msg('leaving tearDown')
        return result
    @defer.inlineCallbacks
    def test_00(self):
        log.msg('00 test body')
        sample_defer = defer.Deferred()
        sample_defer.addCallback(self.logLater)
        reactor.callLater(5, sample_defer.callback, 'Some sample action 00')
        log.msg('waiting reactor deferred')
        yield sample_defer
        log.msg('done with test body')
    @defer.inlineCallbacks
    def test_01(self):
        log.msg('01 test body')
        sample_defer = defer.Deferred()
        sample_defer.addCallback(self.logLater)
        reactor.callLater(5, sample_defer.callback, 'Some sample action 01')
        log.msg('waiting reactor deferred')
        yield sample_defer
    @defer.inlineCallbacks
    def test_02(self):
        log.msg('02 test body')
        sample_defer = defer.Deferred()
        sample_defer.addCallback(self.logLater)
        reactor.callLater(5, sample_defer.callback, 'Some sample action 02')
        log.msg('waiting reactor deferred')
        yield sample_defer

使用以下文件运行以上文件:

trial test-file.py

在每次测试结束时显示"主回路终止"。之后,端口在所有测试中都保持侦听状态(根据netstat-n4lt)。但在第二次和第三次测试体(仅针对第一次)期间,telnet连接时没有回声。

扭曲的文档http://twistedmatrix.com/documents/current/core/howto/testing.html#auto3说:"试运行整个测试套件(超过4000个测试)在一个过程中,使用一个反应器"(不幸的是,我的twistedmatrix注册请求仍未得到验证,因此无法在那里提出问题)。

在我的真实案例中,每次测试都需要很长时间才能重复连接到服务器,所以我想对每类案例至少做一次。

那么,有没有一种方法可以在测试之间保持在线连接?

PS使用python 2.7.1,python扭曲10.2.0-1,Ubuntu 11.04

单元测试是独立的。在一个中设置的连接不应被另一个使用。

全局reactor是一个不幸的共享可变状态。单元测试应该尽可能避免使用它。作为对隔离和独立性的帮助,试验在测试方法之间(尽可能多地)重置它。无法禁用此功能。

由于你有一个昂贵的连接,你应该调查为该连接创建一个经过验证的假连接。一个经过验证的假是一些API的替代实现,它(通常)更易于测试(例如,创建你的假应该很快),有自己的一组单元测试,这些测试表明它的行为与你的真实实现相同。

你的单元测试可以使用这个经过验证的假测试,为每个测试方法创建一个新的测试方法。这避免了违反隔离和独立性的问题,还可以让单元测试快速运行。

最新更新