← Back to branch summary

~certify-web-dev/twisted/certify-trunk

« back to all changes in this revision

Viewing changes to twisted/test/test_ftp.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2007-01-17 14:52:35 UTC
  • mfrom: (1.1.5 upstream) (2.1.2 etch)
  • Revision ID: james.westby@ubuntu.com-20070117145235-btmig6qfmqfen0om
Tags: 2.5.0-0ubuntu1
New upstream version, compatible with python2.5.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/test/test_ftp.py
18
18
from twisted.internet import reactor, protocol, defer, error
19
19
from twisted.cred import portal, checkers, credentials
20
20
from twisted.python import failure
 
21
from twisted.test import proto_helpers
21
22
 
22
23
from twisted.protocols import ftp, loopback
23
24
 
65
66
 
66
67
        # Start the server
67
68
        p = portal.Portal(ftp.FTPRealm(self.directory))
68
 
        p.registerChecker(checkers.AllowAnonymousAccess(), 
 
69
        p.registerChecker(checkers.AllowAnonymousAccess(),
69
70
                          credentials.IAnonymous)
70
71
        self.factory = ftp.FTPFactory(portal=p)
71
72
        self.port = reactor.listenTCP(0, self.factory, interface="127.0.0.1")
106
107
                              chainDeferred=None):
107
108
        """Asserts that a sending an FTP command receives the expected
108
109
        response.
109
 
        
 
110
 
110
111
        Returns a Deferred.  Optionally accepts a deferred to chain its actions
111
112
        to.
112
113
        """
113
 
        if chainDeferred is None: 
 
114
        if chainDeferred is None:
114
115
            chainDeferred = defer.succeed(None)
115
 
        
 
116
 
116
117
        def queueCommand(ignored):
117
118
            d = self.client.queueStringCommand(command)
118
119
            def gotResponse(responseLines):
122
123
 
123
124
    def assertCommandFailed(self, command, expectedResponse=None,
124
125
                            chainDeferred=None):
125
 
        if chainDeferred is None: 
 
126
        if chainDeferred is None:
126
127
            chainDeferred = defer.succeed(None)
127
128
 
128
129
        def queueCommand(ignored):
143
144
            'PASS test@twistedmatrix.com',
144
145
            ['230 Anonymous login ok, access restrictions apply.'],
145
146
            chainDeferred=d)
146
 
        
 
147
 
147
148
 
148
149
class BasicFTPServerTestCase(FTPServerTestCase):
149
150
    def testNotLoggedInReply(self):
228
229
    def testUnknownCommand(self):
229
230
        d = self._anonymousLogin()
230
231
        return self.assertCommandFailed(
231
 
            'GIBBERISH', 
 
232
            'GIBBERISH',
232
233
            ["502 Command 'GIBBERISH' not implemented"],
233
234
            chainDeferred=d)
234
235
 
235
236
    def testRETRBeforePORT(self):
236
237
        d = self._anonymousLogin()
237
238
        return self.assertCommandFailed(
238
 
            'RETR foo', 
 
239
            'RETR foo',
239
240
            ["503 Incorrect sequence of commands: "
240
241
             "PORT or PASV required before RETR"],
241
242
            chainDeferred=d)
415
416
        d = defer.maybeDeferred(
416
417
            FTPServerPasvDataConnectionTestCase.tearDown, self)
417
418
        l.append(d)
418
 
        return defer.DeferredList(l, fireOnOneErrback=True) 
 
419
        return defer.DeferredList(l, fireOnOneErrback=True)
419
420
 
420
421
    def testPORTCannotConnect(self):
421
422
        # Login
440
441
                ["425 Can't open data connection."])
441
442
        return d.addCallback(gotPortNum)
442
443
 
443
 
        
 
444
 
444
445
# -- Client Tests -----------------------------------------------------------
445
446
 
446
447
class PrintLines(protocol.Protocol):
467
468
class FTPFileListingTests(unittest.TestCase):
468
469
    def getFilesForLines(self, lines):
469
470
        fileList = MyFTPFileListProtocol()
470
 
        loopback.loopback(PrintLines(lines), fileList)
471
 
        return fileList.files, fileList.other
 
471
        d = loopback.loopbackAsync(PrintLines(lines), fileList)
 
472
        d.addCallback(lambda _: (fileList.files, fileList.other))
 
473
        return d
472
474
 
473
475
    def testOneLine(self):
474
476
        # This example line taken from the docstring for FTPFileListProtocol
475
477
        line = '-rw-r--r--   1 root     other        531 Jan 29 03:26 README'
476
 
        (file,), other = self.getFilesForLines([line])
477
 
        self.failIf(other, 'unexpect unparsable lines: %s' % repr(other))
478
 
        self.failUnless(file['filetype'] == '-', 'misparsed fileitem')
479
 
        self.failUnless(file['perms'] == 'rw-r--r--', 'misparsed perms')
480
 
        self.failUnless(file['owner'] == 'root', 'misparsed fileitem')
481
 
        self.failUnless(file['group'] == 'other', 'misparsed fileitem')
482
 
        self.failUnless(file['size'] == 531, 'misparsed fileitem')
483
 
        self.failUnless(file['date'] == 'Jan 29 03:26', 'misparsed fileitem')
484
 
        self.failUnless(file['filename'] == 'README', 'misparsed fileitem')
485
 
        self.failUnless(file['nlinks'] == 1, 'misparsed nlinks')
486
 
        self.failIf(file['linktarget'], 'misparsed linktarget')
 
478
        def check(((file,), other)):
 
