1from test.support import run_unittest, unload, check_warnings, CleanImport
2import unittest
3import sys
4import importlib
5from importlib.util import spec_from_file_location
6import pkgutil
7import os
8import os.path
9import tempfile
10import shutil
11import zipfile
12
13# Note: pkgutil.walk_packages is currently tested in test_runpy. This is
14# a hack to get a major issue resolved for 3.3b2. Longer term, it should
15# be moved back here, perhaps by factoring out the helper code for
16# creating interesting package layouts to a separate module.
17# Issue #15348 declares this is indeed a dodgy hack ;)
18
19class PkgutilTests(unittest.TestCase):
20
21    def setUp(self):
22        self.dirname = tempfile.mkdtemp()
23        self.addCleanup(shutil.rmtree, self.dirname)
24        sys.path.insert(0, self.dirname)
25
26    def tearDown(self):
27        del sys.path[0]
28
29    def test_getdata_filesys(self):
30        pkg = 'test_getdata_filesys'
31
32        # Include a LF and a CRLF, to test that binary data is read back
33        RESOURCE_DATA = b'Hello, world!\nSecond line\r\nThird line'
34
35        # Make a package with some resources
36        package_dir = os.path.join(self.dirname, pkg)
37        os.mkdir(package_dir)
38        # Empty init.py
39        f = open(os.path.join(package_dir, '__init__.py'), "wb")
40        f.close()
41        # Resource files, res.txt, sub/res.txt
42        f = open(os.path.join(package_dir, 'res.txt'), "wb")
43        f.write(RESOURCE_DATA)
44        f.close()
45        os.mkdir(os.path.join(package_dir, 'sub'))
46        f = open(os.path.join(package_dir, 'sub', 'res.txt'), "wb")
47        f.write(RESOURCE_DATA)
48        f.close()
49
50        # Check we can read the resources
51        res1 = pkgutil.get_data(pkg, 'res.txt')
52        self.assertEqual(res1, RESOURCE_DATA)
53        res2 = pkgutil.get_data(pkg, 'sub/res.txt')
54        self.assertEqual(res2, RESOURCE_DATA)
55
56        del sys.modules[pkg]
57
58    def test_getdata_zipfile(self):
59        zip = 'test_getdata_zipfile.zip'
60        pkg = 'test_getdata_zipfile'
61
62        # Include a LF and a CRLF, to test that binary data is read back
63        RESOURCE_DATA = b'Hello, world!\nSecond line\r\nThird line'
64
65        # Make a package with some resources
66        zip_file = os.path.join(self.dirname, zip)
67        z = zipfile.ZipFile(zip_file, 'w')
68
69        # Empty init.py
70        z.writestr(pkg + '/__init__.py', "")
71        # Resource files, res.txt, sub/res.txt
72        z.writestr(pkg + '/res.txt', RESOURCE_DATA)
73        z.writestr(pkg + '/sub/res.txt', RESOURCE_DATA)
74        z.close()
75
76        # Check we can read the resources
77        sys.path.insert(0, zip_file)
78        res1 = pkgutil.get_data(pkg, 'res.txt')
79        self.assertEqual(res1, RESOURCE_DATA)
80        res2 = pkgutil.get_data(pkg, 'sub/res.txt')
81        self.assertEqual(res2, RESOURCE_DATA)
82
83        names = []
84        for moduleinfo in pkgutil.iter_modules([zip_file]):
85            self.assertIsInstance(moduleinfo, pkgutil.ModuleInfo)
86            names.append(moduleinfo.name)
87        self.assertEqual(names, ['test_getdata_zipfile'])
88
89        del sys.path[0]
90
91        del sys.modules[pkg]
92
93    def test_unreadable_dir_on_syspath(self):
94        # issue7367 - walk_packages failed if unreadable dir on sys.path
95        package_name = "unreadable_package"
96        d = os.path.join(self.dirname, package_name)
97        # this does not appear to create an unreadable dir on Windows
98        #   but the test should not fail anyway
99        os.mkdir(d, 0)
100        self.addCleanup(os.rmdir, d)
101        for t in pkgutil.walk_packages(path=[self.dirname]):
102            self.fail("unexpected package found")
103
104    def test_walkpackages_filesys(self):
105        pkg1 = 'test_walkpackages_filesys'
106        pkg1_dir = os.path.join(self.dirname, pkg1)
107        os.mkdir(pkg1_dir)
108        f = open(os.path.join(pkg1_dir, '__init__.py'), "wb")
109        f.close()
110        os.mkdir(os.path.join(pkg1_dir, 'sub'))
111        f = open(os.path.join(pkg1_dir, 'sub', '__init__.py'), "wb")
112        f.close()
113        f = open(os.path.join(pkg1_dir, 'sub', 'mod.py'), "wb")
114        f.close()
115
116        # Now, to juice it up, let's add the opposite packages, too.
117        pkg2 = 'sub'
118        pkg2_dir = os.path.join(self.dirname, pkg2)
119        os.mkdir(pkg2_dir)
120        f = open(os.path.join(pkg2_dir, '__init__.py'), "wb")
121        f.close()
122        os.mkdir(os.path.join(pkg2_dir, 'test_walkpackages_filesys'))
123        f = open(os.path.join(pkg2_dir, 'test_walkpackages_filesys', '__init__.py'), "wb")
124        f.close()
125        f = open(os.path.join(pkg2_dir, 'test_walkpackages_filesys', 'mod.py'), "wb")
126        f.close()
127
128        expected = [
129            'sub',
130            'sub.test_walkpackages_filesys',
131            'sub.test_walkpackages_filesys.mod',
132            'test_walkpackages_filesys',
133            'test_walkpackages_filesys.sub',
134            'test_walkpackages_filesys.sub.mod',
135        ]
136        actual= [e[1] for e in pkgutil.walk_packages([self.dirname])]
137        self.assertEqual(actual, expected)
138
139        for pkg in expected:
140            if pkg.endswith('mod'):
141                continue
142            del sys.modules[pkg]
143
144    def test_walkpackages_zipfile(self):
145        """Tests the same as test_walkpackages_filesys, only with a zip file."""
146
147        zip = 'test_walkpackages_zipfile.zip'
148        pkg1 = 'test_walkpackages_zipfile'
149        pkg2 = 'sub'
150
151        zip_file = os.path.join(self.dirname, zip)
152        z = zipfile.ZipFile(zip_file, 'w')
153        z.writestr(pkg2 + '/__init__.py', "")
154        z.writestr(pkg2 + '/' + pkg1 + '/__init__.py', "")
155        z.writestr(pkg2 + '/' + pkg1 + '/mod.py', "")
156        z.writestr(pkg1 + '/__init__.py', "")
157        z.writestr(pkg1 + '/' + pkg2 + '/__init__.py', "")
158        z.writestr(pkg1 + '/' + pkg2 + '/mod.py', "")
159        z.close()
160
161        sys.path.insert(0, zip_file)
162        expected = [
163            'sub',
164            'sub.test_walkpackages_zipfile',
165            'sub.test_walkpackages_zipfile.mod',
166            'test_walkpackages_zipfile',
167            'test_walkpackages_zipfile.sub',
168            'test_walkpackages_zipfile.sub.mod',
169        ]
170        actual= [e[1] for e in pkgutil.walk_packages([zip_file])]
171        self.assertEqual(actual, expected)
172        del sys.path[0]
173
174        for pkg in expected:
175            if pkg.endswith('mod'):
176                continue
177            del sys.modules[pkg]
178
179
180
181class PkgutilPEP302Tests(unittest.TestCase):
182
183    class MyTestLoader(object):
184        def create_module(self, spec):
185            return None
186
187        def exec_module(self, mod):
188            # Count how many times the module is reloaded
189            mod.__dict__['loads'] = mod.__dict__.get('loads', 0) + 1
190
191        def get_data(self, path):
192            return "Hello, world!"
193
194    class MyTestImporter(object):
195        def find_spec(self, fullname, path=None, target=None):
196            loader = PkgutilPEP302Tests.MyTestLoader()
197            return spec_from_file_location(fullname,
198                                           '<%s>' % loader.__class__.__name__,
199                                           loader=loader,
200                                           submodule_search_locations=[])
201
202    def setUp(self):
203        sys.meta_path.insert(0, self.MyTestImporter())
204
205    def tearDown(self):
206        del sys.meta_path[0]
207
208    def test_getdata_pep302(self):
209        # Use a dummy finder/loader
210        self.assertEqual(pkgutil.get_data('foo', 'dummy'), "Hello, world!")
211        del sys.modules['foo']
212
213    def test_alreadyloaded(self):
214        # Ensure that get_data works without reloading - the "loads" module
215        # variable in the example loader should count how many times a reload
216        # occurs.
217        import foo
218        self.assertEqual(foo.loads, 1)
219        self.assertEqual(pkgutil.get_data('foo', 'dummy'), "Hello, world!")
220        self.assertEqual(foo.loads, 1)
221        del sys.modules['foo']
222
223
224# These tests, especially the setup and cleanup, are hideous. They
225# need to be cleaned up once issue 14715 is addressed.
226class ExtendPathTests(unittest.TestCase):
227    def create_init(self, pkgname):
228        dirname = tempfile.mkdtemp()
229        sys.path.insert(0, dirname)
230
231        pkgdir = os.path.join(dirname, pkgname)
232        os.mkdir(pkgdir)
233        with open(os.path.join(pkgdir, '__init__.py'), 'w') as fl:
234            fl.write('from pkgutil import extend_path\n__path__ = extend_path(__path__, __name__)\n')
235
236        return dirname
237
238    def create_submodule(self, dirname, pkgname, submodule_name, value):
239        module_name = os.path.join(dirname, pkgname, submodule_name + '.py')
240        with open(module_name, 'w') as fl:
241            print('value={}'.format(value), file=fl)
242
243    def test_simple(self):
244        pkgname = 'foo'
245        dirname_0 = self.create_init(pkgname)
246        dirname_1 = self.create_init(pkgname)
247        self.create_submodule(dirname_0, pkgname, 'bar', 0)
248        self.create_submodule(dirname_1, pkgname, 'baz', 1)
249        import foo.bar
250        import foo.baz
251        # Ensure we read the expected values
252        self.assertEqual(foo.bar.value, 0)
253        self.assertEqual(foo.baz.value, 1)
254
255        # Ensure the path is set up correctly
256        self.assertEqual(sorted(foo.__path__),
257                         sorted([os.path.join(dirname_0, pkgname),
258                                 os.path.join(dirname_1, pkgname)]))
259
260        # Cleanup
261        shutil.rmtree(dirname_0)
262        shutil.rmtree(dirname_1)
263        del sys.path[0]
264        del sys.path[0]
265        del sys.modules['foo']
266        del sys.modules['foo.bar']
267        del sys.modules['foo.baz']
268
269
270    # Another awful testing hack to be cleaned up once the test_runpy
271    # helpers are factored out to a common location
272    def test_iter_importers(self):
273        iter_importers = pkgutil.iter_importers
274        get_importer = pkgutil.get_importer
275
276        pkgname = 'spam'
277        modname = 'eggs'
278        dirname = self.create_init(pkgname)
279        pathitem = os.path.join(dirname, pkgname)
280        fullname = '{}.{}'.format(pkgname, modname)
281        sys.modules.pop(fullname, None)
282        sys.modules.pop(pkgname, None)
283        try:
284            self.create_submodule(dirname, pkgname, modname, 0)
285
286            importlib.import_module(fullname)
287
288            importers = list(iter_importers(fullname))
289            expected_importer = get_importer(pathitem)
290            for finder in importers:
291                spec = pkgutil._get_spec(finder, fullname)
292                loader = spec.loader
293                try:
294                    loader = loader.loader
295                except AttributeError:
296                    # For now we still allow raw loaders from
297                    # find_module().
298                    pass
299                self.assertIsInstance(finder, importlib.machinery.FileFinder)
300                self.assertEqual(finder, expected_importer)
301                self.assertIsInstance(loader,
302                                      importlib.machinery.SourceFileLoader)
303                self.assertIsNone(pkgutil._get_spec(finder, pkgname))
304
305            with self.assertRaises(ImportError):
306                list(iter_importers('invalid.module'))
307
308            with self.assertRaises(ImportError):
309                list(iter_importers('.spam'))
310        finally:
311            shutil.rmtree(dirname)
312            del sys.path[0]
313            try:
314                del sys.modules['spam']
315                del sys.modules['spam.eggs']
316            except KeyError:
317                pass
318
319
320    def test_mixed_namespace(self):
321        pkgname = 'foo'
322        dirname_0 = self.create_init(pkgname)
323        dirname_1 = self.create_init(pkgname)
324        self.create_submodule(dirname_0, pkgname, 'bar', 0)
325        # Turn this into a PEP 420 namespace package
326        os.unlink(os.path.join(dirname_0, pkgname, '__init__.py'))
327        self.create_submodule(dirname_1, pkgname, 'baz', 1)
328        import foo.bar
329        import foo.baz
330        # Ensure we read the expected values
331        self.assertEqual(foo.bar.value, 0)
332        self.assertEqual(foo.baz.value, 1)
333
334        # Ensure the path is set up correctly
335        self.assertEqual(sorted(foo.__path__),
336                         sorted([os.path.join(dirname_0, pkgname),
337                                 os.path.join(dirname_1, pkgname)]))
338
339        # Cleanup
340        shutil.rmtree(dirname_0)
341        shutil.rmtree(dirname_1)
342        del sys.path[0]
343        del sys.path[0]
344        del sys.modules['foo']
345        del sys.modules['foo.bar']
346        del sys.modules['foo.baz']
347
348    # XXX: test .pkg files
349
350
351class NestedNamespacePackageTest(unittest.TestCase):
352
353    def setUp(self):
354        self.basedir = tempfile.mkdtemp()
355        self.old_path = sys.path[:]
356
357    def tearDown(self):
358        sys.path[:] = self.old_path
359        shutil.rmtree(self.basedir)
360
361    def create_module(self, name, contents):
362        base, final = name.rsplit('.', 1)
363        base_path = os.path.join(self.basedir, base.replace('.', os.path.sep))
364        os.makedirs(base_path, exist_ok=True)
365        with open(os.path.join(base_path, final + ".py"), 'w') as f:
366            f.write(contents)
367
368    def test_nested(self):
369        pkgutil_boilerplate = (
370            'import pkgutil; '
371            '__path__ = pkgutil.extend_path(__path__, __name__)')
372        self.create_module('a.pkg.__init__', pkgutil_boilerplate)
373        self.create_module('b.pkg.__init__', pkgutil_boilerplate)
374        self.create_module('a.pkg.subpkg.__init__', pkgutil_boilerplate)
375        self.create_module('b.pkg.subpkg.__init__', pkgutil_boilerplate)
376        self.create_module('a.pkg.subpkg.c', 'c = 1')
377        self.create_module('b.pkg.subpkg.d', 'd = 2')
378        sys.path.insert(0, os.path.join(self.basedir, 'a'))
379        sys.path.insert(0, os.path.join(self.basedir, 'b'))
380        import pkg
381        self.addCleanup(unload, 'pkg')
382        self.assertEqual(len(pkg.__path__), 2)
383        import pkg.subpkg
384        self.addCleanup(unload, 'pkg.subpkg')
385        self.assertEqual(len(pkg.subpkg.__path__), 2)
386        from pkg.subpkg.c import c
387        from pkg.subpkg.d import d
388        self.assertEqual(c, 1)
389        self.assertEqual(d, 2)
390
391
392class ImportlibMigrationTests(unittest.TestCase):
393    # With full PEP 302 support in the standard import machinery, the
394    # PEP 302 emulation in this module is in the process of being
395    # deprecated in favour of importlib proper
396
397    def check_deprecated(self):
398        return check_warnings(
399            ("This emulation is deprecated, use 'importlib' instead",
400             DeprecationWarning))
401
402    def test_importer_deprecated(self):
403        with self.check_deprecated():
404            pkgutil.ImpImporter("")
405
406    def test_loader_deprecated(self):
407        with self.check_deprecated():
408            pkgutil.ImpLoader("", "", "", "")
409
410    def test_get_loader_avoids_emulation(self):
411        with check_warnings() as w:
412            self.assertIsNotNone(pkgutil.get_loader("sys"))
413            self.assertIsNotNone(pkgutil.get_loader("os"))
414            self.assertIsNotNone(pkgutil.get_loader("test.support"))
415            self.assertEqual(len(w.warnings), 0)
416
417    @unittest.skipIf(__name__ == '__main__', 'not compatible with __main__')
418    def test_get_loader_handles_missing_loader_attribute(self):
419        global __loader__
420        this_loader = __loader__
421        del __loader__
422        try:
423            with check_warnings() as w:
424                self.assertIsNotNone(pkgutil.get_loader(__name__))
425                self.assertEqual(len(w.warnings), 0)
426        finally:
427            __loader__ = this_loader
428
429    def test_get_loader_handles_missing_spec_attribute(self):
430        name = 'spam'
431        mod = type(sys)(name)
432        del mod.__spec__
433        with CleanImport(name):
434            sys.modules[name] = mod
435            loader = pkgutil.get_loader(name)
436        self.assertIsNone(loader)
437
438    def test_get_loader_handles_spec_attribute_none(self):
439        name = 'spam'
440        mod = type(sys)(name)
441        mod.__spec__ = None
442        with CleanImport(name):
443            sys.modules[name] = mod
444            loader = pkgutil.get_loader(name)
445        self.assertIsNone(loader)
446
447    def test_get_loader_None_in_sys_modules(self):
448        name = 'totally bogus'
449        sys.modules[name] = None
450        try:
451            loader = pkgutil.get_loader(name)
452        finally:
453            del sys.modules[name]
454        self.assertIsNone(loader)
455
456    def test_find_loader_missing_module(self):
457        name = 'totally bogus'
458        loader = pkgutil.find_loader(name)
459        self.assertIsNone(loader)
460
461    def test_find_loader_avoids_emulation(self):
462        with check_warnings() as w:
463            self.assertIsNotNone(pkgutil.find_loader("sys"))
464            self.assertIsNotNone(pkgutil.find_loader("os"))
465            self.assertIsNotNone(pkgutil.find_loader("test.support"))
466            self.assertEqual(len(w.warnings), 0)
467
468    def test_get_importer_avoids_emulation(self):
469        # We use an illegal path so *none* of the path hooks should fire
470        with check_warnings() as w:
471            self.assertIsNone(pkgutil.get_importer("*??"))
472            self.assertEqual(len(w.warnings), 0)
473
474    def test_iter_importers_avoids_emulation(self):
475        with check_warnings() as w:
476            for importer in pkgutil.iter_importers(): pass
477            self.assertEqual(len(w.warnings), 0)
478
479
480def test_main():
481    run_unittest(PkgutilTests, PkgutilPEP302Tests, ExtendPathTests,
482                 NestedNamespacePackageTest, ImportlibMigrationTests)
483    # this is necessary if test is run repeated (like when finding leaks)
484    import zipimport
485    import importlib
486    zipimport._zip_directory_cache.clear()
487    importlib.invalidate_caches()
488
489
490if __name__ == '__main__':
491    test_main()
492