def test_passiveRETR(self):
"""
Test the RETR command in passive mode: get a file and verify its
content.
L{ftp.FTPClient.retrieveFile} should return a Deferred which fires
with the protocol instance passed to it after the download has
completed.
(XXX - This API should be based on producers and consumers)
"""
def cbRetr(res, proto):
self.assertEquals(proto.buffer, 'x' * 1000)
def cbConnect(host, port, factory):
self.assertEquals(host, '127.0.0.1')
self.assertEquals(port, 12345)
proto = factory.buildProtocol((host, port))
proto.makeConnection(proto_helpers.StringTransport())
self.client.lineReceived(
'150 File status okay; about to open data connection.')
proto.dataReceived("x" * 1000)
proto.connectionLost(failure.Failure(error.ConnectionDone("")))
self.client.connectFactory = cbConnect
self._testLogin()
proto = _BufferingProtocol()
d = self.client.retrieveFile("spam", proto)
d.addCallback(cbRetr, proto)
self.assertEquals(self.transport.value(), 'PASV\r\n')
self.transport.clear()
self.client.lineReceived('227 Entering Passive Mode (%s).' %
(ftp.encodeHostPort('127.0.0.1', 12345),))
self.assertEquals(self.transport.value(), 'RETR spam\r\n')
self.transport.clear()
self.client.lineReceived('226 Transfer Complete.')
return d
评论列表
文章目录