479
            self.failIf(other, 'unexpect unparsable lines: %s' % repr(other))
 
480
            self.failUnless(file['filetype'] == '-', 'misparsed fileitem')
 
481
            self.failUnless(file['perms'] == 'rw-r--r--', 'misparsed perms')
 
482
            self.failUnless(file['owner'] == 'root', 'misparsed fileitem')
 
483
            self.failUnless(file['group'] == 'other', 'misparsed fileitem')
 
484
            self.failUnless(file['size'] == 531, 'misparsed fileitem')
 
485
            self.failUnless(file['date'] == 'Jan 29 03:26', 'misparsed fileitem')
 
486
            self.failUnless(file['filename'] == 'README', 'misparsed fileitem')
 
487
            self.failUnless(file['nlinks'] == 1, 'misparsed nlinks')
 
488
            self.failIf(file['linktarget'], 'misparsed linktarget')
 
489
        return self.getFilesForLines([line]).addCallback(check)
487
490
 
488
491
    def testVariantLines(self):
489
492
        line1 = 'drw-r--r--   2 root     other        531 Jan  9  2003 A'
490
493
        line2 = 'lrw-r--r--   1 root     other          1 Jan 29 03:26 B -> A'
491
494
        line3 = 'woohoo! '
492
 
        (file1, file2), (other,) = self.getFilesForLines([line1, line2, line3])
493
 
        self.failUnless(other == 'woohoo! \r', 'incorrect other line')
494
 
        # file 1
495
 
        self.failUnless(file1['filetype'] == 'd', 'misparsed fileitem')
496
 
        self.failUnless(file1['perms'] == 'rw-r--r--', 'misparsed perms')
497
 
        self.failUnless(file1['owner'] == 'root', 'misparsed owner')
498
 
        self.failUnless(file1['group'] == 'other', 'misparsed group')
499
 
        self.failUnless(file1['size'] == 531, 'misparsed size')
500
 
        self.failUnless(file1['date'] == 'Jan  9  2003', 'misparsed date')
501
 
        self.failUnless(file1['filename'] == 'A', 'misparsed filename')
502
 
        self.failUnless(file1['nlinks'] == 2, 'misparsed nlinks')
503
 
        self.failIf(file1['linktarget'], 'misparsed linktarget')
504
 
        # file 2
505
 
        self.failUnless(file2['filetype'] == 'l', 'misparsed fileitem')
506
 
        self.failUnless(file2['perms'] == 'rw-r--r--', 'misparsed perms')
507
 
        self.failUnless(file2['owner'] == 'root', 'misparsed owner')
508
 
        self.failUnless(file2['group'] == 'other', 'misparsed group')
509
 
        self.failUnless(file2['size'] == 1, 'misparsed size')
510
 
        self.failUnless(file2['date'] == 'Jan 29 03:26', 'misparsed date')
511
 
        self.failUnless(file2['filename'] == 'B', 'misparsed filename')
512
 
        self.failUnless(file2['nlinks'] == 1, 'misparsed nlinks')
513
 
        self.failUnless(file2['linktarget'] == 'A', 'misparsed linktarget')
 
495
        def check(((file1, file2), (other,))):
 
496
            self.failUnless(other == 'woohoo! \r', 'incorrect other line')
 
497
            # file 1
 
498
            self.failUnless(file1['filetype'] == 'd', 'misparsed fileitem')
 
499
            self.failUnless(file1['perms'] == 'rw-r--r--', 'misparsed perms')
 
500
            self.failUnless(file1['owner'] == 'root', 'misparsed owner')
 
501
            self.failUnless(file1['group'] == 'other', 'misparsed group')
 
502
            self.failUnless(file1['size'] == 531, 'misparsed size')
 
503
            self.failUnless(file1['date'] == 'Jan  9  2003', 'misparsed date')
 
504
            self.failUnless(file1['filename'] == 'A', 'misparsed filename')
 
505
            self.failUnless(file1['nlinks'] == 2, 'misparsed nlinks')
 
506
            self.failIf(file1['linktarget'], 'misparsed linktarget')
 
507
            # file 2
 
508
            self.failUnless(file2['filetype'] == 'l', 'misparsed fileitem')
 
509
            self.failUnless(file2['perms'] == 'rw-r--r--', 'misparsed perms')
 
510
            self.failUnless(file2['owner'] == 'root', 'misparsed owner')
 
511
            self.failUnless(file2['group'] == 'other', 'misparsed group')
 
512
            self.failUnless(file2['size'] == 1, 'misparsed size')
 
513
            self.failUnless(file2['date'] == 'Jan 29 03:26', 'misparsed date')
 
514
            self.failUnless(file2['filename'] == 'B', 'misparsed filename')
 
515
            self.failUnless(file2['nlinks'] == 1, 'misparsed nlinks')
 
516
            self.failUnless(file2['linktarget'] == 'A', 'misparsed linktarget')
 
517
        return self.getFilesForLines([line1, line2, line3]).addCallback(check)
514
518
 
515
519
    def testUnknownLine(self):
516
 
        files, others = self.getFilesForLines(['ABC', 'not a file'])
517
 
        self.failIf(files, 'unexpected file entries')
518
 
        self.failUnless(others == ['ABC\r', 'not a file\r'],
519
 
                        'incorrect unparsable lines: %s' % repr(others))
 
520
        def check((files, others)):
 
521
            self.failIf(files, 'unexpected file entries')
 
522
            self.failUnless(others == ['ABC\r', 'not a file\r'],
 
523
                            'incorrect unparsable lines: %s' % repr(others))
 
524
        return self.getFilesForLines(['ABC', 'not a file']).addCallback(check)
