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

File test-ftp-server-rest-appe.patch, 5.7 KB (added by eternicode, 4 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 = []