Parameterize test_match_glob
diff --git a/sdks/python/apache_beam/io/localfilesystem_test.py b/sdks/python/apache_beam/io/localfilesystem_test.py
index 3bbbd1e..aa6d47b 100644
--- a/sdks/python/apache_beam/io/localfilesystem_test.py
+++ b/sdks/python/apache_beam/io/localfilesystem_test.py
@@ -28,6 +28,8 @@
import unittest
import mock
+from parameterized import parameterized
+from parameterized import param
from apache_beam.io import localfilesystem
from apache_beam.io.filesystem import BeamIOError
@@ -150,35 +152,40 @@
self.fs.match([None])
self.assertEqual(list(error.exception.exception_details.keys()), [None])
- def test_match_glob(self):
- path1 = os.path.join(self.tmpdir, 'f1')
- path2 = os.path.join(self.tmpdir, 'f2')
- open(path1, 'a').close()
- open(path2, 'a').close()
+ @parameterized.expand([
+ param('*',
+ files=['a', 'b', 'c/x'],
+ expected=['a', 'b']),
+ param('**',
+ files=['a', 'b/x', 'c/x'],
+ expected=['a', 'b/x', 'c/x']),
+ param('*/*',
+ files=['a', 'b/x', 'c/x', 'd/x/y'],
+ expected=['b/x', 'c/x']),
+ param('**/*',
+ files=['a', 'b/x', 'c/x', 'd/x/y'],
+ expected=['b/x', 'c/x', 'd/x/y']),
+ ])
+ def test_match_glob(self, pattern, files, expected):
+ for filename in files:
+ full_path = os.path.join(self.tmpdir, filename)
+ dirname = os.path.dirname(full_path)
+ if not dirname == full_path:
+ # Make sure we don't go outside the tmpdir
+ assert os.path.commonprefix([self.tmpdir, full_path]) == self.tmpdir
+ try:
+ self.fs.mkdirs(dirname)
+ except IOError:
+ # Directory exists
+ pass
+
+ open(full_path, 'a').close() # create empty file
# Match both the files in the directory
- path = os.path.join(self.tmpdir, '*')
- result = self.fs.match([path])[0]
- files = [f.path for f in result.metadata_list]
- self.assertItemsEqual(files, [path1, path2])
-
- def test_match_glob_wildcard_directory_name(self):
- base = os.path.join(self.tmpdir, "glob-with-with-suffix")
- dir1 = os.path.join(base, "a")
- dir2 = os.path.join(base, "b")
- for dir in [dir1, dir2]:
- self.fs.mkdirs(dir)
-
- path1 = os.path.join(dir1, "file.txt")
- path2 = os.path.join(dir2, "file.txt")
- for path in [path1, path2]:
- open(path, 'w').close()
-
- pattern = "{}/*/file.txt".format(base)
-
- result = self.fs.match([pattern])[0]
- files = [f.path for f in result.metadata_list]
- self.assertItemsEqual(files, [path1, path2])
+ full_pattern = os.path.join(self.tmpdir, pattern)
+ result = self.fs.match([full_pattern])[0]
+ files = [os.path.relpath(f.path, self.tmpdir) for f in result.metadata_list]
+ self.assertItemsEqual(files, expected)
def test_match_directory(self):
result = self.fs.match([self.tmpdir])[0]
