[Zope3-checkins] SVN: Zope3/branches/3.2/ Merge some FTP bug fixes
from the trunk
Michael Kerrin
michael.kerrin at openapp.biz
Sat Jan 28 16:28:39 EST 2006
Log message for revision 41481:
Merge some FTP bug fixes from the trunk
------------------------------------------------------------------------
r41480 | mkerrin | 2006-01-28 21:05:12 +0000 (Sat, 28 Jan 2006) | 3 lines
Fix issue #533 in collector and bug in the zope.app.ftp
where an incorrect default value was specified
Changed:
U Zope3/branches/3.2/src/zope/app/ftp/__init__.py
U Zope3/branches/3.2/src/zope/app/ftp/tests/test_ftpview.py
U Zope3/branches/3.2/src/zope/app/twisted/ftp/ftp.py
U Zope3/branches/3.2/src/zope/app/twisted/ftp/test/test_zope_ftp.py
U Zope3/branches/3.2/trial.py
-=-
Modified: Zope3/branches/3.2/src/zope/app/ftp/__init__.py
===================================================================
--- Zope3/branches/3.2/src/zope/app/ftp/__init__.py 2006-01-28 21:05:12 UTC (rev 41480)
+++ Zope3/branches/3.2/src/zope/app/ftp/__init__.py 2006-01-28 21:28:38 UTC (rev 41481)
@@ -237,7 +237,7 @@
f = IReadFile(self._dir[name], None)
if f is not None:
return canAccess(f, 'read')
- d = IReadDirectory(self._dir[name], name)
+ d = IReadDirectory(self._dir[name], None)
if d is not None:
return canAccess(d, 'get')
return False
Modified: Zope3/branches/3.2/src/zope/app/ftp/tests/test_ftpview.py
===================================================================
--- Zope3/branches/3.2/src/zope/app/ftp/tests/test_ftpview.py 2006-01-28 21:05:12 UTC (rev 41480)
+++ Zope3/branches/3.2/src/zope/app/ftp/tests/test_ftpview.py 2006-01-28 21:28:38 UTC (rev 41481)
@@ -70,6 +70,12 @@
return r
return Directory()
+
+# Very simple container that is not adaptable to IReadDirectory
+class NonDirectoryContainer(demofs.Directory):
+ implements(IContainer)
+
+
class File(demofs.File):
implements(IReadFile, IWriteFile, IZopeDublinCore)
@@ -282,6 +288,9 @@
self.assert_(self.__view.readable('f'))
self.assert_(not self.__view.readable('notthere'))
self.assert_(self.__view.readable('test'))
+ root = self.root
+ root['nondir'] = NonDirectoryContainer()
+ self.assertEqual(self.__view.readable('nondir'), False)
def test_read_unicode(self):
root = self.root
Modified: Zope3/branches/3.2/src/zope/app/twisted/ftp/ftp.py
===================================================================
--- Zope3/branches/3.2/src/zope/app/twisted/ftp/ftp.py 2006-01-28 21:05:12 UTC (rev 41480)
+++ Zope3/branches/3.2/src/zope/app/twisted/ftp/ftp.py 2006-01-28 21:28:38 UTC (rev 41481)
@@ -171,8 +171,6 @@
return result['name'].encode('utf-8'), ent
def _stat(self, path, keys):
- if self.fs_access.type(path) == 'd':
- raise ftp.WrongFiletype()
result = self._gotlisting(self.fs_access.lsinfo(path), keys)
return result[1]
@@ -244,22 +242,23 @@
ret |= 0040000
return ret
- def openForReading(self, path):
- p = self._path(path)
+ def _checkFileReadAccess(self, fs_access, p):
+ # run all these methods within the one thread.
+ readable = fs_access.readable(p)
+ if not readable:
+ raise ftp.PermissionDeniedError(p)
- def succeed(result):
- if not result:
- raise ftp.PermissionDeniedError(p)
- return ReadFileObj(self.fs_access, p)
+ filetype = fs_access.type(p)
+ if filetype == 'd':
+ raise ftp.FileNotFoundError(p)
- def failed(failure):
- raise ftp.PermissionDeniedError(p)
+ return ReadFileObj(fs_access, p)
- d = threads.deferToThread(self.fs_access.readable, p)
- d.addCallback(succeed)
- d.addErrback(failed)
+ def openForReading(self, path):
+ p = self._path(path)
- return d
+ return threads.deferToThread(self._checkFileReadAccess,
+ self.fs_access, p)
def openForWriting(self, path):
p = self._path(path)
Modified: Zope3/branches/3.2/src/zope/app/twisted/ftp/test/test_zope_ftp.py
===================================================================
--- Zope3/branches/3.2/src/zope/app/twisted/ftp/test/test_zope_ftp.py 2006-01-28 21:05:12 UTC (rev 41480)
+++ Zope3/branches/3.2/src/zope/app/twisted/ftp/test/test_zope_ftp.py 2006-01-28 21:28:38 UTC (rev 41481)
@@ -109,7 +109,8 @@
responseLines
)
-class BasicFTPServerTestCase(FTPServerTestCase, test_ftp.BasicFTPServerTestCase):
+class BasicFTPServerTestCase(FTPServerTestCase,
+ test_ftp.BasicFTPServerTestCase):
def _authLogin(self):
responseLines = wait(self.client.queueStringCommand('USER root'))
self.assertEquals(
@@ -135,17 +136,32 @@
self._authLogin()
responseLines = wait(self.client.queueStringCommand('RMD /newdir'))
- self.assertEqual(['250 Requested File Action Completed OK'], responseLines)
+ self.assertEqual(
+ ['250 Requested File Action Completed OK'], responseLines)
def test_DELE(self):
self.rootfs.writefile_nocheck('/file.txt', StringIO('x' * 20))
self._authLogin()
responseLines = wait(self.client.queueStringCommand('DELE /file.txt'))
- self.assertEqual(['250 Requested File Action Completed OK'], responseLines)
+ self.assertEqual(
+ ['250 Requested File Action Completed OK'], responseLines)
+ def test_SIZE(self):
+ self.rootfs.writefile_nocheck('/file.txt', StringIO('x' * 20))
+
+ self._anonymousLogin()
+ responseLines = wait(self.client.queueStringCommand('SIZE /file.txt'))
+ self.assertEqual(['213 20'], responseLines)
+
+ def test_SIZE_on_dir(self):
+ self._anonymousLogin()
+ responseLines = wait(self.client.queueStringCommand('SIZE /'))
+ self.assertEqual(['213 0'] , responseLines)
+
+
class FTPServerPasvDataConnectionTestCase(FTPServerTestCase,
- test_ftp.FTPServerPasvDataConnectionTestCase):
+ test_ftp.FTPServerPasvDataConnectionTestCase):
def testLIST(self):
# Login
@@ -203,7 +219,8 @@
# Download a range of different size files
for size in range(100000, 110000, 500):
- self.rootfs.writefile_nocheck('/%d.txt' % (size,), StringIO('x' * size))
+ self.rootfs.writefile_nocheck('/%d.txt' % (size,),
+ StringIO('x' * size))
downloader = self._makeDataConnection()
d = self.client.queueStringCommand('RETR %d.txt' % (size,))
@@ -211,7 +228,8 @@
self.assertEqual('x' * size, downloader.buffer)
-class FTPServerPortDataConnectionTestCaes(FTPServerPasvDataConnectionTestCase, test_ftp.FTPServerPortDataConnectionTestCase):
+class FTPServerPortDataConnectionTestCaes(FTPServerPasvDataConnectionTestCase,
+ test_ftp.FTPServerPortDataConnectionTestCase):
def setUp(self):
FTPServerPasvDataConnectionTestCase.setUp(self)
self.dataPorts = []
@@ -264,7 +282,8 @@
deferred = self.client.queueStringCommand('CWD /nosuchdir')
failureResponseLines = self._waitForCommandFailure(deferred)
self.failUnless(failureResponseLines[-1].startswith('550'),
- "Response didn't start with 550: %r" % failureResponseLines[-1])
+ "Response didn't start with 550: %r" %
+ failureResponseLines[-1])
def testListNonPermission(self):
self._michaelLogin()
Modified: Zope3/branches/3.2/trial.py
===================================================================
--- Zope3/branches/3.2/trial.py 2006-01-28 21:05:12 UTC (rev 41480)
+++ Zope3/branches/3.2/trial.py 2006-01-28 21:28:38 UTC (rev 41481)
@@ -1,4 +1,4 @@
-#!/usr/bin/env python2.3
+#!/usr/bin/env python2.4
##############################################################################
#
# Copyright (c) 2004 Zope Corporation and Contributors.
More information about the Zope3-Checkins
mailing list