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