520
525
 
521
526
    def testYear(self):
522
527
        # This example derived from bug description in issue 514.
527
532
            def connectionMade(self):
528
533
                self.transport.write(exampleLine)
529
534
                self.transport.loseConnection()
530
 
        loopback.loopback(PrintLine(), fileList)
531
 
        file = fileList.files[0]
532
 
        self.failUnless(file['size'] == 531, 'misparsed fileitem')
533
 
        self.failUnless(file['date'] == 'Jan 29 2003', 'misparsed fileitem')
534
 
        self.failUnless(file['filename'] == 'README', 'misparsed fileitem')
 
535
 
 
536
        def check(ignored):
 
537
            file = fileList.files[0]
 
538
            self.failUnless(file['size'] == 531, 'misparsed fileitem')
 
539
            self.failUnless(file['date'] == 'Jan 29 2003', 'misparsed fileitem')
 
540
            self.failUnless(file['filename'] == 'README', 'misparsed fileitem')
 
541
 
 
542
        d = loopback.loopbackAsync(PrintLine(), fileList)
 
543
        return d.addCallback(check)
535
544
 
536
545
 
537
546
class FTPClientTests(unittest.TestCase):
540
549
        port = getattr(self, 'port', None)
541
550
        if port is not None:
542
551
            return port.stopListening()
543
 
            
 
552
 
544
553
    def testFailedRETR(self):
545
554
        f = protocol.Factory()
546
555
        f.noisy = 0
583
592
        self.failUnless(m, m)
584
593
 
585
594
 
 
595
 
 
596
class FTPClientTestCase(unittest.TestCase):
 
597
    """
 
598
    Test advanced FTP client commands.
 
599
    """
 
600
    def setUp(self):
 
601
        """
 
602
        Create a FTP client and connect it to fake transport.
 
603
        """
 
604
        self.client = ftp.FTPClient()
 
605
        self.transport = proto_helpers.StringTransport()
 
606
        self.client.makeConnection(self.transport)
 
607
 
 
608
 
 
609
    def tearDown(self):
 
610
        """
 
611
        Deliver disconnection notification to the client so that it can
 
612
        perform any cleanup which may be required.
 
613
        """
 
614
        self.client.connectionLost(error.ConnectionLost())
 
615
 
 
616
 
 
617
    def _testLogin(self):
 
618
        """
 
619
        Test the login part.
 
620
        """
 
621
        self.assertEquals(self.transport.value(), '')
 
622
        self.client.lineReceived(
 
623
            '331 Guest login ok, type your email address as password.')
 
624
        self.assertEquals(self.transport.value(), 'USER anonymous\r\n')
 
625
        self.transport.clear()
 
626
        self.client.lineReceived(
 
627
            '230 Anonymous login ok, access restrictions apply.')
 
628
        self.assertEquals(self.transport.value(), 'TYPE I\r\n')
 
629
        self.transport.clear()
 
630
        self.client.lineReceived('200 Type set to I.')
 
631
 
 
632
 
 
633
    def test_CDUP(self):
 
634
        """
 
635
        Test the CDUP command.
 
636
 
 
637
        L{ftp.FTPClient.cdup} should return a Deferred which fires with a
 
638
        sequence of one element which is the string the server sent
 
639
        indicating that the command was executed successfully.
 
640
 
 
641
        (XXX - This is a bad API)
 
642
        """
 
643
        def cbCdup(res):
 
644
            self.assertEquals(res[0], '250 Requested File Action Completed OK')
 
645
 
 
646
        self._testLogin()
 
647
        d = self.client.cdup().addCallback(cbCdup)
 
648
        self.assertEquals(self.transport.value(), 'CDUP\r\n')
 
649
        self.transport.clear()
 
650
        self.client.lineReceived('250 Requested File Action Completed OK')
 
651
        return d
 
652
 
 
653
 
 
654
    def test_failedCDUP(self):
 
655
        """
 
656
        Test L{ftp.FTPClient.cdup}'s handling of a failed CDUP command.
 
657
 
 
658
        When the CDUP command fails, the returned Deferred should errback
 
659
        with L{ftp.CommandFailed}.
 
660
        """
 
661
        self._testLogin()
 
662
        d = self.client.cdup()
 
663
        self.assertFailure(d, ftp.CommandFailed)
 
664
        self.assertEquals(self.transport.value(), 'CDUP\r\n')
 
665
        self.transport.clear()
 
666
        self.client.lineReceived('550 ..: No such file or directory')
 
667
        return d
 
668
 
 
669
 
 
670
    def test_PWD(self):
 
671
        """
 
672
        Test the PWD command.
 
673
 
 
674
        L{ftp.FTPClient.pwd} should return a Deferred which fires with a
 
675
        sequence of one element which is a string representing the current
 
676
        working directory on the server.
 
677
 
 
678
        (XXX - This is a bad API)
 
679
        """
 
680
        def cbPwd(res):
 
681
            self.assertEquals(ftp.parsePWDResponse(res[0]), "/bar/baz")
 
682
 
 
683
        self._testLogin()
 
684
        d = self.client.pwd().addCallback(cbPwd)
 
685
        self.assertEquals(self.transport.value(), 'PWD\r\n')
 
686
        self.client.lineReceived('257 "/bar/baz"')
 
687
        return d
 
688
 
 
689
 
 
690
    def test_failedPWD(self):
 
691
        """
 
692
        Test a failure in PWD command.
 
693
 
 
694
        When the PWD command fails, the returned Deferred should errback
 
695
        with L{ftp.CommandFailed}.
 
696
        """
 
