[Twisted-Python] TDD: How to test that a Timeout Exception occurs within the timeout period

[Tom Sheffler]

I am investigating timeout properties of the FTP server module.  I
have written the following trial test, that when functioning properly
will throw a PortConnectionError with a "DTPFactory timeout"

I would like to use something like

   self.failUnlessRaises()

but the test is also constructed using defer.deferredGenerator, and it
doesn't seem possible to apply the test for an exception to a
Generator-based test.  This is a useful type of test to be able to
write: to assert that an exceptional condition *will* occur within a
timeout period.  Failure is if the exceptional condition is *never*
raised.  Suggestions for attacking these types of properties is welcome.

FYI: Here is my test, along with the trial directive that if it runs
for 60 seconds it is a failure.  But first, here is the ERROR
traceback that a "successful" run of this test produces.

===============================================================================
[ERROR]
Traceback (most recent call last):
Failure: twisted.protocols.ftp.PortConnectionError: DTPFactory timeout

test_ftp_timeouts.TimeoutsFTPServerTestCase.test_FtpControlChannelTimesOut
-------------------------------------------------------------------------------


    def test_FtpControlChannelTimesOut(self):
        """Test that after the FTP timeout, the control channel closes
itself."""

        # Set the timeout to something small, but greater than DTPTimeout
        self.serverProtocol.timeOut = 15

        # Login
        wfd = defer.waitForDeferred(self._anonymousLogin())
        yield wfd
        wfd.getResult()
        log.msg("Login Anonymous")

        # Wait for N seconds, protocol timeOut not fired yet
        wait1 = defer.waitForDeferred(self.mywait(2))
        yield wait1

        # Should get here and not fail
        log.msg("FTP is ok here")

        # Issue a PASV command, and extract the host and port from the response
        pasvCmd = defer.waitForDeferred(self.client.queueStringCommand('PASV'))
        yield pasvCmd
        responseLines = pasvCmd.getResult()
        log.msg("PASV responseLines", responseLines)
        host, port = ftp.decodeHostPort(responseLines[-1][4:])

        # Wait for N seconds, protocol timeOut will fire
        wait2 = defer.waitForDeferred(self.mywait(20))
        yield wait2
        log.msg("FTP should have timed out and left reactor clean")


    test_FtpControlChannelTimesOut =
defer.deferredGenerator(test_FtpControlChannelTimesOut)
    # test_FtpControlChannelTimesOut.skip = "skipping"
    test_FtpControlChannelTimesOut.timeout = 60 # is a real error if
it takes this long