1
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
2
# See LICENSE for details.
7
Maintainer: Andrew Bennetts
13
from zope.interface import implements
15
from twisted.trial import unittest, util
16
from twisted.protocols import basic
17
from twisted.internet import reactor, task, protocol, defer, error
18
from twisted.internet.interfaces import IConsumer
19
from twisted.cred import portal, checkers, credentials
20
from twisted.python import failure, filepath
21
from twisted.test import proto_helpers
23
from twisted.protocols import ftp, loopback
26
_changeDirectorySuppression = util.suppress(
27
category=DeprecationWarning,
29
r"FTPClient\.changeDirectory is deprecated in Twisted 8\.2 and "
30
r"newer\. Use FTPClient\.cwd instead\."))
33
class Dummy(basic.LineReceiver):
38
def connectionMade(self):
39
self.f = self.factory # to save typing in pdb :-)
40
def lineReceived(self,line):
41
self.lines.append(line)
42
def rawDataReceived(self, data):
43
self.rawData.append(data)
44
def lineLengthExceeded(self, line):
48
class _BufferingProtocol(protocol.Protocol):
49
def connectionMade(self):
51
self.d = defer.Deferred()
52
def dataReceived(self, data):
54
def connectionLost(self, reason):
59
class FTPServerTestCase(unittest.TestCase):
61
Simple tests for an FTP server with the default settings.
63
@ivar clientFactory: class used as ftp client.
65
clientFactory = ftp.FTPClientBasic
66
userAnonymous = "anonymous"
70
self.directory = self.mktemp()
71
os.mkdir(self.directory)
72
self.dirPath = filepath.FilePath(self.directory)
75
p = portal.Portal(ftp.FTPRealm(self.directory))
76
p.registerChecker(checkers.AllowAnonymousAccess(),
77
credentials.IAnonymous)
78
self.factory = ftp.FTPFactory(portal=p,
79
userAnonymous=self.userAnonymous)
80
port = reactor.listenTCP(0, self.factory, interface="127.0.0.1")
81
self.addCleanup(port.stopListening)
83
# Hook the server's buildProtocol to make the protocol instance
84
# accessible to tests.
85
buildProtocol = self.factory.buildProtocol
87
def _rememberProtocolInstance(addr):
89
del self.factory.buildProtocol
91
protocol = buildProtocol(addr)
92
self.serverProtocol = protocol.wrappedProtocol
94
if self.serverProtocol.transport is not None:
95
self.serverProtocol.transport.loseConnection()
96
self.addCleanup(cleanupServer)
99
self.factory.buildProtocol = _rememberProtocolInstance
101
# Connect a client to it
102
portNum = port.getHost().port
103
clientCreator = protocol.ClientCreator(reactor, self.clientFactory)
104
d2 = clientCreator.connectTCP("127.0.0.1", portNum)
105
def gotClient(client):
107
self.addCleanup(self.client.transport.loseConnection)
108
d2.addCallback(gotClient)
109
return defer.gatherResults([d1, d2])
111
def assertCommandResponse(self, command, expectedResponseLines,
113
"""Asserts that a sending an FTP command receives the expected
116
Returns a Deferred. Optionally accepts a deferred to chain its actions
119
if chainDeferred is None:
120
chainDeferred = defer.succeed(None)
122
def queueCommand(ignored):
123
d = self.client.queueStringCommand(command)
124
def gotResponse(responseLines):
125
self.assertEquals(expectedResponseLines, responseLines)
126
return d.addCallback(gotResponse)
127
return chainDeferred.addCallback(queueCommand)
129
def assertCommandFailed(self, command, expectedResponse=None,
131
if chainDeferred is None:
132
chainDeferred = defer.succeed(None)
134
def queueCommand(ignored):
135
return self.client.queueStringCommand(command)
136
chainDeferred.addCallback(queueCommand)
137
self.assertFailure(chainDeferred, ftp.CommandFailed)
138
def failed(exception):
139
if expectedResponse is not None:
140
self.failUnlessEqual(
141
expectedResponse, exception.args[0])
142
return chainDeferred.addCallback(failed)
144
def _anonymousLogin(self):
145
d = self.assertCommandResponse(
147
['331 Guest login ok, type your email address as password.'])
148
return self.assertCommandResponse(
149
'PASS test@twistedmatrix.com',
150
['230 Anonymous login ok, access restrictions apply.'],
155
class FTPAnonymousTestCase(FTPServerTestCase):
157
Simple tests for an FTP server with different anonymous username.
158
The new anonymous username used in this test case is "guest"
160
userAnonymous = "guest"
162
def test_anonymousLogin(self):
164
Tests whether the changing of the anonymous username is working or not.
165
The FTP server should not comply about the need of password for the
166
username 'guest', letting it login as anonymous asking just an email
169
d = self.assertCommandResponse(
171
['331 Guest login ok, type your email address as password.'])
172
return self.assertCommandResponse(
173
'PASS test@twistedmatrix.com',
174
['230 Anonymous login ok, access restrictions apply.'],
179
class BasicFTPServerTestCase(FTPServerTestCase):
180
def testNotLoggedInReply(self):
181
"""When not logged in, all commands other than USER and PASS should
182
get NOT_LOGGED_IN errors.
184
commandList = ['CDUP', 'CWD', 'LIST', 'MODE', 'PASV',
185
'PWD', 'RETR', 'STRU', 'SYST', 'TYPE']
187
# Issue commands, check responses
188
def checkResponse(exception):
189
failureResponseLines = exception.args[0]
190
self.failUnless(failureResponseLines[-1].startswith("530"),
191
"Response didn't start with 530: %r"
192
% (failureResponseLines[-1],))
194
for command in commandList:
195
deferred = self.client.queueStringCommand(command)
196
self.assertFailure(deferred, ftp.CommandFailed)
197
deferred.addCallback(checkResponse)
198
deferreds.append(deferred)
199
return defer.DeferredList(deferreds, fireOnOneErrback=True)
201
def testPASSBeforeUSER(self):
202
"""Issuing PASS before USER should give an error."""
203
return self.assertCommandFailed(
205
["503 Incorrect sequence of commands: "
206
"USER required before PASS"])
208
def testNoParamsForUSER(self):
209
"""Issuing USER without a username is a syntax error."""
210
return self.assertCommandFailed(
212
['500 Syntax error: USER requires an argument.'])
214
def testNoParamsForPASS(self):
215
"""Issuing PASS without a password is a syntax error."""
216
d = self.client.queueStringCommand('USER foo')
217
return self.assertCommandFailed(
219
['500 Syntax error: PASS requires an argument.'],
222
def testAnonymousLogin(self):
223
return self._anonymousLogin()
226
"""Issuing QUIT should return a 221 message."""
227
d = self._anonymousLogin()
228
return self.assertCommandResponse(
233
def testAnonymousLoginDenied(self):
234
# Reconfigure the server to disallow anonymous access, and to have an
235
# IUsernamePassword checker that always rejects.
236
self.factory.allowAnonymous = False
237
denyAlwaysChecker = checkers.InMemoryUsernamePasswordDatabaseDontUse()
238
self.factory.portal.registerChecker(denyAlwaysChecker,
239
credentials.IUsernamePassword)
241
# Same response code as allowAnonymous=True, but different text.
242
d = self.assertCommandResponse(
244
['331 Password required for anonymous.'])
246
# It will be denied. No-one can login.
247
d = self.assertCommandFailed(
248
'PASS test@twistedmatrix.com',
249
['530 Sorry, Authentication failed.'],
252
# It's not just saying that. You aren't logged in.
253
d = self.assertCommandFailed(
255
['530 Please login with USER and PASS.'],
259
def testUnknownCommand(self):
260
d = self._anonymousLogin()
261
return self.assertCommandFailed(
263
["502 Command 'GIBBERISH' not implemented"],
266
def testRETRBeforePORT(self):
267
d = self._anonymousLogin()
268
return self.assertCommandFailed(
270
["503 Incorrect sequence of commands: "
271
"PORT or PASV required before RETR"],
274
def testSTORBeforePORT(self):
275
d = self._anonymousLogin()
276
return self.assertCommandFailed(
278
["503 Incorrect sequence of commands: "
279
"PORT or PASV required before STOR"],
282
def testBadCommandArgs(self):
283
d = self._anonymousLogin()
284
self.assertCommandFailed(
286
["504 Not implemented for parameter 'z'."],
288
self.assertCommandFailed(
290
["504 Not implemented for parameter 'I'."],
294
def testDecodeHostPort(self):
295
self.assertEquals(ftp.decodeHostPort('25,234,129,22,100,23'),
296
('25.234.129.22', 25623))
299
badValue = list(nums)
301
s = ','.join(map(str, badValue))
302
self.assertRaises(ValueError, ftp.decodeHostPort, s)
306
wfd = defer.waitForDeferred(self._anonymousLogin())
310
# Issue a PASV command, and extract the host and port from the response
311
pasvCmd = defer.waitForDeferred(self.client.queueStringCommand('PASV'))
313
responseLines = pasvCmd.getResult()
314
host, port = ftp.decodeHostPort(responseLines[-1][4:])
316
# Make sure the server is listening on the port it claims to be
317
self.assertEqual(port, self.serverProtocol.dtpPort.getHost().port)
319
# Semi-reasonable way to force cleanup
320
self.serverProtocol.transport.loseConnection()
321
testPASV = defer.deferredGenerator(testPASV)
324
d = self._anonymousLogin()
325
self.assertCommandResponse('SYST', ["215 UNIX Type: L8"],
330
def test_portRangeForwardError(self):
332
Exceptions other than L{error.CannotListenError} which are raised by
333
C{listenFactory} should be raised to the caller of L{FTP.getDTPPort}.
335
def listenFactory(portNumber, factory):
337
self.serverProtocol.listenFactory = listenFactory
339
self.assertRaises(RuntimeError, self.serverProtocol.getDTPPort,
343
def test_portRange(self):
345
L{FTP.passivePortRange} should determine the ports which
346
L{FTP.getDTPPort} attempts to bind. If no port from that iterator can
347
be bound, L{error.CannotListenError} should be raised, otherwise the
348
first successful result from L{FTP.listenFactory} should be returned.
350
def listenFactory(portNumber, factory):
351
if portNumber in (22032, 22033, 22034):
352
raise error.CannotListenError('localhost', portNumber, 'error')
354
self.serverProtocol.listenFactory = listenFactory
356
port = self.serverProtocol.getDTPPort(protocol.Factory())
357
self.assertEquals(port, 0)
359
self.serverProtocol.passivePortRange = xrange(22032, 65536)
360
port = self.serverProtocol.getDTPPort(protocol.Factory())
361
self.assertEquals(port, 22035)
363
self.serverProtocol.passivePortRange = xrange(22032, 22035)
364
self.assertRaises(error.CannotListenError,
365
self.serverProtocol.getDTPPort,
369
def test_portRangeInheritedFromFactory(self):
371
The L{FTP} instances created by L{ftp.FTPFactory.buildProtocol} have
372
their C{passivePortRange} attribute set to the same object the
373
factory's C{passivePortRange} attribute is set to.
375
portRange = xrange(2017, 2031)
376
self.factory.passivePortRange = portRange
377
protocol = self.factory.buildProtocol(None)
378
self.assertEqual(portRange, protocol.wrappedProtocol.passivePortRange)
382
class FTPServerTestCaseAdvancedClient(FTPServerTestCase):
384
Test FTP server with the L{ftp.FTPClient} class.
386
clientFactory = ftp.FTPClient
388
def test_anonymousSTOR(self):
390
Try to make an STOR as anonymous, and check that we got a permission
394
res.trap(ftp.CommandFailed)
395
self.assertEquals(res.value.args[0][0],
396
'550 foo: Permission denied.')
397
d1, d2 = self.client.storeFile('foo')
399
return defer.gatherResults([d1, d2])
402
class FTPServerPasvDataConnectionTestCase(FTPServerTestCase):
403
def _makeDataConnection(self, ignored=None):
404
# Establish a passive data connection (i.e. client connecting to
406
d = self.client.queueStringCommand('PASV')
407
def gotPASV(responseLines):
408
host, port = ftp.decodeHostPort(responseLines[-1][4:])
409
cc = protocol.ClientCreator(reactor, _BufferingProtocol)
410
return cc.connectTCP('127.0.0.1', port)
411
return d.addCallback(gotPASV)
413
def _download(self, command, chainDeferred=None):
414
if chainDeferred is None:
415
chainDeferred = defer.succeed(None)
417
chainDeferred.addCallback(self._makeDataConnection)
418
def queueCommand(downloader):
419
# wait for the command to return, and the download connection to be
421
d1 = self.client.queueStringCommand(command)
423
return defer.gatherResults([d1, d2])
424
chainDeferred.addCallback(queueCommand)
426
def downloadDone((ignored, downloader)):
427
return downloader.buffer
428
return chainDeferred.addCallback(downloadDone)
430
def testEmptyLIST(self):
432
d = self._anonymousLogin()
434
# No files, so the file listing should be empty
435
self._download('LIST', chainDeferred=d)
436
def checkEmpty(result):
437
self.assertEqual('', result)
438
return d.addCallback(checkEmpty)
440
def testTwoDirLIST(self):
441
# Make some directories
442
os.mkdir(os.path.join(self.directory, 'foo'))
443
os.mkdir(os.path.join(self.directory, 'bar'))
446
d = self._anonymousLogin()
448
# We expect 2 lines because there are two files.
449
self._download('LIST', chainDeferred=d)
450
def checkDownload(download):
451
self.assertEqual(2, len(download[:-2].split('\r\n')))
452
d.addCallback(checkDownload)
454
# Download a names-only listing.
455
self._download('NLST ', chainDeferred=d)
456
def checkDownload(download):
457
filenames = download[:-2].split('\r\n')
459
self.assertEqual(['bar', 'foo'], filenames)
460
d.addCallback(checkDownload)
462
# Download a listing of the 'foo' subdirectory. 'foo' has no files, so
463
# the file listing should be empty.
464
self._download('LIST foo', chainDeferred=d)
465
def checkDownload(download):
466
self.assertEqual('', download)
467
d.addCallback(checkDownload)
469
# Change the current working directory to 'foo'.
471
return self.client.queueStringCommand('CWD foo')
474
# Download a listing from within 'foo', and again it should be empty,
475
# because LIST uses the working directory by default.
476
self._download('LIST', chainDeferred=d)
477
def checkDownload(download):
478
self.assertEqual('', download)
479
return d.addCallback(checkDownload)
481
def testManyLargeDownloads(self):
483
d = self._anonymousLogin()
485
# Download a range of different size files
486
for size in range(100000, 110000, 500):
487
fObj = file(os.path.join(self.directory, '%d.txt' % (size,)), 'wb')
488
fObj.write('x' * size)
491
self._download('RETR %d.txt' % (size,), chainDeferred=d)
492
def checkDownload(download, size=size):
493
self.assertEqual('x' * size, download)
494
d.addCallback(checkDownload)
498
def test_NLSTEmpty(self):
500
NLST with no argument returns the directory listing for the current
504
d = self._anonymousLogin()
506
# Touch a file in the current working directory
507
self.dirPath.child('test.txt').touch()
508
# Make a directory in the current working directory
509
self.dirPath.child('foo').createDirectory()
511
self._download('NLST ', chainDeferred=d)
512
def checkDownload(download):
513
filenames = download[:-2].split('\r\n')
515
self.assertEquals(['foo', 'test.txt'], filenames)
516
return d.addCallback(checkDownload)
519
def test_NLSTNonexistent(self):
521
NLST on a non-existent file/directory returns nothing.
524
d = self._anonymousLogin()
526
self._download('NLST nonexistent.txt', chainDeferred=d)
527
def checkDownload(download):
528
self.assertEquals('', download)
529
return d.addCallback(checkDownload)
532
def test_NLSTOnPathToFile(self):
534
NLST on an existent file returns only the path to that file.
537
d = self._anonymousLogin()
539
# Touch a file in the current working directory
540
self.dirPath.child('test.txt').touch()
542
self._download('NLST test.txt', chainDeferred=d)
543
def checkDownload(download):
544
filenames = download[:-2].split('\r\n')
545
self.assertEquals(['test.txt'], filenames)
546
return d.addCallback(checkDownload)
550
class FTPServerPortDataConnectionTestCase(FTPServerPasvDataConnectionTestCase):
553
return FTPServerPasvDataConnectionTestCase.setUp(self)
555
def _makeDataConnection(self, ignored=None):
556
# Establish an active data connection (i.e. server connecting to
558
deferred = defer.Deferred()
559
class DataFactory(protocol.ServerFactory):
560
protocol = _BufferingProtocol
561
def buildProtocol(self, addr):
562
p = protocol.ServerFactory.buildProtocol(self, addr)
563
reactor.callLater(0, deferred.callback, p)
565
dataPort = reactor.listenTCP(0, DataFactory(), interface='127.0.0.1')
566
self.dataPorts.append(dataPort)
567
cmd = 'PORT ' + ftp.encodeHostPort('127.0.0.1', dataPort.getHost().port)
568
self.client.queueStringCommand(cmd)
572
l = [defer.maybeDeferred(port.stopListening) for port in self.dataPorts]
573
d = defer.maybeDeferred(
574
FTPServerPasvDataConnectionTestCase.tearDown, self)
576
return defer.DeferredList(l, fireOnOneErrback=True)
578
def testPORTCannotConnect(self):
580
d = self._anonymousLogin()
582
# Listen on a port, and immediately stop listening as a way to find a
583
# port number that is definitely closed.
584
def loggedIn(ignored):
585
port = reactor.listenTCP(0, protocol.Factory(),
586
interface='127.0.0.1')
587
portNum = port.getHost().port
588
d = port.stopListening()
589
d.addCallback(lambda _: portNum)
591
d.addCallback(loggedIn)
593
# Tell the server to connect to that port with a PORT command, and
594
# verify that it fails with the right error.
595
def gotPortNum(portNum):
596
return self.assertCommandFailed(
597
'PORT ' + ftp.encodeHostPort('127.0.0.1', portNum),
598
["425 Can't open data connection."])
599
return d.addCallback(gotPortNum)
603
class DTPFactoryTests(unittest.TestCase):
605
Tests for L{ftp.DTPFactory}.
609
Create a fake protocol interpreter and a L{ftp.DTPFactory} instance to
612
self.reactor = task.Clock()
614
class ProtocolInterpreter(object):
617
self.protocolInterpreter = ProtocolInterpreter()
618
self.factory = ftp.DTPFactory(
619
self.protocolInterpreter, None, self.reactor)
622
def test_setTimeout(self):
624
L{ftp.DTPFactory.setTimeout} uses the reactor passed to its initializer
625
to set up a timed event to time out the DTP setup after the specified
628
# Make sure the factory's deferred fails with the right exception, and
629
# make it so we can tell exactly when it fires.
631
d = self.assertFailure(self.factory.deferred, ftp.PortConnectionError)
632
d.addCallback(finished.append)
634
self.factory.setTimeout(6)
636
# Advance the clock almost to the timeout
637
self.reactor.advance(5)
639
# Nothing should have happened yet.
640
self.assertFalse(finished)
642
# Advance it to the configured timeout.
643
self.reactor.advance(1)
645
# Now the Deferred should have failed with TimeoutError.
646
self.assertTrue(finished)
648
# There should also be no calls left in the reactor.
649
self.assertFalse(self.reactor.calls)
652
def test_buildProtocolOnce(self):
654
A L{ftp.DTPFactory} instance's C{buildProtocol} method can be used once
655
to create a L{ftp.DTP} instance.
657
protocol = self.factory.buildProtocol(None)
658
self.assertIsInstance(protocol, ftp.DTP)
660
# A subsequent call returns None.
661
self.assertIdentical(self.factory.buildProtocol(None), None)
664
def test_timeoutAfterConnection(self):
666
If a timeout has been set up using L{ftp.DTPFactory.setTimeout}, it is
667
cancelled by L{ftp.DTPFactory.buildProtocol}.
669
self.factory.setTimeout(10)
670
protocol = self.factory.buildProtocol(None)
671
# Make sure the call is no longer active.
672
self.assertFalse(self.reactor.calls)
675
def test_connectionAfterTimeout(self):
677
If L{ftp.DTPFactory.buildProtocol} is called after the timeout
678
specified by L{ftp.DTPFactory.setTimeout} has elapsed, C{None} is
681
# Handle the error so it doesn't get logged.
682
d = self.assertFailure(self.factory.deferred, ftp.PortConnectionError)
684
# Set up the timeout and then cause it to elapse so the Deferred does
686
self.factory.setTimeout(10)
687
self.reactor.advance(10)
689
# Try to get a protocol - we should not be able to.
690
self.assertIdentical(self.factory.buildProtocol(None), None)
692
# Make sure the Deferred is doing the right thing.
696
def test_timeoutAfterConnectionFailed(self):
698
L{ftp.DTPFactory.deferred} fails with L{PortConnectionError} when
699
L{ftp.DTPFactory.clientConnectionFailed} is called. If the timeout
700
specified with L{ftp.DTPFactory.setTimeout} expires after that, nothing
704
d = self.assertFailure(self.factory.deferred, ftp.PortConnectionError)
705
d.addCallback(finished.append)
707
self.factory.setTimeout(10)
708
self.assertFalse(finished)
709
self.factory.clientConnectionFailed(None, None)
710
self.assertTrue(finished)
711
self.reactor.advance(10)
715
def test_connectionFailedAfterTimeout(self):
717
If L{ftp.DTPFactory.clientConnectionFailed} is called after the timeout
718
specified by L{ftp.DTPFactory.setTimeout} has elapsed, nothing beyond
719
the normal timeout before happens.
721
# Handle the error so it doesn't get logged.
722
d = self.assertFailure(self.factory.deferred, ftp.PortConnectionError)
724
# Set up the timeout and then cause it to elapse so the Deferred does
726
self.factory.setTimeout(10)
727
self.reactor.advance(10)
729
# Now fail the connection attempt. This should do nothing. In
730
# particular, it should not raise an exception.
731
self.factory.clientConnectionFailed(None, defer.TimeoutError("foo"))
733
# Give the Deferred to trial so it can make sure it did what we
739
# -- Client Tests -----------------------------------------------------------
741
class PrintLines(protocol.Protocol):
742
"""Helper class used by FTPFileListingTests."""
744
def __init__(self, lines):
747
def connectionMade(self):
748
for line in self._lines:
749
self.transport.write(line + "\r\n")
750
self.transport.loseConnection()
753
class MyFTPFileListProtocol(ftp.FTPFileListProtocol):
756
ftp.FTPFileListProtocol.__init__(self)
758
def unknownLine(self, line):
759
self.other.append(line)
762
class FTPFileListingTests(unittest.TestCase):
763
def getFilesForLines(self, lines):
764
fileList = MyFTPFileListProtocol()
765
d = loopback.loopbackAsync(PrintLines(lines), fileList)
766
d.addCallback(lambda _: (fileList.files, fileList.other))
769
def testOneLine(self):
770
# This example line taken from the docstring for FTPFileListProtocol
771
line = '-rw-r--r-- 1 root other 531 Jan 29 03:26 README'
772
def check(((file,), other)):
773
self.failIf(other, 'unexpect unparsable lines: %s' % repr(other))
774
self.failUnless(file['filetype'] == '-', 'misparsed fileitem')
775
self.failUnless(file['perms'] == 'rw-r--r--', 'misparsed perms')
776
self.failUnless(file['owner'] == 'root', 'misparsed fileitem')
777
self.failUnless(file['group'] == 'other', 'misparsed fileitem')
778
self.failUnless(file['size'] == 531, 'misparsed fileitem')
779
self.failUnless(file['date'] == 'Jan 29 03:26', 'misparsed fileitem')
780
self.failUnless(file['filename'] == 'README', 'misparsed fileitem')
781
self.failUnless(file['nlinks'] == 1, 'misparsed nlinks')
782
self.failIf(file['linktarget'], 'misparsed linktarget')
783
return self.getFilesForLines([line]).addCallback(check)
785
def testVariantLines(self):
786
line1 = 'drw-r--r-- 2 root other 531 Jan 9 2003 A'
787
line2 = 'lrw-r--r-- 1 root other 1 Jan 29 03:26 B -> A'
789
def check(((file1, file2), (other,))):
790
self.failUnless(other == 'woohoo! \r', 'incorrect other line')
792
self.failUnless(file1['filetype'] == 'd', 'misparsed fileitem')
793
self.failUnless(file1['perms'] == 'rw-r--r--', 'misparsed perms')
794
self.failUnless(file1['owner'] == 'root', 'misparsed owner')
795
self.failUnless(file1['group'] == 'other', 'misparsed group')
796
self.failUnless(file1['size'] == 531, 'misparsed size')
797
self.failUnless(file1['date'] == 'Jan 9 2003', 'misparsed date')
798
self.failUnless(file1['filename'] == 'A', 'misparsed filename')
799
self.failUnless(file1['nlinks'] == 2, 'misparsed nlinks')
800
self.failIf(file1['linktarget'], 'misparsed linktarget')
802
self.failUnless(file2['filetype'] == 'l', 'misparsed fileitem')
803
self.failUnless(file2['perms'] == 'rw-r--r--', 'misparsed perms')
804
self.failUnless(file2['owner'] == 'root', 'misparsed owner')
805
self.failUnless(file2['group'] == 'other', 'misparsed group')
806
self.failUnless(file2['size'] == 1, 'misparsed size')
807
self.failUnless(file2['date'] == 'Jan 29 03:26', 'misparsed date')
808
self.failUnless(file2['filename'] == 'B', 'misparsed filename')
809
self.failUnless(file2['nlinks'] == 1, 'misparsed nlinks')
810
self.failUnless(file2['linktarget'] == 'A', 'misparsed linktarget')
811
return self.getFilesForLines([line1, line2, line3]).addCallback(check)
813
def testUnknownLine(self):
814
def check((files, others)):
815
self.failIf(files, 'unexpected file entries')
816
self.failUnless(others == ['ABC\r', 'not a file\r'],
817
'incorrect unparsable lines: %s' % repr(others))
818
return self.getFilesForLines(['ABC', 'not a file']).addCallback(check)
821
# This example derived from bug description in issue 514.
822
fileList = ftp.FTPFileListProtocol()
824
'-rw-r--r-- 1 root other 531 Jan 29 2003 README\n')
825
class PrintLine(protocol.Protocol):
826
def connectionMade(self):
827
self.transport.write(exampleLine)
828
self.transport.loseConnection()
831
file = fileList.files[0]
832
self.failUnless(file['size'] == 531, 'misparsed fileitem')
833
self.failUnless(file['date'] == 'Jan 29 2003', 'misparsed fileitem')
834
self.failUnless(file['filename'] == 'README', 'misparsed fileitem')
836
d = loopback.loopbackAsync(PrintLine(), fileList)
837
return d.addCallback(check)
840
class FTPClientTests(unittest.TestCase):
842
def testFailedRETR(self):
843
f = protocol.Factory()
845
port = reactor.listenTCP(0, f, interface="127.0.0.1")
846
self.addCleanup(port.stopListening)
847
portNum = port.getHost().port
848
# This test data derived from a bug report by ranty on #twisted
849
responses = ['220 ready, dude (vsFTPd 1.0.0: beat me, break me)',
851
'331 Please specify the password.',
852
# PASS twisted@twistedmatrix.com
853
'230 Login successful. Have fun.',
855
'200 Binary it is, then.',
857
'227 Entering Passive Mode (127,0,0,1,%d,%d)' %
858
(portNum >> 8, portNum & 0xff),
859
# RETR /file/that/doesnt/exist
860
'550 Failed to open file.']
861
f.buildProtocol = lambda addr: PrintLines(responses)
863
client = ftp.FTPClient(passive=1)
864
cc = protocol.ClientCreator(reactor, ftp.FTPClient, passive=1)
865
d = cc.connectTCP('127.0.0.1', portNum)
866
def gotClient(client):
867
p = protocol.Protocol()
868
return client.retrieveFile('/file/that/doesnt/exist', p)
869
d.addCallback(gotClient)
870
return self.assertFailure(d, ftp.CommandFailed)
872
def test_errbacksUponDisconnect(self):
874
Test the ftp command errbacks when a connection lost happens during
877
ftpClient = ftp.FTPClient()
878
tr = proto_helpers.StringTransportWithDisconnection()
879
ftpClient.makeConnection(tr)
880
tr.protocol = ftpClient
881
d = ftpClient.list('some path', Dummy())
887
from twisted.internet.main import CONNECTION_LOST
888
ftpClient.connectionLost(failure.Failure(CONNECTION_LOST))
889
self.failUnless(m, m)
894
class FTPClientTestCase(unittest.TestCase):
896
Test advanced FTP client commands.
900
Create a FTP client and connect it to fake transport.
902
self.client = ftp.FTPClient()
903
self.transport = proto_helpers.StringTransportWithDisconnection()
904
self.client.makeConnection(self.transport)
905
self.transport.protocol = self.client
910
Deliver disconnection notification to the client so that it can
911
perform any cleanup which may be required.
913
self.client.connectionLost(error.ConnectionLost())
916
def _testLogin(self):
920
self.assertEquals(self.transport.value(), '')
921
self.client.lineReceived(
922
'331 Guest login ok, type your email address as password.')
923
self.assertEquals(self.transport.value(), 'USER anonymous\r\n')
924
self.transport.clear()
925
self.client.lineReceived(
926
'230 Anonymous login ok, access restrictions apply.')
927
self.assertEquals(self.transport.value(), 'TYPE I\r\n')
928
self.transport.clear()
929
self.client.lineReceived('200 Type set to I.')
934
Test the CDUP command.
936
L{ftp.FTPClient.cdup} should return a Deferred which fires with a
937
sequence of one element which is the string the server sent
938
indicating that the command was executed successfully.
940
(XXX - This is a bad API)
943
self.assertEquals(res[0], '250 Requested File Action Completed OK')
946
d = self.client.cdup().addCallback(cbCdup)
947
self.assertEquals(self.transport.value(), 'CDUP\r\n')
948
self.transport.clear()
949
self.client.lineReceived('250 Requested File Action Completed OK')
953
def test_failedCDUP(self):
955
Test L{ftp.FTPClient.cdup}'s handling of a failed CDUP command.
957
When the CDUP command fails, the returned Deferred should errback
958
with L{ftp.CommandFailed}.
961
d = self.client.cdup()
962
self.assertFailure(d, ftp.CommandFailed)
963
self.assertEquals(self.transport.value(), 'CDUP\r\n')
964
self.transport.clear()
965
self.client.lineReceived('550 ..: No such file or directory')
971
Test the PWD command.
973
L{ftp.FTPClient.pwd} should return a Deferred which fires with a
974
sequence of one element which is a string representing the current
975
working directory on the server.
977
(XXX - This is a bad API)
980
self.assertEquals(ftp.parsePWDResponse(res[0]), "/bar/baz")
983
d = self.client.pwd().addCallback(cbPwd)
984
self.assertEquals(self.transport.value(), 'PWD\r\n')
985
self.client.lineReceived('257 "/bar/baz"')
989
def test_failedPWD(self):
991
Test a failure in PWD command.
993
When the PWD command fails, the returned Deferred should errback
994
with L{ftp.CommandFailed}.
997
d = self.client.pwd()
998
self.assertFailure(d, ftp.CommandFailed)
999
self.assertEquals(self.transport.value(), 'PWD\r\n')
1000
self.client.lineReceived('550 /bar/baz: No such file or directory')
1006
Test the CWD command.
1008
L{ftp.FTPClient.cwd} should return a Deferred which fires with a
1009
sequence of one element which is the string the server sent
1010
indicating that the command was executed successfully.
1012
(XXX - This is a bad API)
1015
self.assertEquals(res[0], '250 Requested File Action Completed OK')
1018
d = self.client.cwd("bar/foo").addCallback(cbCwd)
1019
self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
1020
self.client.lineReceived('250 Requested File Action Completed OK')
1024
def test_failedCWD(self):
1026
Test a failure in CWD command.
1028
When the PWD command fails, the returned Deferred should errback
1029
with L{ftp.CommandFailed}.
1032
d = self.client.cwd("bar/foo")
1033
self.assertFailure(d, ftp.CommandFailed)
1034
self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
1035
self.client.lineReceived('550 bar/foo: No such file or directory')
1039
def test_passiveRETR(self):
1041
Test the RETR command in passive mode: get a file and verify its
1044
L{ftp.FTPClient.retrieveFile} should return a Deferred which fires
1045
with the protocol instance passed to it after the download has
1048
(XXX - This API should be based on producers and consumers)
1050
def cbRetr(res, proto):
1051
self.assertEquals(proto.buffer, 'x' * 1000)
1053
def cbConnect(host, port, factory):
1054
self.assertEquals(host, '127.0.0.1')
1055
self.assertEquals(port, 12345)
1056
proto = factory.buildProtocol((host, port))
1057
proto.makeConnection(proto_helpers.StringTransport())
1058
self.client.lineReceived(
1059
'150 File status okay; about to open data connection.')
1060
proto.dataReceived("x" * 1000)
1061
proto.connectionLost(failure.Failure(error.ConnectionDone("")))
1063
self.client.connectFactory = cbConnect
1065
proto = _BufferingProtocol()
1066
d = self.client.retrieveFile("spam", proto)
1067
d.addCallback(cbRetr, proto)
1068
self.assertEquals(self.transport.value(), 'PASV\r\n')
1069
self.transport.clear()
1070
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1071
(ftp.encodeHostPort('127.0.0.1', 12345),))
1072
self.assertEquals(self.transport.value(), 'RETR spam\r\n')
1073
self.transport.clear()
1074
self.client.lineReceived('226 Transfer Complete.')
1078
def test_RETR(self):
1080
Test the RETR command in non-passive mode.
1082
Like L{test_passiveRETR} but in the configuration where the server
1083
establishes the data connection to the client, rather than the other
1086
self.client.passive = False
1088
def generatePort(portCmd):
1089
portCmd.text = 'PORT %s' % (ftp.encodeHostPort('127.0.0.1', 9876),)
1090
portCmd.protocol.makeConnection(proto_helpers.StringTransport())
1091
portCmd.protocol.dataReceived("x" * 1000)
1092
portCmd.protocol.connectionLost(
1093
failure.Failure(error.ConnectionDone("")))
1095
def cbRetr(res, proto):
1096
self.assertEquals(proto.buffer, 'x' * 1000)
1098
self.client.generatePortCommand = generatePort
1100
proto = _BufferingProtocol()
1101
d = self.client.retrieveFile("spam", proto)
1102
d.addCallback(cbRetr, proto)
1103
self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
1104
(ftp.encodeHostPort('127.0.0.1', 9876),))
1105
self.transport.clear()
1106
self.client.lineReceived('200 PORT OK')
1107
self.assertEquals(self.transport.value(), 'RETR spam\r\n')
1108
self.transport.clear()
1109
self.client.lineReceived('226 Transfer Complete.')
1113
def test_failedRETR(self):
1115
Try to RETR an unexisting file.
1117
L{ftp.FTPClient.retrieveFile} should return a Deferred which
1118
errbacks with L{ftp.CommandFailed} if the server indicates the file
1119
cannot be transferred for some reason.
1121
def cbConnect(host, port, factory):
1122
self.assertEquals(host, '127.0.0.1')
1123
self.assertEquals(port, 12345)
1124
proto = factory.buildProtocol((host, port))
1125
proto.makeConnection(proto_helpers.StringTransport())
1126
self.client.lineReceived(
1127
'150 File status okay; about to open data connection.')
1128
proto.connectionLost(failure.Failure(error.ConnectionDone("")))
1130
self.client.connectFactory = cbConnect
1132
proto = _BufferingProtocol()
1133
d = self.client.retrieveFile("spam", proto)
1134
self.assertFailure(d, ftp.CommandFailed)
1135
self.assertEquals(self.transport.value(), 'PASV\r\n')
1136
self.transport.clear()
1137
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1138
(ftp.encodeHostPort('127.0.0.1', 12345),))
1139
self.assertEquals(self.transport.value(), 'RETR spam\r\n')
1140
self.transport.clear()
1141
self.client.lineReceived('550 spam: No such file or directory')
1145
def test_lostRETR(self):
1147
Try a RETR, but disconnect during the transfer.
1148
L{ftp.FTPClient.retrieveFile} should return a Deferred which
1149
errbacks with L{ftp.ConnectionLost)
1151
self.client.passive = False
1154
def generatePort(portCmd):
1155
portCmd.text = 'PORT %s' % (ftp.encodeHostPort('127.0.0.1', 9876),)
1156
tr = proto_helpers.StringTransportWithDisconnection()
1157
portCmd.protocol.makeConnection(tr)
1158
tr.protocol = portCmd.protocol
1159
portCmd.protocol.dataReceived("x" * 500)
1162
self.client.generatePortCommand = generatePort
1164
proto = _BufferingProtocol()
1165
d = self.client.retrieveFile("spam", proto)
1166
self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
1167
(ftp.encodeHostPort('127.0.0.1', 9876),))
1168
self.transport.clear()
1169
self.client.lineReceived('200 PORT OK')
1170
self.assertEquals(self.transport.value(), 'RETR spam\r\n')
1173
l[0].loseConnection()
1174
self.transport.loseConnection()
1175
self.assertFailure(d, ftp.ConnectionLost)
1179
def test_passiveSTOR(self):
1181
Test the STOR command: send a file and verify its content.
1183
L{ftp.FTPClient.storeFile} should return a two-tuple of Deferreds.
1184
The first of which should fire with a protocol instance when the
1185
data connection has been established and is responsible for sending
1186
the contents of the file. The second of which should fire when the
1187
upload has completed, the data connection has been closed, and the
1188
server has acknowledged receipt of the file.
1190
(XXX - storeFile should take a producer as an argument, instead, and
1191
only return a Deferred which fires when the upload has succeeded or
1194
tr = proto_helpers.StringTransport()
1195
def cbStore(sender):
1196
self.client.lineReceived(
1197
'150 File status okay; about to open data connection.')
1198
sender.transport.write("x" * 1000)
1200
sender.connectionLost(failure.Failure(error.ConnectionDone("")))
1203
self.assertEquals(tr.value(), "x" * 1000)
1205
def cbConnect(host, port, factory):
1206
self.assertEquals(host, '127.0.0.1')
1207
self.assertEquals(port, 12345)
1208
proto = factory.buildProtocol((host, port))
1209
proto.makeConnection(tr)
1211
self.client.connectFactory = cbConnect
1213
d1, d2 = self.client.storeFile("spam")
1214
d1.addCallback(cbStore)
1215
d2.addCallback(cbFinish)
1216
self.assertEquals(self.transport.value(), 'PASV\r\n')
1217
self.transport.clear()
1218
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1219
(ftp.encodeHostPort('127.0.0.1', 12345),))
1220
self.assertEquals(self.transport.value(), 'STOR spam\r\n')
1221
self.transport.clear()
1222
self.client.lineReceived('226 Transfer Complete.')
1223
return defer.gatherResults([d1, d2])
1226
def test_failedSTOR(self):
1228
Test a failure in the STOR command.
1230
If the server does not acknowledge successful receipt of the
1231
uploaded file, the second Deferred returned by
1232
L{ftp.FTPClient.storeFile} should errback with L{ftp.CommandFailed}.
1234
tr = proto_helpers.StringTransport()
1235
def cbStore(sender):
1236
self.client.lineReceived(
1237
'150 File status okay; about to open data connection.')
1238
sender.transport.write("x" * 1000)
1240
sender.connectionLost(failure.Failure(error.ConnectionDone("")))
1242
def cbConnect(host, port, factory):
1243
self.assertEquals(host, '127.0.0.1')
1244
self.assertEquals(port, 12345)
1245
proto = factory.buildProtocol((host, port))
1246
proto.makeConnection(tr)
1248
self.client.connectFactory = cbConnect
1250
d1, d2 = self.client.storeFile("spam")
1251
d1.addCallback(cbStore)
1252
self.assertFailure(d2, ftp.CommandFailed)
1253
self.assertEquals(self.transport.value(), 'PASV\r\n')
1254
self.transport.clear()
1255
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1256
(ftp.encodeHostPort('127.0.0.1', 12345),))
1257
self.assertEquals(self.transport.value(), 'STOR spam\r\n')
1258
self.transport.clear()
1259
self.client.lineReceived(
1260
'426 Transfer aborted. Data connection closed.')
1261
return defer.gatherResults([d1, d2])
1264
def test_STOR(self):
1266
Test the STOR command in non-passive mode.
1268
Like L{test_passiveSTOR} but in the configuration where the server
1269
establishes the data connection to the client, rather than the other
1272
tr = proto_helpers.StringTransport()
1273
self.client.passive = False
1274
def generatePort(portCmd):
1275
portCmd.text = 'PORT %s' % ftp.encodeHostPort('127.0.0.1', 9876)
1276
portCmd.protocol.makeConnection(tr)
1278
def cbStore(sender):
1279
self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
1280
(ftp.encodeHostPort('127.0.0.1', 9876),))
1281
self.transport.clear()
1282
self.client.lineReceived('200 PORT OK')
1283
self.assertEquals(self.transport.value(), 'STOR spam\r\n')
1284
self.transport.clear()
1285
self.client.lineReceived(
1286
'150 File status okay; about to open data connection.')
1287
sender.transport.write("x" * 1000)
1289
sender.connectionLost(failure.Failure(error.ConnectionDone("")))
1290
self.client.lineReceived('226 Transfer Complete.')
1293
self.assertEquals(tr.value(), "x" * 1000)
1295
self.client.generatePortCommand = generatePort
1297
d1, d2 = self.client.storeFile("spam")
1298
d1.addCallback(cbStore)
1299
d2.addCallback(cbFinish)
1300
return defer.gatherResults([d1, d2])
1303
def test_passiveLIST(self):
1305
Test the LIST command.
1307
L{ftp.FTPClient.list} should return a Deferred which fires with a
1308
protocol instance which was passed to list after the command has
1311
(XXX - This is a very unfortunate API; if my understanding is
1312
correct, the results are always at least line-oriented, so allowing
1313
a per-line parser function to be specified would make this simpler,
1314
but a default implementation should really be provided which knows
1315
how to deal with all the formats used in real servers, so
1316
application developers never have to care about this insanity. It
1317
would also be nice to either get back a Deferred of a list of
1318
filenames or to be able to consume the files as they are received
1319
(which the current API does allow, but in a somewhat inconvenient
1322
def cbList(res, fileList):
1323
fls = [f["filename"] for f in fileList.files]
1324
expected = ["foo", "bar", "baz"]
1327
self.assertEquals(fls, expected)
1329
def cbConnect(host, port, factory):
1330
self.assertEquals(host, '127.0.0.1')
1331
self.assertEquals(port, 12345)
1332
proto = factory.buildProtocol((host, port))
1333
proto.makeConnection(proto_helpers.StringTransport())
1334
self.client.lineReceived(
1335
'150 File status okay; about to open data connection.')
1337
'-rw-r--r-- 0 spam egg 100 Oct 10 2006 foo\r\n',
1338
'-rw-r--r-- 3 spam egg 100 Oct 10 2006 bar\r\n',
1339
'-rw-r--r-- 4 spam egg 100 Oct 10 2006 baz\r\n',
1342
proto.dataReceived(i)
1343
proto.connectionLost(failure.Failure(error.ConnectionDone("")))
1345
self.client.connectFactory = cbConnect
1347
fileList = ftp.FTPFileListProtocol()
1348
d = self.client.list('foo/bar', fileList).addCallback(cbList, fileList)
1349
self.assertEquals(self.transport.value(), 'PASV\r\n')
1350
self.transport.clear()
1351
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1352
(ftp.encodeHostPort('127.0.0.1', 12345),))
1353
self.assertEquals(self.transport.value(), 'LIST foo/bar\r\n')
1354
self.client.lineReceived('226 Transfer Complete.')
1358
def test_LIST(self):
1360
Test the LIST command in non-passive mode.
1362
Like L{test_passiveLIST} but in the configuration where the server
1363
establishes the data connection to the client, rather than the other
1366
self.client.passive = False
1367
def generatePort(portCmd):
1368
portCmd.text = 'PORT %s' % (ftp.encodeHostPort('127.0.0.1', 9876),)
1369
portCmd.protocol.makeConnection(proto_helpers.StringTransport())
1370
self.client.lineReceived(
1371
'150 File status okay; about to open data connection.')
1373
'-rw-r--r-- 0 spam egg 100 Oct 10 2006 foo\r\n',
1374
'-rw-r--r-- 3 spam egg 100 Oct 10 2006 bar\r\n',
1375
'-rw-r--r-- 4 spam egg 100 Oct 10 2006 baz\r\n',
1378
portCmd.protocol.dataReceived(i)
1379
portCmd.protocol.connectionLost(
1380
failure.Failure(error.ConnectionDone("")))
1382
def cbList(res, fileList):
1383
fls = [f["filename"] for f in fileList.files]
1384
expected = ["foo", "bar", "baz"]
1387
self.assertEquals(fls, expected)
1389
self.client.generatePortCommand = generatePort
1391
fileList = ftp.FTPFileListProtocol()
1392
d = self.client.list('foo/bar', fileList).addCallback(cbList, fileList)
1393
self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
1394
(ftp.encodeHostPort('127.0.0.1', 9876),))
1395
self.transport.clear()
1396
self.client.lineReceived('200 PORT OK')
1397
self.assertEquals(self.transport.value(), 'LIST foo/bar\r\n')
1398
self.transport.clear()
1399
self.client.lineReceived('226 Transfer Complete.')
1403
def test_failedLIST(self):
1405
Test a failure in LIST command.
1407
L{ftp.FTPClient.list} should return a Deferred which fails with
1408
L{ftp.CommandFailed} if the server indicates the indicated path is
1409
invalid for some reason.
1411
def cbConnect(host, port, factory):
1412
self.assertEquals(host, '127.0.0.1')
1413
self.assertEquals(port, 12345)
1414
proto = factory.buildProtocol((host, port))
1415
proto.makeConnection(proto_helpers.StringTransport())
1416
self.client.lineReceived(
1417
'150 File status okay; about to open data connection.')
1418
proto.connectionLost(failure.Failure(error.ConnectionDone("")))
1420
self.client.connectFactory = cbConnect
1422
fileList = ftp.FTPFileListProtocol()
1423
d = self.client.list('foo/bar', fileList)
1424
self.assertFailure(d, ftp.CommandFailed)
1425
self.assertEquals(self.transport.value(), 'PASV\r\n')
1426
self.transport.clear()
1427
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1428
(ftp.encodeHostPort('127.0.0.1', 12345),))
1429
self.assertEquals(self.transport.value(), 'LIST foo/bar\r\n')
1430
self.client.lineReceived('550 foo/bar: No such file or directory')
1434
def test_NLST(self):
1436
Test the NLST command in non-passive mode.
1438
L{ftp.FTPClient.nlst} should return a Deferred which fires with a
1439
list of filenames when the list command has completed.
1441
self.client.passive = False
1442
def generatePort(portCmd):
1443
portCmd.text = 'PORT %s' % (ftp.encodeHostPort('127.0.0.1', 9876),)
1444
portCmd.protocol.makeConnection(proto_helpers.StringTransport())
1445
self.client.lineReceived(
1446
'150 File status okay; about to open data connection.')
1447
portCmd.protocol.dataReceived('foo\r\n')
1448
portCmd.protocol.dataReceived('bar\r\n')
1449
portCmd.protocol.dataReceived('baz\r\n')
1450
portCmd.protocol.connectionLost(
1451
failure.Failure(error.ConnectionDone("")))
1453
def cbList(res, proto):
1454
fls = proto.buffer.splitlines()
1455
expected = ["foo", "bar", "baz"]
1458
self.assertEquals(fls, expected)
1460
self.client.generatePortCommand = generatePort
1462
lstproto = _BufferingProtocol()
1463
d = self.client.nlst('foo/bar', lstproto).addCallback(cbList, lstproto)
1464
self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
1465
(ftp.encodeHostPort('127.0.0.1', 9876),))
1466
self.transport.clear()
1467
self.client.lineReceived('200 PORT OK')
1468
self.assertEquals(self.transport.value(), 'NLST foo/bar\r\n')
1469
self.client.lineReceived('226 Transfer Complete.')
1473
def test_passiveNLST(self):
1475
Test the NLST command.
1477
Like L{test_passiveNLST} but in the configuration where the server
1478
establishes the data connection to the client, rather than the other
1481
def cbList(res, proto):
1482
fls = proto.buffer.splitlines()
1483
expected = ["foo", "bar", "baz"]
1486
self.assertEquals(fls, expected)
1488
def cbConnect(host, port, factory):
1489
self.assertEquals(host, '127.0.0.1')
1490
self.assertEquals(port, 12345)
1491
proto = factory.buildProtocol((host, port))
1492
proto.makeConnection(proto_helpers.StringTransport())
1493
self.client.lineReceived(
1494
'150 File status okay; about to open data connection.')
1495
proto.dataReceived('foo\r\n')
1496
proto.dataReceived('bar\r\n')
1497
proto.dataReceived('baz\r\n')
1498
proto.connectionLost(failure.Failure(error.ConnectionDone("")))
1500
self.client.connectFactory = cbConnect
1502
lstproto = _BufferingProtocol()
1503
d = self.client.nlst('foo/bar', lstproto).addCallback(cbList, lstproto)
1504
self.assertEquals(self.transport.value(), 'PASV\r\n')
1505
self.transport.clear()
1506
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1507
(ftp.encodeHostPort('127.0.0.1', 12345),))
1508
self.assertEquals(self.transport.value(), 'NLST foo/bar\r\n')
1509
self.client.lineReceived('226 Transfer Complete.')
1513
def test_failedNLST(self):
1515
Test a failure in NLST command.
1517
L{ftp.FTPClient.nlst} should return a Deferred which fails with
1518
L{ftp.CommandFailed} if the server indicates the indicated path is
1519
invalid for some reason.
1521
tr = proto_helpers.StringTransport()
1522
def cbConnect(host, port, factory):
1523
self.assertEquals(host, '127.0.0.1')
1524
self.assertEquals(port, 12345)
1525
proto = factory.buildProtocol((host, port))
1526
proto.makeConnection(tr)
1527
self.client.lineReceived(
1528
'150 File status okay; about to open data connection.')
1529
proto.connectionLost(failure.Failure(error.ConnectionDone("")))
1531
self.client.connectFactory = cbConnect
1533
lstproto = _BufferingProtocol()
1534
d = self.client.nlst('foo/bar', lstproto)
1535
self.assertFailure(d, ftp.CommandFailed)
1536
self.assertEquals(self.transport.value(), 'PASV\r\n')
1537
self.transport.clear()
1538
self.client.lineReceived('227 Entering Passive Mode (%s).' %
1539
(ftp.encodeHostPort('127.0.0.1', 12345),))
1540
self.assertEquals(self.transport.value(), 'NLST foo/bar\r\n')
1541
self.client.lineReceived('550 foo/bar: No such file or directory')
1545
def test_changeDirectoryDeprecated(self):
1547
L{ftp.FTPClient.changeDirectory} is deprecated and the direct caller of
1548
it is warned of this.
1551
d = self.assertWarns(
1553
"FTPClient.changeDirectory is deprecated in Twisted 8.2 and "
1554
"newer. Use FTPClient.cwd instead.",
1556
lambda: self.client.changeDirectory('.'))
1557
# This is necessary to make the Deferred fire. The Deferred needs
1558
# to fire so that tearDown doesn't cause it to errback and fail this
1559
# or (more likely) a later test.
1560
self.client.lineReceived('250 success')
1564
def test_changeDirectory(self):
1566
Test the changeDirectory method.
1568
L{ftp.FTPClient.changeDirectory} should return a Deferred which fires
1569
with True if succeeded.
1572
self.assertEquals(res, True)
1575
d = self.client.changeDirectory("bar/foo").addCallback(cbCd)
1576
self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
1577
self.client.lineReceived('250 Requested File Action Completed OK')
1579
test_changeDirectory.suppress = [_changeDirectorySuppression]
1582
def test_failedChangeDirectory(self):
1584
Test a failure in the changeDirectory method.
1586
The behaviour here is the same as a failed CWD.
1589
d = self.client.changeDirectory("bar/foo")
1590
self.assertFailure(d, ftp.CommandFailed)
1591
self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
1592
self.client.lineReceived('550 bar/foo: No such file or directory')
1594
test_failedChangeDirectory.suppress = [_changeDirectorySuppression]
1597
def test_strangeFailedChangeDirectory(self):
1599
Test a strange failure in changeDirectory method.
1601
L{ftp.FTPClient.changeDirectory} is stricter than CWD as it checks
1602
code 250 for success.
1605
d = self.client.changeDirectory("bar/foo")
1606
self.assertFailure(d, ftp.CommandFailed)
1607
self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
1608
self.client.lineReceived('252 I do what I want !')
1610
test_strangeFailedChangeDirectory.suppress = [_changeDirectorySuppression]
1613
def test_renameFromTo(self):
1615
L{ftp.FTPClient.rename} issues I{RNTO} and I{RNFR} commands and returns
1616
a L{Deferred} which fires when a file has successfully been renamed.
1620
d = self.client.rename("/spam", "/ham")
1621
self.assertEqual(self.transport.value(), 'RNFR /spam\r\n')
1622
self.transport.clear()
1625
'350 Requested file action pending further information.\r\n')
1626
self.client.lineReceived(fromResponse)
1627
self.assertEqual(self.transport.value(), 'RNTO /ham\r\n')
1629
'250 Requested File Action Completed OK')
1630
self.client.lineReceived(toResponse)
1632
d.addCallback(self.assertEqual, ([fromResponse], [toResponse]))
1636
def test_renameFromToEscapesPaths(self):
1638
L{ftp.FTPClient.rename} issues I{RNTO} and I{RNFR} commands with paths
1639
escaped according to U{http://cr.yp.to/ftp/filesystem.html}.
1643
fromFile = "/foo/ba\nr/baz"
1645
self.client.rename(fromFile, toFile)
1646
self.client.lineReceived("350 ")
1647
self.client.lineReceived("250 ")
1649
self.transport.value(),
1650
"RNFR /foo/ba\x00r/baz\r\n"
1651
"RNTO /qu\x00ux\r\n")
1654
def test_renameFromToFailingOnFirstError(self):
1656
The L{Deferred} returned by L{ftp.FTPClient.rename} is errbacked with
1657
L{CommandFailed} if the I{RNFR} command receives an error response code
1658
(for example, because the file does not exist).
1662
d = self.client.rename("/spam", "/ham")
1663
self.assertEqual(self.transport.value(), 'RNFR /spam\r\n')
1664
self.transport.clear()
1666
self.client.lineReceived('550 Requested file unavailable.\r\n')
1667
# The RNTO should not execute since the RNFR failed.
1668
self.assertEqual(self.transport.value(), '')
1670
return self.assertFailure(d, ftp.CommandFailed)
1673
def test_renameFromToFailingOnRenameTo(self):
1675
The L{Deferred} returned by L{ftp.FTPClient.rename} is errbacked with
1676
L{CommandFailed} if the I{RNTO} command receives an error response code
1677
(for example, because the destination directory does not exist).
1681
d = self.client.rename("/spam", "/ham")
1682
self.assertEqual(self.transport.value(), 'RNFR /spam\r\n')
1683
self.transport.clear()
1685
self.client.lineReceived('350 Requested file action pending further information.\r\n')
1686
self.assertEqual(self.transport.value(), 'RNTO /ham\r\n')
1687
self.client.lineReceived('550 Requested file unavailable.\r\n')
1688
return self.assertFailure(d, ftp.CommandFailed)
1691
def test_makeDirectory(self):
1693
L{ftp.FTPClient.makeDirectory} issues a I{MKD} command and returns a
1694
L{Deferred} which is called back with the server's response if the
1695
directory is created.
1699
d = self.client.makeDirectory("/spam")
1700
self.assertEquals(self.transport.value(), 'MKD /spam\r\n')
1701
self.client.lineReceived('257 "/spam" created.')
1702
return d.addCallback(self.assertEqual, ['257 "/spam" created.'])
1705
def test_makeDirectoryPathEscape(self):
1707
L{ftp.FTPClient.makeDirectory} escapes the path name it sends according
1708
to U{http://cr.yp.to/ftp/filesystem.html}.
1711
d = self.client.makeDirectory("/sp\nam")
1712
self.assertEqual(self.transport.value(), 'MKD /sp\x00am\r\n')
1713
# This is necessary to make the Deferred fire. The Deferred needs
1714
# to fire so that tearDown doesn't cause it to errback and fail this
1715
# or (more likely) a later test.
1716
self.client.lineReceived('257 win')
1720
def test_failedMakeDirectory(self):
1722
L{ftp.FTPClient.makeDirectory} returns a L{Deferred} which is errbacked
1723
with L{CommandFailed} if the server returns an error response code.
1727
d = self.client.makeDirectory("/spam")
1728
self.assertEquals(self.transport.value(), 'MKD /spam\r\n')
1729
self.client.lineReceived('550 PERMISSION DENIED')
1730
return self.assertFailure(d, ftp.CommandFailed)
1733
def test_getDirectory(self):
1735
Test the getDirectory method.
1737
L{ftp.FTPClient.getDirectory} should return a Deferred which fires with
1738
the current directory on the server. It wraps PWD command.
1741
self.assertEquals(res, "/bar/baz")
1744
d = self.client.getDirectory().addCallback(cbGet)
1745
self.assertEquals(self.transport.value(), 'PWD\r\n')
1746
self.client.lineReceived('257 "/bar/baz"')
1750
def test_failedGetDirectory(self):
1752
Test a failure in getDirectory method.
1754
The behaviour should be the same as PWD.
1757
d = self.client.getDirectory()
1758
self.assertFailure(d, ftp.CommandFailed)
1759
self.assertEquals(self.transport.value(), 'PWD\r\n')
1760
self.client.lineReceived('550 /bar/baz: No such file or directory')
1764
def test_anotherFailedGetDirectory(self):
1766
Test a different failure in getDirectory method.
1768
The response should be quoted to be parsed, so it returns an error
1772
d = self.client.getDirectory()
1773
self.assertFailure(d, ftp.CommandFailed)
1774
self.assertEquals(self.transport.value(), 'PWD\r\n')
1775
self.client.lineReceived('257 /bar/baz')
1779
def test_removeFile(self):
1781
L{ftp.FTPClient.removeFile} sends a I{DELE} command to the server for
1782
the indicated file and returns a Deferred which fires after the server
1783
sends a 250 response code.
1786
d = self.client.removeFile("/tmp/test")
1787
self.assertEquals(self.transport.value(), 'DELE /tmp/test\r\n')
1788
response = '250 Requested file action okay, completed.'
1789
self.client.lineReceived(response)
1790
return d.addCallback(self.assertEqual, [response])
1793
def test_failedRemoveFile(self):
1795
If the server returns a response code other than 250 in response to a
1796
I{DELE} sent by L{ftp.FTPClient.removeFile}, the L{Deferred} returned
1797
by C{removeFile} is errbacked with a L{Failure} wrapping a
1801
d = self.client.removeFile("/tmp/test")
1802
self.assertEquals(self.transport.value(), 'DELE /tmp/test\r\n')
1803
response = '501 Syntax error in parameters or arguments.'
1804
self.client.lineReceived(response)
1805
d = self.assertFailure(d, ftp.CommandFailed)
1806
d.addCallback(lambda exc: self.assertEqual(exc.args, ([response],)))
1810
def test_unparsableRemoveFileResponse(self):
1812
If the server returns a response line which cannot be parsed, the
1813
L{Deferred} returned by L{ftp.FTPClient.removeFile} is errbacked with a
1814
L{BadResponse} containing the response.
1817
d = self.client.removeFile("/tmp/test")
1818
response = '765 blah blah blah'
1819
self.client.lineReceived(response)
1820
d = self.assertFailure(d, ftp.BadResponse)
1821
d.addCallback(lambda exc: self.assertEqual(exc.args, ([response],)))
1825
def test_multilineRemoveFileResponse(self):
1827
If the server returns multiple response lines, the L{Deferred} returned
1828
by L{ftp.FTPClient.removeFile} is still fired with a true value if the
1829
ultimate response code is 250.
1832
d = self.client.removeFile("/tmp/test")
1833
response = ['250-perhaps a progress report',
1835
map(self.client.lineReceived, response)
1836
return d.addCallback(self.assertTrue)
1840
class FTPClientBasicTests(unittest.TestCase):
1842
def testGreeting(self):
1843
# The first response is captured as a greeting.
1844
ftpClient = ftp.FTPClientBasic()
1845
ftpClient.lineReceived('220 Imaginary FTP.')
1846
self.failUnlessEqual(['220 Imaginary FTP.'], ftpClient.greeting)
1848
def testResponseWithNoMessage(self):
1849
# Responses with no message are still valid, i.e. three digits followed
1850
# by a space is complete response.
1851
ftpClient = ftp.FTPClientBasic()
1852
ftpClient.lineReceived('220 ')
1853
self.failUnlessEqual(['220 '], ftpClient.greeting)
1855
def testMultilineResponse(self):
1856
ftpClient = ftp.FTPClientBasic()
1857
ftpClient.transport = proto_helpers.StringTransport()
1858
ftpClient.lineReceived('220 Imaginary FTP.')
1860
# Queue (and send) a dummy command, and set up a callback to capture the
1862
deferred = ftpClient.queueStringCommand('BLAH')
1864
deferred.addCallback(result.append)
1865
deferred.addErrback(self.fail)
1867
# Send the first line of a multiline response.
1868
ftpClient.lineReceived('210-First line.')
1869
self.failUnlessEqual([], result)
1871
# Send a second line, again prefixed with "nnn-".
1872
ftpClient.lineReceived('123-Second line.')
1873
self.failUnlessEqual([], result)
1875
# Send a plain line of text, no prefix.
1876
ftpClient.lineReceived('Just some text.')
1877
self.failUnlessEqual([], result)
1879
# Now send a short (less than 4 chars) line.
1880
ftpClient.lineReceived('Hi')
1881
self.failUnlessEqual([], result)
1883
# Now send an empty line.
1884
ftpClient.lineReceived('')
1885
self.failUnlessEqual([], result)
1887
# And a line with 3 digits in it, and nothing else.
1888
ftpClient.lineReceived('321')
1889
self.failUnlessEqual([], result)
1892
ftpClient.lineReceived('210 Done.')
1893
self.failUnlessEqual(
1900
'210 Done.'], result[0])
1903
def test_noPasswordGiven(self):
1905
Passing None as the password avoids sending the PASS command.
1907
# Create a client, and give it a greeting.
1908
ftpClient = ftp.FTPClientBasic()
1909
ftpClient.transport = proto_helpers.StringTransport()
1910
ftpClient.lineReceived('220 Welcome to Imaginary FTP.')
1912
# Queue a login with no password
1913
ftpClient.queueLogin('bob', None)
1914
self.assertEquals('USER bob\r\n', ftpClient.transport.value())
1916
# Clear the test buffer, acknowledge the USER command.
1917
ftpClient.transport.clear()
1918
ftpClient.lineReceived('200 Hello bob.')
1920
# The client shouldn't have sent anything more (i.e. it shouldn't have
1921
# sent a PASS command).
1922
self.assertEquals('', ftpClient.transport.value())
1925
def test_noPasswordNeeded(self):
1927
Receiving a 230 response to USER prevents PASS from being sent.
1929
# Create a client, and give it a greeting.
1930
ftpClient = ftp.FTPClientBasic()
1931
ftpClient.transport = proto_helpers.StringTransport()
1932
ftpClient.lineReceived('220 Welcome to Imaginary FTP.')
1934
# Queue a login with no password
1935
ftpClient.queueLogin('bob', 'secret')
1936
self.assertEquals('USER bob\r\n', ftpClient.transport.value())
1938
# Clear the test buffer, acknowledge the USER command with a 230
1940
ftpClient.transport.clear()
1941
ftpClient.lineReceived('230 Hello bob. No password needed.')
1943
# The client shouldn't have sent anything more (i.e. it shouldn't have
1944
# sent a PASS command).
1945
self.assertEquals('', ftpClient.transport.value())
1949
class PathHandling(unittest.TestCase):
1950
def testNormalizer(self):
1951
for inp, outp in [('a', ['a']),
1954
('a/b/c', ['a', 'b', 'c']),
1955
('/a/b/c', ['a', 'b', 'c']),
1958
self.assertEquals(ftp.toSegments([], inp), outp)
1960
for inp, outp in [('b', ['a', 'b']),
1964
('b/c', ['a', 'b', 'c']),
1965
('b/c/', ['a', 'b', 'c']),
1966
('/b/c', ['b', 'c']),
1967
('/b/c/', ['b', 'c'])]:
1968
self.assertEquals(ftp.toSegments(['a'], inp), outp)
1970
for inp, outp in [('//', []),
1973
('a//b', ['a', 'b'])]:
1974
self.assertEquals(ftp.toSegments([], inp), outp)
1976
for inp, outp in [('//', []),
1978
('b//c', ['a', 'b', 'c'])]:
1979
self.assertEquals(ftp.toSegments(['a'], inp), outp)
1981
for inp, outp in [('..', []),
1986
('/a/b/../', ['a']),
1987
('/a/b/../c', ['a', 'c']),
1988
('/a/b/../c/', ['a', 'c']),
1989
('/a/b/../../c', ['c']),
1990
('/a/b/../../c/', ['c']),
1991
('/a/b/../../c/..', []),
1992
('/a/b/../../c/../', [])]:
1993
self.assertEquals(ftp.toSegments(['x'], inp), outp)
1995
for inp in ['..', '../', 'a/../..', 'a/../../',
1996
'/..', '/../', '/a/../..', '/a/../../',
1998
self.assertRaises(ftp.InvalidPath, ftp.toSegments, [], inp)
2000
for inp in ['../..', '../../', '../a/../..']:
2001
self.assertRaises(ftp.InvalidPath, ftp.toSegments, ['x'], inp)
2005
class ErrnoToFailureTestCase(unittest.TestCase):
2007
Tests for L{ftp.errnoToFailure} errno checking.
2010
def test_notFound(self):
2012
C{errno.ENOENT} should be translated to L{ftp.FileNotFoundError}.
2014
d = ftp.errnoToFailure(errno.ENOENT, "foo")
2015
return self.assertFailure(d, ftp.FileNotFoundError)
2018
def test_permissionDenied(self):
2020
C{errno.EPERM} should be translated to L{ftp.PermissionDeniedError}.
2022
d = ftp.errnoToFailure(errno.EPERM, "foo")
2023
return self.assertFailure(d, ftp.PermissionDeniedError)
2026
def test_accessDenied(self):
2028
C{errno.EACCES} should be translated to L{ftp.PermissionDeniedError}.
2030
d = ftp.errnoToFailure(errno.EACCES, "foo")
2031
return self.assertFailure(d, ftp.PermissionDeniedError)
2034
def test_notDirectory(self):
2036
C{errno.ENOTDIR} should be translated to L{ftp.IsNotADirectoryError}.
2038
d = ftp.errnoToFailure(errno.ENOTDIR, "foo")
2039
return self.assertFailure(d, ftp.IsNotADirectoryError)
2042
def test_fileExists(self):
2044
C{errno.EEXIST} should be translated to L{ftp.FileExistsError}.
2046
d = ftp.errnoToFailure(errno.EEXIST, "foo")
2047
return self.assertFailure(d, ftp.FileExistsError)
2050
def test_isDirectory(self):
2052
C{errno.EISDIR} should be translated to L{ftp.IsADirectoryError}.
2054
d = ftp.errnoToFailure(errno.EISDIR, "foo")
2055
return self.assertFailure(d, ftp.IsADirectoryError)
2058
def test_passThrough(self):
2060
If an unknown errno is passed to L{ftp.errnoToFailure}, it should let
2061
the originating exception pass through.
2064
raise RuntimeError("bar")
2066
d = ftp.errnoToFailure(-1, "foo")
2067
return self.assertFailure(d, RuntimeError)
2071
class AnonymousFTPShellTestCase(unittest.TestCase):
2073
Test anynomous shell properties.
2076
def test_anonymousWrite(self):
2078
Check that L{ftp.FTPAnonymousShell} returns an error when trying to
2079
open it in write mode.
2081
shell = ftp.FTPAnonymousShell('')
2082
d = shell.openForWriting(('foo',))
2083
self.assertFailure(d, ftp.PermissionDeniedError)
2088
class IFTPShellTestsMixin:
2090
Generic tests for the C{IFTPShell} interface.
2093
def directoryExists(self, path):
2095
Test if the directory exists at C{path}.
2097
@param path: the relative path to check.
2100
@return: C{True} if C{path} exists and is a directory, C{False} if
2104
raise NotImplementedError()
2107
def createDirectory(self, path):
2109
Create a directory in C{path}.
2111
@param path: the relative path of the directory to create, with one
2115
raise NotImplementedError()
2118
def fileExists(self, path):
2120
Test if the file exists at C{path}.
2122
@param path: the relative path to check.
2125
@return: C{True} if C{path} exists and is a file, C{False} if it's not
2129
raise NotImplementedError()
2132
def createFile(self, path, fileContent=''):
2134
Create a file named C{path} with some content.
2136
@param path: the relative path of the file to create, without
2140
@param fileContent: the content of the file.
2141
@type fileContent: C{str}
2143
raise NotImplementedError()
2146
def test_createDirectory(self):
2148
C{directoryExists} should report correctly about directory existence,
2149
and C{createDirectory} should create a directory detectable by
2152
self.assertFalse(self.directoryExists('bar'))
2153
self.createDirectory('bar')
2154
self.assertTrue(self.directoryExists('bar'))
2157
def test_createFile(self):
2159
C{fileExists} should report correctly about file existence, and
2160
C{createFile} should create a file detectable by C{fileExists}.
2162
self.assertFalse(self.fileExists('file.txt'))
2163
self.createFile('file.txt')
2164
self.assertTrue(self.fileExists('file.txt'))
2167
def test_makeDirectory(self):
2169
Create a directory and check it ends in the filesystem.
2171
d = self.shell.makeDirectory(('foo',))
2173
self.assertTrue(self.directoryExists('foo'))
2174
return d.addCallback(cb)
2177
def test_makeDirectoryError(self):
2179
Creating a directory that already exists should fail with a
2180
C{ftp.FileExistsError}.
2182
self.createDirectory('foo')
2183
d = self.shell.makeDirectory(('foo',))
2184
return self.assertFailure(d, ftp.FileExistsError)
2187
def test_removeDirectory(self):
2189
Try to remove a directory and check it's removed from the filesystem.
2191
self.createDirectory('bar')
2192
d = self.shell.removeDirectory(('bar',))
2194
self.assertFalse(self.directoryExists('bar'))
2195
return d.addCallback(cb)
2198
def test_removeDirectoryOnFile(self):
2200
removeDirectory should not work in file and fail with a
2201
C{ftp.IsNotADirectoryError}.
2203
self.createFile('file.txt')
2204
d = self.shell.removeDirectory(('file.txt',))
2205
return self.assertFailure(d, ftp.IsNotADirectoryError)
2208
def test_removeNotExistingDirectory(self):
2210
Removing directory that doesn't exist should fail with a
2211
C{ftp.FileNotFoundError}.
2213
d = self.shell.removeDirectory(('bar',))
2214
return self.assertFailure(d, ftp.FileNotFoundError)
2217
def test_removeFile(self):
2219
Try to remove a file and check it's removed from the filesystem.
2221
self.createFile('file.txt')
2222
d = self.shell.removeFile(('file.txt',))
2224
self.assertFalse(self.fileExists('file.txt'))
2229
def test_removeFileOnDirectory(self):
2231
removeFile should not work on directory.
2233
self.createDirectory('ned')
2234
d = self.shell.removeFile(('ned',))
2235
return self.assertFailure(d, ftp.IsADirectoryError)
2238
def test_removeNotExistingFile(self):
2240
Try to remove a non existent file, and check it raises a
2241
L{ivfs.NotFoundError}.
2243
d = self.shell.removeFile(('foo',))
2244
return self.assertFailure(d, ftp.FileNotFoundError)
2247
def test_list(self):
2249
Check the output of the list method.
2251
self.createDirectory('ned')
2252
self.createFile('file.txt')
2253
d = self.shell.list(('.',))
2256
self.assertEquals(l,
2257
[('file.txt', []), ('ned', [])])
2258
return d.addCallback(cb)
2261
def test_listWithStat(self):
2263
Check the output of list with asked stats.
2265
self.createDirectory('ned')
2266
self.createFile('file.txt')
2267
d = self.shell.list(('.',), ('size', 'permissions',))
2270
self.assertEquals(len(l), 2)
2271
self.assertEquals(l[0][0], 'file.txt')
2272
self.assertEquals(l[1][0], 'ned')
2273
# Size and permissions are reported differently between platforms
2274
# so just check they are present
2275
self.assertEquals(len(l[0][1]), 2)
2276
self.assertEquals(len(l[1][1]), 2)
2277
return d.addCallback(cb)
2280
def test_listWithInvalidStat(self):
2282
Querying an invalid stat should result to a C{AttributeError}.
2284
self.createDirectory('ned')
2285
d = self.shell.list(('.',), ('size', 'whateverstat',))
2286
return self.assertFailure(d, AttributeError)
2289
def test_listFile(self):
2291
Check the output of the list method on a file.
2293
self.createFile('file.txt')
2294
d = self.shell.list(('file.txt',))
2297
self.assertEquals(l,
2299
return d.addCallback(cb)
2302
def test_listNotExistingDirectory(self):
2304
list on a directory that doesn't exist should fail with a
2305
L{ftp.FileNotFoundError}.
2307
d = self.shell.list(('foo',))
2308
return self.assertFailure(d, ftp.FileNotFoundError)
2311
def test_access(self):
2313
Try to access a resource.
2315
self.createDirectory('ned')
2316
d = self.shell.access(('ned',))
2320
def test_accessNotFound(self):
2322
access should fail on a resource that doesn't exist.
2324
d = self.shell.access(('foo',))
2325
return self.assertFailure(d, ftp.FileNotFoundError)
2328
def test_openForReading(self):
2330
Check that openForReading returns an object providing C{ftp.IReadFile}.
2332
self.createFile('file.txt')
2333
d = self.shell.openForReading(('file.txt',))
2335
self.assertTrue(ftp.IReadFile.providedBy(res))
2340
def test_openForReadingNotFound(self):
2342
openForReading should fail with a C{ftp.FileNotFoundError} on a file
2345
d = self.shell.openForReading(('ned',))
2346
return self.assertFailure(d, ftp.FileNotFoundError)
2349
def test_openForReadingOnDirectory(self):
2351
openForReading should not work on directory.
2353
self.createDirectory('ned')
2354
d = self.shell.openForReading(('ned',))
2355
return self.assertFailure(d, ftp.IsADirectoryError)
2358
def test_openForWriting(self):
2360
Check that openForWriting returns an object providing C{ftp.IWriteFile}.
2362
d = self.shell.openForWriting(('foo',))
2364
self.assertTrue(ftp.IWriteFile.providedBy(res))
2365
return res.receive().addCallback(cb2)
2367
self.assertTrue(IConsumer.providedBy(res))
2372
def test_openForWritingExistingDirectory(self):
2374
openForWriting should not be able to open a directory that already
2377
self.createDirectory('ned')
2378
d = self.shell.openForWriting(('ned',))
2379
return self.assertFailure(d, ftp.IsADirectoryError)
2382
def test_openForWritingInNotExistingDirectory(self):
2384
openForWring should fail with a L{ftp.FileNotFoundError} if you specify
2385
a file in a directory that doesn't exist.
2387
self.createDirectory('ned')
2388
d = self.shell.openForWriting(('ned', 'idonotexist', 'foo'))
2389
return self.assertFailure(d, ftp.FileNotFoundError)
2392
def test_statFile(self):
2394
Check the output of the stat method on a file.
2396
fileContent = 'wobble\n'
2397
self.createFile('file.txt', fileContent)
2398
d = self.shell.stat(('file.txt',), ('size', 'directory'))
2400
self.assertEquals(res[0], len(fileContent))
2401
self.assertFalse(res[1])
2406
def test_statDirectory(self):
2408
Check the output of the stat method on a directory.
2410
self.createDirectory('ned')
2411
d = self.shell.stat(('ned',), ('size', 'directory'))
2413
self.assertTrue(res[1])
2418
def test_statOwnerGroup(self):
2420
Check the owner and groups stats.
2422
self.createDirectory('ned')
2423
d = self.shell.stat(('ned',), ('owner', 'group'))
2425
self.assertEquals(len(res), 2)
2430
def test_statNotExisting(self):
2432
stat should fail with L{ftp.FileNotFoundError} on a file that doesn't
2435
d = self.shell.stat(('foo',), ('size', 'directory'))
2436
return self.assertFailure(d, ftp.FileNotFoundError)
2439
def test_invalidStat(self):
2441
Querying an invalid stat should result to a C{AttributeError}.
2443
self.createDirectory('ned')
2444
d = self.shell.stat(('ned',), ('size', 'whateverstat'))
2445
return self.assertFailure(d, AttributeError)
2448
def test_rename(self):
2450
Try to rename a directory.
2452
self.createDirectory('ned')
2453
d = self.shell.rename(('ned',), ('foo',))
2455
self.assertTrue(self.directoryExists('foo'))
2456
self.assertFalse(self.directoryExists('ned'))
2457
return d.addCallback(cb)
2460
def test_renameNotExisting(self):
2462
Renaming a directory that doesn't exist should fail with
2463
L{ftp.FileNotFoundError}.
2465
d = self.shell.rename(('foo',), ('bar',))
2466
return self.assertFailure(d, ftp.FileNotFoundError)
2470
class FTPShellTestCase(unittest.TestCase, IFTPShellTestsMixin):
2472
Tests for the C{ftp.FTPShell} object.
2477
Create a root directory and instantiate a shell.
2479
self.root = filepath.FilePath(self.mktemp())
2480
self.root.createDirectory()
2481
self.shell = ftp.FTPShell(self.root)
2484
def directoryExists(self, path):
2486
Test if the directory exists at C{path}.
2488
return self.root.child(path).isdir()
2491
def createDirectory(self, path):
2493
Create a directory in C{path}.
2495
return self.root.child(path).createDirectory()
2498
def fileExists(self, path):
2500
Test if the file exists at C{path}.
2502
return self.root.child(path).isfile()
2505
def createFile(self, path, fileContent=''):
2507
Create a file named C{path} with some content.
2509
return self.root.child(path).setContent(fileContent)
2513
class TestConsumer(object):
2515
A simple consumer for tests. It only works with non-streaming producers.
2517
@ivar producer: an object providing
2518
L{twisted.internet.interfaces.IPullProducer}.
2521
implements(IConsumer)
2524
def registerProducer(self, producer, streaming):
2526
Simple register of producer, checks that no register has happened
2529
assert self.producer is None
2531
self.producer = producer
2532
self.producer.resumeProducing()
2535
def unregisterProducer(self):
2537
Unregister the producer, it should be done after a register.
2539
assert self.producer is not None
2540
self.producer = None
2543
def write(self, data):
2545
Save the data received.
2547
self.buffer.append(data)
2548
self.producer.resumeProducing()
2552
class TestProducer(object):
2557
def __init__(self, toProduce, consumer):
2559
@param toProduce: data to write
2560
@type toProduce: C{str}
2561
@param consumer: the consumer of data.
2562
@type consumer: C{IConsumer}
2564
self.toProduce = toProduce
2565
self.consumer = consumer
2570
Send the data to consume.
2572
self.consumer.write(self.toProduce)
2576
class IReadWriteTestsMixin:
2578
Generic tests for the C{IReadFile} and C{IWriteFile} interfaces.
2581
def getFileReader(self, content):
2583
Return an object providing C{IReadFile}, ready to send data C{content}.
2585
raise NotImplementedError()
2588
def getFileWriter(self):
2590
Return an object providing C{IWriteFile}, ready to receive data.
2592
raise NotImplementedError()
2595
def getFileContent(self):
2597
Return the content of the file used.
2599
raise NotImplementedError()
2602
def test_read(self):
2604
Test L{ftp.IReadFile}: the implementation should have a send method
2605
returning a C{Deferred} which fires when all the data has been sent
2606
to the consumer, and the data should be correctly send to the consumer.
2608
content = 'wobble\n'
2609
consumer = TestConsumer()
2611
return reader.send(consumer).addCallback(cbSend)
2613
self.assertEquals("".join(consumer.buffer), content)
2614
return self.getFileReader(content).addCallback(cbGet)
2617
def test_write(self):
2619
Test L{ftp.IWriteFile}: the implementation should have a receive method
2620
returning a C{Deferred} with fires with a consumer ready to receive
2623
content = 'elbbow\n'
2625
return writer.receive().addCallback(cbReceive)
2626
def cbReceive(consumer):
2627
producer = TestProducer(content, consumer)
2628
consumer.registerProducer(None, True)
2630
consumer.unregisterProducer()
2631
self.assertEquals(self.getFileContent(), content)
2632
return self.getFileWriter().addCallback(cbGet)
2636
class FTPReadWriteTestCase(unittest.TestCase, IReadWriteTestsMixin):
2638
Tests for C{ftp._FileReader} and C{ftp._FileWriter}, the objects returned
2639
by the shell in C{openForReading}/C{openForWriting}.
2644
Create a temporary file used later.
2646
self.root = filepath.FilePath(self.mktemp())
2647
self.root.createDirectory()
2648
self.shell = ftp.FTPShell(self.root)
2649
self.filename = "file.txt"
2652
def getFileReader(self, content):
2654
Return a C{ftp._FileReader} instance with a file opened for reading.
2656
self.root.child(self.filename).setContent(content)
2657
return self.shell.openForReading((self.filename,))
2660
def getFileWriter(self):
2662
Return a C{ftp._FileWriter} instance with a file opened for writing.
2664
return self.shell.openForWriting((self.filename,))
2667
def getFileContent(self):
2669
Return the content of the temporary file.
2671
return self.root.child(self.filename).getContent()
added
removed
vendor/Twisted-10.0.0/twisted/test/test_ftp.py