697
        self._testLogin()
 
698
        d = self.client.pwd()
 
699
        self.assertFailure(d, ftp.CommandFailed)
 
700
        self.assertEquals(self.transport.value(), 'PWD\r\n')
 
701
        self.client.lineReceived('550 /bar/baz: No such file or directory')
 
702
        return d
 
703
 
 
704
 
 
705
    def test_CWD(self):
 
706
        """
 
707
        Test the CWD command.
 
708
 
 
709
        L{ftp.FTPClient.cwd} should return a Deferred which fires with a
 
710
        sequence of one element which is the string the server sent
 
711
        indicating that the command was executed successfully.
 
712
 
 
713
        (XXX - This is a bad API)
 
714
        """
 
715
        def cbCwd(res):
 
716
            self.assertEquals(res[0], '250 Requested File Action Completed OK')
 
717
 
 
718
        self._testLogin()
 
719
        d = self.client.cwd("bar/foo").addCallback(cbCwd)
 
720
        self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
 
721
        self.client.lineReceived('250 Requested File Action Completed OK')
 
722
        return d
 
723
 
 
724
 
 
725
    def test_failedCWD(self):
 
726
        """
 
727
        Test a failure in CWD command.
 
728
 
 
729
        When the PWD command fails, the returned Deferred should errback
 
730
        with L{ftp.CommandFailed}.
 
731
        """
 
732
        self._testLogin()
 
733
        d = self.client.cwd("bar/foo")
 
734
        self.assertFailure(d, ftp.CommandFailed)
 
735
        self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
 
736
        self.client.lineReceived('550 bar/foo: No such file or directory')
 
737
        return d
 
738
 
 
739
 
 
740
    def test_passiveRETR(self):
 
741
        """
 
742
        Test the RETR command in passive mode: get a file and verify its
 
743
        content.
 
744
 
 
745
        L{ftp.FTPClient.retrieveFile} should return a Deferred which fires
 
746
        with the protocol instance passed to it after the download has
 
747
        completed.
 
748
 
 
749
        (XXX - This API should be based on producers and consumers)
 
750
        """
 
751
        def cbRetr(res, proto):
 
752
            self.assertEquals(proto.buffer, 'x' * 1000)
 
753
 
 
754
        def cbConnect(host, port, factory):
 
755
            self.assertEquals(host, '127.0.0.1')
 
756
            self.assertEquals(port, 12345)
 
757
            proto = factory.buildProtocol((host, port))
 
758
            proto.makeConnection(proto_helpers.StringTransport())
 
759
            self.client.lineReceived(
 
760
                '150 File status okay; about to open data connection.')
 
761
            proto.dataReceived("x" * 1000)
 
762
            proto.connectionLost(failure.Failure(error.ConnectionDone("")))
 
763
 
 
764
        self.client.connectFactory = cbConnect
 
765
        self._testLogin()
 
766
        proto = _BufferingProtocol()
 
767
        d = self.client.retrieveFile("spam", proto)
 
768
        d.addCallback(cbRetr, proto)
 
769
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
770
        self.transport.clear()
 
771
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
772
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
773
        self.assertEquals(self.transport.value(), 'RETR spam\r\n')
 
774
        self.transport.clear()
 
775
        self.client.lineReceived('226 Transfer Complete.')
 
776
        return d
 
777
 
 
778
 
 
779
    def test_RETR(self):
 
780
        """
 
781
        Test the RETR command in non-passive mode.
 
782
 
 
783
        Like L{test_passiveRETR} but in the configuration where the server
 
784
        establishes the data connection to the client, rather than the other
 
785
        way around.
 
786
        """
 
787
        self.client.passive = False
 
788
 
 
789
        def generatePort(portCmd):
 
790
            portCmd.text = 'PORT %s' % (ftp.encodeHostPort('127.0.0.1', 9876),)
 
791
            portCmd.protocol.makeConnection(proto_helpers.StringTransport())
 
792
            portCmd.protocol.dataReceived("x" * 1000)
 
793
            portCmd.protocol.connectionLost(
 
794
                failure.Failure(error.ConnectionDone("")))
 
795
 
 
796
        def cbRetr(res, proto):
 
797
            self.assertEquals(proto.buffer, 'x' * 1000)
 
798
 
 
799
        self.client.generatePortCommand = generatePort
 
800
        self._testLogin()
 
801
        proto = _BufferingProtocol()
 
802
        d = self.client.retrieveFile("spam", proto)
 
803
        d.addCallback(cbRetr, proto)
 
804
        self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
 
805
            (ftp.encodeHostPort('127.0.0.1', 9876),))
 
806
        self.transport.clear()
 
807
        self.client.lineReceived('200 PORT OK')
 
808
        self.assertEquals(self.transport.value(), 'RETR spam\r\n')
 
809
        self.transport.clear()
 
810
        self.client.lineReceived('226 Transfer Complete.')
 
811
        return d
 
812
 
 
813
 
 
814
    def test_failedRETR(self):
 
815
        """
 
816
        Try to RETR an unexisting file.
 
817
 
 
818
        L{ftp.FTPClient.retrieveFile} should return a Deferred which
 
819
        errbacks with L{ftp.CommandFailed} if the server indicates the file
 
820
        cannot be transferred for some reason.
 
821
        """
 
822
        def cbConnect(host, port, factory):
 
823
            self.assertEquals(host, '127.0.0.1')
 
824
            self.assertEquals(port, 12345)
 
825
            proto = factory.buildProtocol((host, port))
 
826
            proto.makeConnection(proto_helpers.StringTransport())
 
