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

Question

有没有办法在测试之间保持连接在线？

我想在运行多个测试时让用户保持登录状态，但发现了一些意外的问题：每次测试结束后，反应器就停止了（也就是说，一旦建立的连接就变得不可用了）。

为了研究这个问题，我准备了一个示例测试文件（见下文）。

这个文件创建了一个简单的回声服务器，并把它存储在类的字段中。总共运行了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 连接时没有回声（只有第一个测试有回声）。

Twisted 文档 http://twistedmatrix.com/documents/current/core/howto/testing.html#auto3 上说：

“Trial 在一个进程中运行整个测试套件（超过四千个测试），使用一个反应器” （不幸的是，我的 twistedmatrix 注册请求仍未通过验证，所以无法在那儿提问）。

在我的实际案例中，连接到服务器的时间太长，无法为每个测试重复这个过程，所以我希望至少为每类情况做一次连接。

那么，有没有办法在测试之间保持连接在线呢？

PS 使用的是：

python 2.7.1，

python-twisted 10.2.0-1，

Ubuntu 11.04

unittest twisted reactor asynchronous programming testing framework echo server trial connection persistence

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

1 个回答

撰写回答