Ticket #4181: 4181-nlst-globbing-2.diff

twisted/test/test_ftp.py

@@ -825 +825 @@
                 ["425 Can't open data connection."])
         return d.addCallback(gotPortNum)
 
+    def test_NLSTGlobbing(self):
+        """
+        When Unix shell globbing is used with NLST only files matching
+        the pattern will be returned.
+        """
+        self.dirPath.child('test.txt').touch()
+        self.dirPath.child('ceva.txt').touch()
+        self.dirPath.child('no.match').touch()
+        d = self._anonymousLogin()
 
+        self._download('NLST *.txt', chainDeferred=d)
 
+        def checkDownload(download):
+            filenames = download[:-2].split('\r\n')
+            filenames.sort()
+            self.assertEqual(['ceva.txt', 'test.txt'], filenames)
+
+        return d.addCallback(checkDownload)
+
+
 class DTPFactoryTests(unittest.TestCase):
     """
     Tests for L{ftp.DTPFactory}.
@@ -2330 +2348 @@
             self.assertRaises(ftp.InvalidPath, ftp.toSegments, ['x'], inp)
 
 
+
+class IsGlobbingExpressionTests(unittest.TestCase):
+    """
+    Tests for _isGlobbingExpression utility function.
+    """
+
+    def test_isGlobbingExpressionEmptySegments(self):
+        """
+        _isGlobbingExpression will return False for None, or empty
+        segments.
+        """
+        self.assertFalse(ftp._isGlobbingExpression())
+        self.assertFalse(ftp._isGlobbingExpression([]))
+        self.assertFalse(ftp._isGlobbingExpression(None))
+
+
+    def test_isGlobbingExpressionNoGlob(self):
+        """
+        _isGlobbingExpression will return False for plain segments.
+
+        Also, it only checks the last segment part (filename) and will not
+        check the path name.
+        """
+        self.assertFalse(ftp._isGlobbingExpression(['ignore', 'expr']))
+        self.assertFalse(ftp._isGlobbingExpression(['*.txt', 'expr']))
+
+
+    def test_isGlobbingExpressionGlob(self):
+        """
+        _isGlobbingExpression will return True for segments which contains
+        globbing characters in the last segment part (filename).
+        """
+        self.assertTrue(ftp._isGlobbingExpression(['ignore', '*.txt']))
+        self.assertTrue(ftp._isGlobbingExpression(['ignore', '[a-b].txt']))
+        self.assertTrue(ftp._isGlobbingExpression(['ignore', 'fil?.txt']))
+
+
+
 class BaseFTPRealmTests(unittest.TestCase):
     """
     Tests for L{ftp.BaseFTPRealm}, a base class to help define L{IFTPShell}

twisted/protocols/ftp.py

@@ -224 +224 @@
         return defer.fail()
 
 
+def _isGlobbingExpression(segments=None):
+    """
+    Helper for checking if a FTPShell `segments` contains a wildcard Unix
+    expression.
 
+    Only filename globbing is supported.
+    This means that wildcards can only be presents in the last element of
+    `segments`.
+
+    @type  segments: C{list}
+    @param segments: List of path elements as used by the FTP server protocol.
+
+    @rtype: Boolean
+    @return: True if `segments` contains a globbing expression.
+    """
+    if not segments:
+        return False
+
+    # To check that something is a glob expression, we convert it to
+    # Regular Expression. If the result is the same as the original expression
+    # then it contains no globbing expression.
+    globCandidate = segments[-1]
+    # A set of default regex rules is added to all strings.
+    emtpyTranslations = fnmatch.translate('')
+    globTranslations = fnmatch.translate(globCandidate)
+
+    if globCandidate + emtpyTranslations == globTranslations:
+        return False
+    else:
+        return True
+
+
 class FTPCmdError(Exception):
     """
     Generic exception for FTP commands.
@@ -962 +993 @@
         except InvalidPath:
             return defer.fail(FileNotFoundError(path))
 
-        def cbList(results):
+        def cbList(results, glob=None):
             """
             Send, line by line, each file in the directory listing, and then
             close the connection.
 
+            If `glob` is not None, the result will be filtered using
+            Unix shell-style wildcards
+            (http://docs.python.org/2/library/fnmatch.html).
+
             @type results: A C{list} of C{tuple}. The first element of each
                 C{tuple} is a C{str} and the second element is a C{list}.
             @param results: The names of the files in the directory.
@@ -977 +1012 @@
             """
             self.reply(DATA_CNX_ALREADY_OPEN_START_XFR)
             for (name, ignored) in results:
-                self.dtpInstance.sendLine(name)
-            self.dtpInstance.transport.loseConnection()
-            return (TXFR_COMPLETE_OK,)
-
-        def cbGlob(results):
-            self.reply(DATA_CNX_ALREADY_OPEN_START_XFR)
-            for (name, ignored) in results:
-                if fnmatch.fnmatch(name, segments[-1]):
+                if not glob or (glob and fnmatch.fnmatch(name, glob)):
                     self.dtpInstance.sendLine(name)
             self.dtpInstance.transport.loseConnection()
             return (TXFR_COMPLETE_OK,)
@@ -1006 +1034 @@
             self.dtpInstance.transport.loseConnection()
             return (TXFR_COMPLETE_OK,)
 
-        # XXX This globbing may be incomplete: see #4181
-        if segments and (
-            '*' in segments[-1] or '?' in segments[-1] or
-            ('[' in segments[-1] and ']' in segments[-1])):
-            d = self.shell.list(segments[:-1])
-            d.addCallback(cbGlob)
+        if _isGlobbingExpression(segments):
+            # Remove globbing expression from path
+            # and keep to be used for filtering.
+            glob = segments.pop()
         else:
-            d = self.shell.list(segments)
-            d.addCallback(cbList)
-            # self.shell.list will generate an error if the path is invalid
-            d.addErrback(listErr)
+            glob = None
+
+        d = self.shell.list(segments)
+        d.addCallback(cbList, glob)
+        # self.shell.list will generate an error if the path is invalid
+        d.addErrback(listErr)
         return d