827
            self.client.lineReceived(
 
828
                '150 File status okay; about to open data connection.')
 
829
            proto.connectionLost(failure.Failure(error.ConnectionDone("")))
 
830
 
 
831
        self.client.connectFactory = cbConnect
 
832
        self._testLogin()
 
833
        proto = _BufferingProtocol()
 
834
        d = self.client.retrieveFile("spam", proto)
 
835
        self.assertFailure(d, ftp.CommandFailed)
 
836
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
837
        self.transport.clear()
 
838
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
839
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
840
        self.assertEquals(self.transport.value(), 'RETR spam\r\n')
 
841
        self.transport.clear()
 
842
        self.client.lineReceived('550 spam: No such file or directory')
 
843
        return d
 
844
 
 
845
 
 
846
    def test_passiveSTOR(self):
 
847
        """
 
848
        Test the STOR command: send a file and verify its content.
 
849
 
 
850
        L{ftp.FTPClient.storeFile} should return a two-tuple of Deferreds.
 
851
        The first of which should fire with a protocol instance when the
 
852
        data connection has been established and is responsible for sending
 
853
        the contents of the file.  The second of which should fire when the
 
854
        upload has completed, the data connection has been closed, and the
 
855
        server has acknowledged receipt of the file.
 
856
 
 
857
        (XXX - storeFile should take a producer as an argument, instead, and
 
858
        only return a Deferred which fires when the upload has succeeded or
 
859
        failed).
 
860
        """
 
861
        tr = proto_helpers.StringTransport()
 
862
        def cbStore(sender):
 
863
            self.client.lineReceived(
 
864
                '150 File status okay; about to open data connection.')
 
865
            sender.transport.write("x" * 1000)
 
866
            sender.finish()
 
867
            sender.connectionLost(failure.Failure(error.ConnectionDone("")))
 
868
 
 
869
        def cbFinish(ign):
 
870
            self.assertEquals(tr.value(), "x" * 1000)
 
871
 
 
872
        def cbConnect(host, port, factory):
 
873
            self.assertEquals(host, '127.0.0.1')
 
874
            self.assertEquals(port, 12345)
 
875
            proto = factory.buildProtocol((host, port))
 
876
            proto.makeConnection(tr)
 
877
 
 
878
        self.client.connectFactory = cbConnect
 
879
        self._testLogin()
 
880
        d1, d2 = self.client.storeFile("spam")
 
881
        d1.addCallback(cbStore)
 
882
        d2.addCallback(cbFinish)
 
883
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
884
        self.transport.clear()
 
885
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
886
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
887
        self.assertEquals(self.transport.value(), 'STOR spam\r\n')
 
888
        self.transport.clear()
 
889
        self.client.lineReceived('226 Transfer Complete.')
 
890
        return defer.gatherResults([d1, d2])
 
891
 
 
892
 
 
893
    def test_failedSTOR(self):
 
894
        """
 
895
        Test a failure in the STOR command.
 
896
 
 
897
        If the server does not acknowledge successful receipt of the
 
898
        uploaded file, the second Deferred returned by
 
899
        L{ftp.FTPClient.storeFile} should errback with L{ftp.CommandFailed}.
 
900
        """
 
901
        tr = proto_helpers.StringTransport()
 
902
        def cbStore(sender):
 
903
            self.client.lineReceived(
 
904
                '150 File status okay; about to open data connection.')
 
905
            sender.transport.write("x" * 1000)
 
906
            sender.finish()
 
907
            sender.connectionLost(failure.Failure(error.ConnectionDone("")))
 
908
 
 
909
        def cbConnect(host, port, factory):
 
910
            self.assertEquals(host, '127.0.0.1')
 
911
            self.assertEquals(port, 12345)
 
912
            proto = factory.buildProtocol((host, port))
 
913
            proto.makeConnection(tr)
 
914
 
 
915
        self.client.connectFactory = cbConnect
 
916
        self._testLogin()
 
917
        d1, d2 = self.client.storeFile("spam")
 
918
        d1.addCallback(cbStore)
 
919
        self.assertFailure(d2, ftp.CommandFailed)
 
920
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
921
        self.transport.clear()
 
922
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
923
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
924
        self.assertEquals(self.transport.value(), 'STOR spam\r\n')
 
925
        self.transport.clear()
 
926
        self.client.lineReceived(
 
927
            '426 Transfer aborted.  Data connection closed.')
 
928
        return defer.gatherResults([d1, d2])
 
929
 
 
930
 
 
931
    def test_STOR(self):
 
932
        """
 
933
        Test the STOR command in non-passive mode.
 
934
 
 
935
        Like L{test_passiveSTOR} but in the configuration where the server
 
936
        establishes the data connection to the client, rather than the other
 
937
        way around.
 
938
        """
 
939
        tr = proto_helpers.StringTransport()
 
940
        self.client.passive = False
 
941
        def generatePort(portCmd):
 
942
            portCmd.text = 'PORT %s' % ftp.encodeHostPort('127.0.0.1', 9876)
 
943
            portCmd.protocol.makeConnection(tr)
 
944
 
 
945
        def cbStore(sender):
 
946
            self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
 
947
                (ftp.encodeHostPort('127.0.0.1', 9876),))
 
948
            self.transport.clear()
 
949
            self.client.lineReceived('200 PORT OK')
 
950
            self.assertEquals(self.transport.value(), 'STOR spam\r\n')
 
951
            self.transport.clear()
 
952
            self.client.lineReceived(
 
953
                '150 File status okay; about to open data connection.')
 
