Ticket #4819: test-ftp-server-rest-appe.patch

File test-ftp-server-rest-appe.patch, 5.7 KB (added by eternicode, 6 years ago)
  • twisted/test/test_ftp.py

     
    6464        self.buffer += data
    6565    def connectionLost(self, reason):
    6666        self.d.callback(self)
     67    def write(self, content):
     68        self.transport.write(content)
     69        self.transport.loseConnection()
    6770
    6871
    6972
     
    8285        os.mkdir(self.directory)
    8386        self.dirPath = filepath.FilePath(self.directory)
    8487
     88        # Create a "/home" directory and home for "user"
     89        self.homeDirectory = self.mktemp()
     90        os.mkdir(self.homeDirectory)
     91        self.homeDirPath = filepath.FilePath(self.homeDirectory)
     92        self.userHome = self.homeDirPath.child('user')
     93        if not self.userHome.exists():
     94          self.userHome.createDirectory()
     95
    8596        # Start the server
    86         p = portal.Portal(ftp.FTPRealm(self.directory))
     97        p = portal.Portal(ftp.FTPRealm(self.directory, self.homeDirectory))
    8798        p.registerChecker(checkers.AllowAnonymousAccess(),
    8899                          credentials.IAnonymous)
     100        p.registerChecker(checkers.InMemoryUsernamePasswordDatabaseDontUse(
     101                              user='password'),
     102                          credentials.IUsernamePassword)
    89103        self.factory = ftp.FTPFactory(portal=p,
    90104                                      userAnonymous=self.userAnonymous)
    91105        port = reactor.listenTCP(0, self.factory, interface="127.0.0.1")
     
    162176            chainDeferred=d)
    163177
    164178
     179    def _userLogin(self):
     180        # Log in
     181        d = self.assertCommandResponse(
     182                'USER user',
     183                ['331 Password required for user.'])
     184        return self.assertCommandResponse(
     185                'PASS password',
     186                ['230 User logged in, proceed'],
     187                chainDeferred=d)
    165188
     189
     190
    166191class FTPAnonymousTestCase(FTPServerTestCase):
    167192    """
    168193    Simple tests for an FTP server with different anonymous username.
     
    435460        return d.addCallback(gotPASV)
    436461
    437462    def _download(self, command, chainDeferred=None):
     463        # Send a command to the server and download the response from the
     464        # DTP connection
    438465        if chainDeferred is None:
    439466            chainDeferred = defer.succeed(None)
    440467
     
    451478            return downloader.buffer
    452479        return chainDeferred.addCallback(downloadDone)
    453480
     481
     482    def _upload(self, command, content, chainDeferred=None):
     483        # Send a command to the server and upload content through the DTP
     484        # connection
     485        if chainDeferred is None:
     486            chainDeferred = defer.succeed(None)
     487
     488        chainDeferred.addCallback(self._makeDataConnection)
     489        def queueCommand(uploader):
     490            # wait for the command to return, send content immediately
     491            d1 = self.client.queueStringCommand(command)
     492            uploader.write(content)
     493            return d1
     494        return chainDeferred.addCallback(queueCommand)
     495
     496
    454497    def testEmptyLIST(self):
    455498        # Login
    456499        d = self._anonymousLogin()
     
    570613        return d.addCallback(checkDownload)
    571614
    572615
     616    def test_RESTContinuedDownload(self):
     617        """
     618        REST on an existing file with a positive offset followed by a RETR for
     619        that file will return that file's content starting at that offset.
     620        """
     621        # Login
     622        d = self._anonymousLogin()
    573623
     624        # Create some content in a file in the current working directory
     625        readfile = self.dirPath.child('test.txt').open('w')
     626        readfile.write('x' * 1000)
     627        readfile.close()
     628
     629        d = self.assertCommandResponse(
     630                'REST 500',
     631                ['350 Requested file action pending further information.'],
     632                chainDeferred=d)
     633
     634        d = self._download('RETR test.txt', chainDeferred=d)
     635        def checkDownload(download):
     636          self.assertEquals('x' * 500, download)
     637        return d.addCallback(checkDownload)
     638    test_RESTContinuedDownload.todo = "Need to implement REST server-side"
     639
     640
     641    def test_APPEContinuedUpload(self):
     642        """
     643        APPE on an existing file appends sent data to that file.
     644        """
     645        # Login
     646        d = self._userLogin()
     647
     648        # Create some content in a file in the home directory
     649        appendfile = self.userHome.child('test.txt').open('w')
     650        appendfile.write('x' * 500)
     651        appendfile.close()
     652
     653        d = self._upload('APPE test.txt', 'x' * 500, chainDeferred=d)
     654
     655        def checkContents(ignored=None):
     656          appendfile = self.userHome.child('test.txt').open('r')
     657          contents = appendfile.read()
     658          appendfile.close()
     659          self.assertEquals('x' * 1000, contents)
     660        return d.addCallback(checkContents)
     661    test_APPEContinuedUpload.todo = "Need to implement APPE server-side"
     662
     663
     664    def test_STORUploadFile(self):
     665        """
     666        STOR on a non-existing file saves sent data to the that file.
     667        """
     668        # Login
     669        d = self._userLogin()
     670
     671        # Ensure file does not exist yet
     672        appendfile = self.userHome.child('test.txt')
     673        if appendfile.exists():
     674          appendfile.remove()
     675
     676        # Upload file
     677        d = self._upload('STOR test.txt', 'x' * 1000, chainDeferred=d)
     678
     679        def checkContents(ignored=None):
     680          appendfile = self.userHome.child('test.txt').open('r')
     681          contents = appendfile.read()
     682          appendfile.close()
     683          self.assertEquals('x' * 1000, contents)
     684        return d.addCallback(checkContents)
     685
     686
     687
    574688class FTPServerPortDataConnectionTestCase(FTPServerPasvDataConnectionTestCase):
    575689    def setUp(self):
    576690        self.dataPorts = []