2012-01-20 48 views
2

有没有一种方法可以在测试之间保持连接在线?使用twisted.trial.unittest时测试站之间会停止测试

我试图让用户在运行多个测试集时保持记录,但注意到意外的行为:每次测试后反应器都会停止(所以一旦建立的连接变得不可用)。

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

这创建了简单的侦听回声服务器并将其存储在类字段中。 运行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 away\r\n") 

    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连接在第二和第三个测试主体(仅针对第一个测试主体)时没有回显。

扭曲DOC http://twistedmatrix.com/documents/current/core/howto/testing.html#auto3说: “试运行在一个单一的过程中,整个测试套件(四千测试),以单一反应器” (不幸的是我twistedmatrix注册请求仍然没有验证,因此无法在那里提问)。

在我的真实情况下,与服务器的连接需要很长时间才能重复进行每次测试,所以我希望为每类病例至少进行一次这样的操作。

那么,有没有办法在测试之间保持连接在线?

PS使用 蟒2.7.1, 蟒捻10.2.0-1, Ubuntu的11.04

回答

2

单元测试是为了进行分离和独立的。在另一个中设置的连接不应该被使用。

全局反应堆是一个不幸的共享可变状态。在可能的情况下,单元测试应该完全避免使用它。作为孤立和独立的辅助手段,审判会在测试方法之间重新设置它(尽可能多)。此功能无法禁用。

由于您的连接设置昂贵,因此您应该调查为该连接创建验证伪造。经过验证的伪造是一些API的替代实现,它通常是更适合测试的(例如,创建伪造应该是快速的),并且拥有自己的单元测试集合,证明它的行为与实际执行方式相同。

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

+0

因此,如果我做得对,重置反应堆是一种预期的行为。 虽然我的测试客户机/服务器也是基于双绞线的(因此取决于单元测试所用的同一个电抗器),但双绞线总是会尝试重置其连接。 _问题:_扭曲目前不允许/鼓励在测试套件(或其他类级别的动作,如setUpClass)中的测试之间分配和共享连接,并且这是一个有意识的决定,将来要转向安全和严格的测试分离。 _是否正确?_ – yan

+0

这是正确的。 :)最终可能有多个反应堆,然后测试可能会共享一个与一个试验所用反应堆不同的连接(然后这样的连接只会是更多的状态,试验不知道并且无法控制)。 –