954
            sender.transport.write("x" * 1000)
 
955
            sender.finish()
 
956
            sender.connectionLost(failure.Failure(error.ConnectionDone("")))
 
957
            self.client.lineReceived('226 Transfer Complete.')
 
958
 
 
959
        def cbFinish(ign):
 
960
            self.assertEquals(tr.value(), "x" * 1000)
 
961
 
 
962
        self.client.generatePortCommand = generatePort
 
963
        self._testLogin()
 
964
        d1, d2 = self.client.storeFile("spam")
 
965
        d1.addCallback(cbStore)
 
966
        d2.addCallback(cbFinish)
 
967
        return defer.gatherResults([d1, d2])
 
968
 
 
969
 
 
970
    def test_passiveLIST(self):
 
971
        """
 
972
        Test the LIST command.
 
973
 
 
974
        L{ftp.FTPClient.list} should return a Deferred which fires with a
 
975
        protocol instance which was passed to list after the command has
 
976
        succeeded.
 
977
 
 
978
        (XXX - This is a very unfortunate API; if my understanding is
 
979
        correct, the results are always at least line-oriented, so allowing
 
980
        a per-line parser function to be specified would make this simpler,
 
981
        but a default implementation should really be provided which knows
 
982
        how to deal with all the formats used in real servers, so
 
983
        application developers never have to care about this insanity.  It
 
984
        would also be nice to either get back a Deferred of a list of
 
985
        filenames or to be able to consume the files as they are received
 
986
        (which the current API does allow, but in a somewhat inconvenient
 
987
        fashion) -exarkun)
 
988
        """
 
989
        def cbList(res, fileList):
 
990
            fls = [f["filename"] for f in fileList.files]
 
991
            expected = ["foo", "bar", "baz"]
 
992
            expected.sort()
 
993
            fls.sort()
 
994
            self.assertEquals(fls, expected)
 
995
 
 
996
        def cbConnect(host, port, factory):
 
997
            self.assertEquals(host, '127.0.0.1')
 
998
            self.assertEquals(port, 12345)
 
999
            proto = factory.buildProtocol((host, port))
 
1000
            proto.makeConnection(proto_helpers.StringTransport())
 
1001
            self.client.lineReceived(
 
1002
                '150 File status okay; about to open data connection.')
 
1003
            sending = [
 
1004
                '-rw-r--r--    0 spam      egg      100 Oct 10 2006 foo\r\n',
 
1005
                '-rw-r--r--    3 spam      egg      100 Oct 10 2006 bar\r\n',
 
1006
                '-rw-r--r--    4 spam      egg      100 Oct 10 2006 baz\r\n',
 
1007
            ]
 
1008
            for i in sending:
 
1009
                proto.dataReceived(i)
 
1010
            proto.connectionLost(failure.Failure(error.ConnectionDone("")))
 
1011
 
 
1012
        self.client.connectFactory = cbConnect
 
1013
        self._testLogin()
 
1014
        fileList = ftp.FTPFileListProtocol()
 
1015
        d = self.client.list('foo/bar', fileList).addCallback(cbList, fileList)
 
1016
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
1017
        self.transport.clear()
 
1018
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
1019
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
1020
        self.assertEquals(self.transport.value(), 'LIST foo/bar\r\n')
 
1021
        self.client.lineReceived('226 Transfer Complete.')
 
1022
        return d
 
1023
 
 
1024
 
 
1025
    def test_LIST(self):
 
1026
        """
 
1027
        Test the LIST command in non-passive mode.
 
1028
 
 
1029
        Like L{test_passiveLIST} but in the configuration where the server
 
1030
        establishes the data connection to the client, rather than the other
 
1031
        way around.
 
1032
        """
 
1033
        self.client.passive = False
 
1034
        def generatePort(portCmd):
 
1035
            portCmd.text = 'PORT %s' % (ftp.encodeHostPort('127.0.0.1', 9876),)
 
1036
            portCmd.protocol.makeConnection(proto_helpers.StringTransport())
 
1037
            self.client.lineReceived(
 
1038
                '150 File status okay; about to open data connection.')
 
1039
            sending = [
 
1040
                '-rw-r--r--    0 spam      egg      100 Oct 10 2006 foo\r\n',
 
1041
                '-rw-r--r--    3 spam      egg      100 Oct 10 2006 bar\r\n',
 
1042
                '-rw-r--r--    4 spam      egg      100 Oct 10 2006 baz\r\n',
 
1043
            ]
 
1044
            for i in sending:
 
1045
                portCmd.protocol.dataReceived(i)
 
1046
            portCmd.protocol.connectionLost(
 
1047
                failure.Failure(error.ConnectionDone("")))
 
1048
 
 
1049
        def cbList(res, fileList):
 
1050
            fls = [f["filename"] for f in fileList.files]
 
1051
            expected = ["foo", "bar", "baz"]
 
1052
            expected.sort()
 
1053
            fls.sort()
 
1054
            self.assertEquals(fls, expected)
 
1055
 
 
1056
        self.client.generatePortCommand = generatePort
 
1057
        self._testLogin()
 
1058
        fileList = ftp.FTPFileListProtocol()
 
1059
        d = self.client.list('foo/bar', fileList).addCallback(cbList, fileList)
 
1060
        self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
 
1061
            (ftp.encodeHostPort('127.0.0.1', 9876),))
 
1062
        self.transport.clear()
 
1063
        self.client.lineReceived('200 PORT OK')
 
1064
        self.assertEquals(self.transport.value(), 'LIST foo/bar\r\n')
 
1065
        self.transport.clear()
 
