1import glob
2import os
3import shutil
4import sys
5import unittest
6
7from test.support.os_helper import (TESTFN, skip_unless_symlink,
8                                    can_symlink, create_empty_file, change_cwd)
9
10
11class GlobTests(unittest.TestCase):
12    dir_fd = None
13
14    def norm(self, *parts):
15        return os.path.normpath(os.path.join(self.tempdir, *parts))
16
17    def joins(self, *tuples):
18        return [os.path.join(self.tempdir, *parts) for parts in tuples]
19
20    def mktemp(self, *parts):
21        filename = self.norm(*parts)
22        base, file = os.path.split(filename)
23        if not os.path.exists(base):
24            os.makedirs(base)
25        create_empty_file(filename)
26
27    def setUp(self):
28        self.tempdir = TESTFN + "_dir"
29        self.mktemp('a', 'D')
30        self.mktemp('aab', 'F')
31        self.mktemp('.aa', 'G')
32        self.mktemp('.bb', 'H')
33        self.mktemp('aaa', 'zzzF')
34        self.mktemp('ZZZ')
35        self.mktemp('EF')
36        self.mktemp('a', 'bcd', 'EF')
37        self.mktemp('a', 'bcd', 'efg', 'ha')
38        if can_symlink():
39            os.symlink(self.norm('broken'), self.norm('sym1'))
40            os.symlink('broken', self.norm('sym2'))
41            os.symlink(os.path.join('a', 'bcd'), self.norm('sym3'))
42        if {os.open, os.stat} <= os.supports_dir_fd and os.scandir in os.supports_fd:
43            self.dir_fd = os.open(self.tempdir, os.O_RDONLY | os.O_DIRECTORY)
44        else:
45            self.dir_fd = None
46
47    def tearDown(self):
48        if self.dir_fd is not None:
49            os.close(self.dir_fd)
50        shutil.rmtree(self.tempdir)
51
52    def glob(self, *parts, **kwargs):
53        if len(parts) == 1:
54            pattern = parts[0]
55        else:
56            pattern = os.path.join(*parts)
57        p = os.path.join(self.tempdir, pattern)
58        res = glob.glob(p, **kwargs)
59        self.assertCountEqual(glob.iglob(p, **kwargs), res)
60        bres = [os.fsencode(x) for x in res]
61        self.assertCountEqual(glob.glob(os.fsencode(p), **kwargs), bres)
62        self.assertCountEqual(glob.iglob(os.fsencode(p), **kwargs), bres)
63
64        with change_cwd(self.tempdir):
65            res2 = glob.glob(pattern, **kwargs)
66            for x in res2:
67                self.assertFalse(os.path.isabs(x), x)
68            if pattern == '**' or pattern == '**' + os.sep:
69                expected = res[1:]
70            else:
71                expected = res
72            self.assertCountEqual([os.path.join(self.tempdir, x) for x in res2],
73                                  expected)
74            self.assertCountEqual(glob.iglob(pattern, **kwargs), res2)
75            bpattern = os.fsencode(pattern)
76            bres2 = [os.fsencode(x) for x in res2]
77            self.assertCountEqual(glob.glob(bpattern, **kwargs), bres2)
78            self.assertCountEqual(glob.iglob(bpattern, **kwargs), bres2)
79
80        self.assertCountEqual(glob.glob(pattern, root_dir=self.tempdir, **kwargs), res2)
81        self.assertCountEqual(glob.iglob(pattern, root_dir=self.tempdir, **kwargs), res2)
82        btempdir = os.fsencode(self.tempdir)
83        self.assertCountEqual(
84            glob.glob(bpattern, root_dir=btempdir, **kwargs), bres2)
85        self.assertCountEqual(
86            glob.iglob(bpattern, root_dir=btempdir, **kwargs), bres2)
87
88        if self.dir_fd is not None:
89            self.assertCountEqual(
90                glob.glob(pattern, dir_fd=self.dir_fd, **kwargs), res2)
91            self.assertCountEqual(
92                glob.iglob(pattern, dir_fd=self.dir_fd, **kwargs), res2)
93            self.assertCountEqual(
94                glob.glob(bpattern, dir_fd=self.dir_fd, **kwargs), bres2)
95            self.assertCountEqual(
96                glob.iglob(bpattern, dir_fd=self.dir_fd, **kwargs), bres2)
97
98        return res
99
100    def assertSequencesEqual_noorder(self, l1, l2):
101        l1 = list(l1)
102        l2 = list(l2)
103        self.assertEqual(set(l1), set(l2))
104        self.assertEqual(sorted(l1), sorted(l2))
105
106    def test_glob_literal(self):
107        eq = self.assertSequencesEqual_noorder
108        eq(self.glob('a'), [self.norm('a')])
109        eq(self.glob('a', 'D'), [self.norm('a', 'D')])
110        eq(self.glob('aab'), [self.norm('aab')])
111        eq(self.glob('zymurgy'), [])
112
113        res = glob.glob('*')
114        self.assertEqual({type(r) for r in res}, {str})
115        res = glob.glob(os.path.join(os.curdir, '*'))
116        self.assertEqual({type(r) for r in res}, {str})
117
118        res = glob.glob(b'*')
119        self.assertEqual({type(r) for r in res}, {bytes})
120        res = glob.glob(os.path.join(os.fsencode(os.curdir), b'*'))
121        self.assertEqual({type(r) for r in res}, {bytes})
122
123    def test_glob_empty_pattern(self):
124        self.assertEqual(glob.glob(''), [])
125        self.assertEqual(glob.glob(b''), [])
126        self.assertEqual(glob.glob('', root_dir=self.tempdir), [])
127        self.assertEqual(glob.glob(b'', root_dir=os.fsencode(self.tempdir)), [])
128        self.assertEqual(glob.glob('', dir_fd=self.dir_fd), [])
129        self.assertEqual(glob.glob(b'', dir_fd=self.dir_fd), [])
130
131    def test_glob_one_directory(self):
132        eq = self.assertSequencesEqual_noorder
133        eq(self.glob('a*'), map(self.norm, ['a', 'aab', 'aaa']))
134        eq(self.glob('*a'), map(self.norm, ['a', 'aaa']))
135        eq(self.glob('.*'), map(self.norm, ['.aa', '.bb']))
136        eq(self.glob('?aa'), map(self.norm, ['aaa']))
137        eq(self.glob('aa?'), map(self.norm, ['aaa', 'aab']))
138        eq(self.glob('aa[ab]'), map(self.norm, ['aaa', 'aab']))
139        eq(self.glob('*q'), [])
140
141    def test_glob_nested_directory(self):
142        eq = self.assertSequencesEqual_noorder
143        if os.path.normcase("abCD") == "abCD":
144            # case-sensitive filesystem
145            eq(self.glob('a', 'bcd', 'E*'), [self.norm('a', 'bcd', 'EF')])
146        else:
147            # case insensitive filesystem
148            eq(self.glob('a', 'bcd', 'E*'), [self.norm('a', 'bcd', 'EF'),
149                                             self.norm('a', 'bcd', 'efg')])
150        eq(self.glob('a', 'bcd', '*g'), [self.norm('a', 'bcd', 'efg')])
151
152    def test_glob_directory_names(self):
153        eq = self.assertSequencesEqual_noorder
154        eq(self.glob('*', 'D'), [self.norm('a', 'D')])
155        eq(self.glob('*', '*a'), [])
156        eq(self.glob('a', '*', '*', '*a'),
157           [self.norm('a', 'bcd', 'efg', 'ha')])
158        eq(self.glob('?a?', '*F'), [self.norm('aaa', 'zzzF'),
159                                    self.norm('aab', 'F')])
160
161    def test_glob_directory_with_trailing_slash(self):
162        # Patterns ending with a slash shouldn't match non-dirs
163        res = glob.glob(self.norm('Z*Z') + os.sep)
164        self.assertEqual(res, [])
165        res = glob.glob(self.norm('ZZZ') + os.sep)
166        self.assertEqual(res, [])
167        # When there is a wildcard pattern which ends with os.sep, glob()
168        # doesn't blow up.
169        res = glob.glob(self.norm('aa*') + os.sep)
170        self.assertEqual(len(res), 2)
171        # either of these results is reasonable
172        self.assertIn(set(res), [
173                      {self.norm('aaa'), self.norm('aab')},
174                      {self.norm('aaa') + os.sep, self.norm('aab') + os.sep},
175                      ])
176
177    def test_glob_bytes_directory_with_trailing_slash(self):
178        # Same as test_glob_directory_with_trailing_slash, but with a
179        # bytes argument.
180        res = glob.glob(os.fsencode(self.norm('Z*Z') + os.sep))
181        self.assertEqual(res, [])
182        res = glob.glob(os.fsencode(self.norm('ZZZ') + os.sep))
183        self.assertEqual(res, [])
184        res = glob.glob(os.fsencode(self.norm('aa*') + os.sep))
185        self.assertEqual(len(res), 2)
186        # either of these results is reasonable
187        self.assertIn(set(res), [
188                      {os.fsencode(self.norm('aaa')),
189                       os.fsencode(self.norm('aab'))},
190                      {os.fsencode(self.norm('aaa') + os.sep),
191                       os.fsencode(self.norm('aab') + os.sep)},
192                      ])
193
194    @skip_unless_symlink
195    def test_glob_symlinks(self):
196        eq = self.assertSequencesEqual_noorder
197        eq(self.glob('sym3'), [self.norm('sym3')])
198        eq(self.glob('sym3', '*'), [self.norm('sym3', 'EF'),
199                                    self.norm('sym3', 'efg')])
200        self.assertIn(self.glob('sym3' + os.sep),
201                      [[self.norm('sym3')], [self.norm('sym3') + os.sep]])
202        eq(self.glob('*', '*F'),
203           [self.norm('aaa', 'zzzF'),
204            self.norm('aab', 'F'), self.norm('sym3', 'EF')])
205
206    @skip_unless_symlink
207    def test_glob_broken_symlinks(self):
208        eq = self.assertSequencesEqual_noorder
209        eq(self.glob('sym*'), [self.norm('sym1'), self.norm('sym2'),
210                               self.norm('sym3')])
211        eq(self.glob('sym1'), [self.norm('sym1')])
212        eq(self.glob('sym2'), [self.norm('sym2')])
213
214    @unittest.skipUnless(sys.platform == "win32", "Win32 specific test")
215    def test_glob_magic_in_drive(self):
216        eq = self.assertSequencesEqual_noorder
217        eq(glob.glob('*:'), [])
218        eq(glob.glob(b'*:'), [])
219        eq(glob.glob('?:'), [])
220        eq(glob.glob(b'?:'), [])
221        eq(glob.glob('\\\\?\\c:\\'), ['\\\\?\\c:\\'])
222        eq(glob.glob(b'\\\\?\\c:\\'), [b'\\\\?\\c:\\'])
223        eq(glob.glob('\\\\*\\*\\'), [])
224        eq(glob.glob(b'\\\\*\\*\\'), [])
225
226    def check_escape(self, arg, expected):
227        self.assertEqual(glob.escape(arg), expected)
228        self.assertEqual(glob.escape(os.fsencode(arg)), os.fsencode(expected))
229
230    def test_escape(self):
231        check = self.check_escape
232        check('abc', 'abc')
233        check('[', '[[]')
234        check('?', '[?]')
235        check('*', '[*]')
236        check('[[_/*?*/_]]', '[[][[]_/[*][?][*]/_]]')
237        check('/[[_/*?*/_]]/', '/[[][[]_/[*][?][*]/_]]/')
238
239    @unittest.skipUnless(sys.platform == "win32", "Win32 specific test")
240    def test_escape_windows(self):
241        check = self.check_escape
242        check('?:?', '?:[?]')
243        check('*:*', '*:[*]')
244        check(r'\\?\c:\?', r'\\?\c:\[?]')
245        check(r'\\*\*\*', r'\\*\*\[*]')
246        check('//?/c:/?', '//?/c:/[?]')
247        check('//*/*/*', '//*/*/[*]')
248
249    def rglob(self, *parts, **kwargs):
250        return self.glob(*parts, recursive=True, **kwargs)
251
252    def test_recursive_glob(self):
253        eq = self.assertSequencesEqual_noorder
254        full = [('EF',), ('ZZZ',),
255                ('a',), ('a', 'D'),
256                ('a', 'bcd'),
257                ('a', 'bcd', 'EF'),
258                ('a', 'bcd', 'efg'),
259                ('a', 'bcd', 'efg', 'ha'),
260                ('aaa',), ('aaa', 'zzzF'),
261                ('aab',), ('aab', 'F'),
262               ]
263        if can_symlink():
264            full += [('sym1',), ('sym2',),
265                     ('sym3',),
266                     ('sym3', 'EF'),
267                     ('sym3', 'efg'),
268                     ('sym3', 'efg', 'ha'),
269                    ]
270        eq(self.rglob('**'), self.joins(('',), *full))
271        eq(self.rglob(os.curdir, '**'),
272            self.joins((os.curdir, ''), *((os.curdir,) + i for i in full)))
273        dirs = [('a', ''), ('a', 'bcd', ''), ('a', 'bcd', 'efg', ''),
274                ('aaa', ''), ('aab', '')]
275        if can_symlink():
276            dirs += [('sym3', ''), ('sym3', 'efg', '')]
277        eq(self.rglob('**', ''), self.joins(('',), *dirs))
278
279        eq(self.rglob('a', '**'), self.joins(
280            ('a', ''), ('a', 'D'), ('a', 'bcd'), ('a', 'bcd', 'EF'),
281            ('a', 'bcd', 'efg'), ('a', 'bcd', 'efg', 'ha')))
282        eq(self.rglob('a**'), self.joins(('a',), ('aaa',), ('aab',)))
283        expect = [('a', 'bcd', 'EF'), ('EF',)]
284        if can_symlink():
285            expect += [('sym3', 'EF')]
286        eq(self.rglob('**', 'EF'), self.joins(*expect))
287        expect = [('a', 'bcd', 'EF'), ('aaa', 'zzzF'), ('aab', 'F'), ('EF',)]
288        if can_symlink():
289            expect += [('sym3', 'EF')]
290        eq(self.rglob('**', '*F'), self.joins(*expect))
291        eq(self.rglob('**', '*F', ''), [])
292        eq(self.rglob('**', 'bcd', '*'), self.joins(
293            ('a', 'bcd', 'EF'), ('a', 'bcd', 'efg')))
294        eq(self.rglob('a', '**', 'bcd'), self.joins(('a', 'bcd')))
295
296        with change_cwd(self.tempdir):
297            join = os.path.join
298            eq(glob.glob('**', recursive=True), [join(*i) for i in full])
299            eq(glob.glob(join('**', ''), recursive=True),
300                [join(*i) for i in dirs])
301            eq(glob.glob(join('**', '*'), recursive=True),
302                [join(*i) for i in full])
303            eq(glob.glob(join(os.curdir, '**'), recursive=True),
304                [join(os.curdir, '')] + [join(os.curdir, *i) for i in full])
305            eq(glob.glob(join(os.curdir, '**', ''), recursive=True),
306                [join(os.curdir, '')] + [join(os.curdir, *i) for i in dirs])
307            eq(glob.glob(join(os.curdir, '**', '*'), recursive=True),
308                [join(os.curdir, *i) for i in full])
309            eq(glob.glob(join('**','zz*F'), recursive=True),
310                [join('aaa', 'zzzF')])
311            eq(glob.glob('**zz*F', recursive=True), [])
312            expect = [join('a', 'bcd', 'EF'), 'EF']
313            if can_symlink():
314                expect += [join('sym3', 'EF')]
315            eq(glob.glob(join('**', 'EF'), recursive=True), expect)
316
317    def test_glob_many_open_files(self):
318        depth = 30
319        base = os.path.join(self.tempdir, 'deep')
320        p = os.path.join(base, *(['d']*depth))
321        os.makedirs(p)
322        pattern = os.path.join(base, *(['*']*depth))
323        iters = [glob.iglob(pattern, recursive=True) for j in range(100)]
324        for it in iters:
325            self.assertEqual(next(it), p)
326        pattern = os.path.join(base, '**', 'd')
327        iters = [glob.iglob(pattern, recursive=True) for j in range(100)]
328        p = base
329        for i in range(depth):
330            p = os.path.join(p, 'd')
331            for it in iters:
332                self.assertEqual(next(it), p)
333
334
335@skip_unless_symlink
336class SymlinkLoopGlobTests(unittest.TestCase):
337
338    def test_selflink(self):
339        tempdir = TESTFN + "_dir"
340        os.makedirs(tempdir)
341        self.addCleanup(shutil.rmtree, tempdir)
342        with change_cwd(tempdir):
343            os.makedirs('dir')
344            create_empty_file(os.path.join('dir', 'file'))
345            os.symlink(os.curdir, os.path.join('dir', 'link'))
346
347            results = glob.glob('**', recursive=True)
348            self.assertEqual(len(results), len(set(results)))
349            results = set(results)
350            depth = 0
351            while results:
352                path = os.path.join(*(['dir'] + ['link'] * depth))
353                self.assertIn(path, results)
354                results.remove(path)
355                if not results:
356                    break
357                path = os.path.join(path, 'file')
358                self.assertIn(path, results)
359                results.remove(path)
360                depth += 1
361
362            results = glob.glob(os.path.join('**', 'file'), recursive=True)
363            self.assertEqual(len(results), len(set(results)))
364            results = set(results)
365            depth = 0
366            while results:
367                path = os.path.join(*(['dir'] + ['link'] * depth + ['file']))
368                self.assertIn(path, results)
369                results.remove(path)
370                depth += 1
371
372            results = glob.glob(os.path.join('**', ''), recursive=True)
373            self.assertEqual(len(results), len(set(results)))
374            results = set(results)
375            depth = 0
376            while results:
377                path = os.path.join(*(['dir'] + ['link'] * depth + ['']))
378                self.assertIn(path, results)
379                results.remove(path)
380                depth += 1
381
382
383if __name__ == "__main__":
384    unittest.main()
385