1066
        self.client.lineReceived('226 Transfer Complete.')
 
1067
        return d
 
1068
 
 
1069
 
 
1070
    def test_failedLIST(self):
 
1071
        """
 
1072
        Test a failure in LIST command.
 
1073
 
 
1074
        L{ftp.FTPClient.list} should return a Deferred which fails with
 
1075
        L{ftp.CommandFailed} if the server indicates the indicated path is
 
1076
        invalid for some reason.
 
1077
        """
 
1078
        def cbConnect(host, port, factory):
 
1079
            self.assertEquals(host, '127.0.0.1')
 
1080
            self.assertEquals(port, 12345)
 
1081
            proto = factory.buildProtocol((host, port))
 
1082
            proto.makeConnection(proto_helpers.StringTransport())
 
1083
            self.client.lineReceived(
 
1084
                '150 File status okay; about to open data connection.')
 
1085
            proto.connectionLost(failure.Failure(error.ConnectionDone("")))
 
1086
 
 
1087
        self.client.connectFactory = cbConnect
 
1088
        self._testLogin()
 
1089
        fileList = ftp.FTPFileListProtocol()
 
1090
        d = self.client.list('foo/bar', fileList)
 
1091
        self.assertFailure(d, ftp.CommandFailed)
 
1092
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
1093
        self.transport.clear()
 
1094
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
1095
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
1096
        self.assertEquals(self.transport.value(), 'LIST foo/bar\r\n')
 
1097
        self.client.lineReceived('550 foo/bar: No such file or directory')
 
1098
        return d
 
1099
 
 
1100
 
 
1101
    def test_NLST(self):
 
1102
        """
 
1103
        Test the NLST command in non-passive mode.
 
1104
 
 
1105
        L{ftp.FTPClient.nlst} should return a Deferred which fires with a
 
1106
        list of filenames when the list command has completed.
 
1107
        """
 
1108
        self.client.passive = False
 
1109
        def generatePort(portCmd):
 
1110
            portCmd.text = 'PORT %s' % (ftp.encodeHostPort('127.0.0.1', 9876),)
 
1111
            portCmd.protocol.makeConnection(proto_helpers.StringTransport())
 
1112
            self.client.lineReceived(
 
1113
                '150 File status okay; about to open data connection.')
 
1114
            portCmd.protocol.dataReceived('foo\r\n')
 
1115
            portCmd.protocol.dataReceived('bar\r\n')
 
1116
            portCmd.protocol.dataReceived('baz\r\n')
 
1117
            portCmd.protocol.connectionLost(
 
1118
                failure.Failure(error.ConnectionDone("")))
 
1119
 
 
1120
        def cbList(res, proto):
 
1121
            fls = proto.buffer.splitlines()
 
1122
            expected = ["foo", "bar", "baz"]
 
1123
            expected.sort()
 
1124
            fls.sort()
 
1125
            self.assertEquals(fls, expected)
 
1126
 
 
1127
        self.client.generatePortCommand = generatePort
 
1128
        self._testLogin()
 
1129
        lstproto = _BufferingProtocol()
 
1130
        d = self.client.nlst('foo/bar', lstproto).addCallback(cbList, lstproto)
 
1131
        self.assertEquals(self.transport.value(), 'PORT %s\r\n' %
 
1132
            (ftp.encodeHostPort('127.0.0.1', 9876),))
 
1133
        self.transport.clear()
 
1134
        self.client.lineReceived('200 PORT OK')
 
1135
        self.assertEquals(self.transport.value(), 'NLST foo/bar\r\n')
 
1136
        self.client.lineReceived('226 Transfer Complete.')
 
1137
        return d
 
1138
 
 
1139
 
 
1140
    def test_passiveNLST(self):
 
1141
        """
 
1142
        Test the NLST command.
 
1143
 
 
1144
        Like L{test_passiveNLST} but in the configuration where the server
 
1145
        establishes the data connection to the client, rather than the other
 
1146
        way around.
 
1147
        """
 
1148
        def cbList(res, proto):
 
1149
            fls = proto.buffer.splitlines()
 
1150
            expected = ["foo", "bar", "baz"]
 
1151
            expected.sort()
 
1152
            fls.sort()
 
1153
            self.assertEquals(fls, expected)
 
1154
 
 
1155
        def cbConnect(host, port, factory):
 
1156
            self.assertEquals(host, '127.0.0.1')
 
1157
            self.assertEquals(port, 12345)
 
1158
            proto = factory.buildProtocol((host, port))
 
1159
            proto.makeConnection(proto_helpers.StringTransport())
 
1160
            self.client.lineReceived(
 
1161
                '150 File status okay; about to open data connection.')
 
1162
            proto.dataReceived('foo\r\n')
 
1163
            proto.dataReceived('bar\r\n')
 
1164
            proto.dataReceived('baz\r\n')
 
1165
            proto.connectionLost(failure.Failure(error.ConnectionDone("")))
 
1166
 
 
1167
        self.client.connectFactory = cbConnect
 
1168
        self._testLogin()
 
1169
        lstproto = _BufferingProtocol()
 
1170
        d = self.client.nlst('foo/bar', lstproto).addCallback(cbList, lstproto)
 
1171
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
1172
        self.transport.clear()
 
1173
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
1174
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
1175
        self.assertEquals(self.transport.value(), 'NLST foo/bar\r\n')
 
1176
        self.client.lineReceived('226 Transfer Complete.')
 
1177
        return d
 
1178
 
 
1179
 
 
1180
    def test_failedNLST(self):
 
1181
        """
 
1182
        Test a failure in NLST command.
 
1183
 
 
1184
        L{ftp.FTPClient.nlst} should return a Deferred which fails with
 
1185
        L{ftp.CommandFailed} if the server indicates the indicated path is
 
1186
        invalid for some reason.
 
1187
        """
 
1188
        tr = proto_helpers.StringTransport()
 
1189
        def cbConnect(host, port, factory):
 
1190
            self.assertEquals(host, '127.0.0.1')
 
1191
            self.assertEquals(port, 12345)
 
1192
            proto = factory.buildProtocol((host, port))
 
1193
            proto.makeConnection(tr)
 
1194
            self.client.lineReceived(
 
1195
                '150 File status okay; about to open data connection.')
 
1196
            proto.connectionLost(failure.Failure(error.ConnectionDone("")))
 
1197
 
 
1198
        self.client.connectFactory = cbConnect
 
1199
        self._testLogin()
 
1200
        lstproto = _BufferingProtocol()
 
1201
        d = self.client.nlst('foo/bar', lstproto)
 
1202
        self.assertFailure(d, ftp.CommandFailed)
 
1203
        self.assertEquals(self.transport.value(), 'PASV\r\n')
 
1204
        self.transport.clear()
 
1205
        self.client.lineReceived('227 Entering Passive Mode (%s).' %
 
1206
            (ftp.encodeHostPort('127.0.0.1', 12345),))
 
1207
        self.assertEquals(self.transport.value(), 'NLST foo/bar\r\n')
 
1208
        self.client.lineReceived('550 foo/bar: No such file or directory')
 
1209
        return d
 
1210
 
 
1211
 
 
1212
    def test_changeDirectory(self):
 
1213
        """
 
1214
        Test the changeDirectory method.
 
1215
 
 
1216
        L{ftp.FTPClient.changeDirectory} should return a Deferred which fires
 
1217
        with True if succeeded.
 
1218
        """
 
1219
        def cbCd(res):
 
1220
            self.assertEquals(res, True)
 
1221
 
 
1222
        self._testLogin()
 
1223
        d = self.client.changeDirectory("bar/foo").addCallback(cbCd)
 
1224
        self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
 
1225
        self.client.lineReceived('250 Requested File Action Completed OK')
 
1226
        return d
 
1227
 
 
1228
 
 
1229
    def test_failedChangeDirectory(self):
 
1230
        """
 
1231
        Test a failure in the changeDirectory method.
 
1232
 
 
1233
        The behaviour here is the same as a failed CWD.
 
1234
        """
 
1235
        self._testLogin()
 
1236
        d = self.client.changeDirectory("bar/foo")
 
1237
        self.assertFailure(d, ftp.CommandFailed)
 
1238
        self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
 
1239
        self.client.lineReceived('550 bar/foo: No such file or directory')
 
1240
        return d
 
1241
 
 
1242
 
 
1243
    def test_strangeFailedChangeDirectory(self):
 
1244
        """
 
1245
        Test a strange failure in changeDirectory method.
 
1246
 
 
1247
        L{ftp.FTPClient.changeDirectory} is stricter than CWD as it checks
 
1248
        code 250 for success.
 
1249
        """
 
1250
        self._testLogin()
 
1251
        d = self.client.changeDirectory("bar/foo")
 
1252
        self.assertFailure(d, ftp.CommandFailed)
 
1253
        self.assertEquals(self.transport.value(), 'CWD bar/foo\r\n')
 
1254
        self.client.lineReceived('252 I do what I want !')
 
1255
        return d
 
1256
 
 
1257
 
 
1258
    def test_getDirectory(self):
 
1259
        """
 
1260
        Test the getDirectory method.
 
1261
 
 
1262
        L{ftp.FTPClient.getDirectory} should return a Deferred which fires with
 
1263
        the current directory on the server. It wraps PWD command.
 
1264
        """
 
1265
        def cbGet(res):
 
1266
            self.assertEquals(res, "/bar/baz")
 
1267
 
 
1268
        self._testLogin()
 
1269
        d = self.client.getDirectory().addCallback(cbGet)
 
1270
        self.assertEquals(self.transport.value(), 'PWD\r\n')
 
1271
        self.client.lineReceived('257 "/bar/baz"')
 
1272
        return d
 
1273
 
 
1274
 
 
1275
    def test_failedGetDirectory(self):
 
1276
        """
 
1277
        Test a failure in getDirectory method.
 
1278
 
 
1279
        The behaviour should be the same as PWD.
 
1280
        """
 
1281
        self._testLogin()
 
1282
        d = self.client.getDirectory()
 
1283
        self.assertFailure(d, ftp.CommandFailed)
 
1284
        self.assertEquals(self.transport.value(), 'PWD\r\n')
 
1285
        self.client.lineReceived('550 /bar/baz: No such file or directory')
 
1286
        return d
 
1287
 
 
1288
 
 
1289
    def test_anotherFailedGetDirectory(self):
 
1290
        """
 
1291
        Test a different failure in getDirectory method.
 
1292
 
 
1293
        The response should be quoted to be parsed, so it returns an error
 
1294
        otherwise.
 
1295
        """
 
1296
        self._testLogin()
 
1297
        d = self.client.getDirectory()
 
1298
        self.assertFailure(d, ftp.CommandFailed)
 
1299
        self.assertEquals(self.transport.value(), 'PWD\r\n')
 
1300
        self.client.lineReceived('257 /bar/baz')
 
1301
        return d
 
1302
 
 
1303
 
 
1304
 
586
1305
class DummyTransport:
587
1306
    def write(self, bytes):
588
1307
        pass