1import unittest
2from test import support
3from test.support import warnings_helper
4import os
5import sys
6import types
7
8try:
9    import _multiprocessing
10except ModuleNotFoundError:
11    _multiprocessing = None
12
13
14if support.check_sanitizer(address=True, memory=True):
15    # bpo-46633: test___all__ is skipped because importing some modules
16    # directly can trigger known problems with ASAN (like tk or crypt).
17    raise unittest.SkipTest("workaround ASAN build issues on loading tests "
18                            "like tk or crypt")
19
20
21class NoAll(RuntimeError):
22    pass
23
24class FailedImport(RuntimeError):
25    pass
26
27
28class AllTest(unittest.TestCase):
29
30    def setUp(self):
31        # concurrent.futures uses a __getattr__ hook. Its __all__ triggers
32        # import of a submodule, which fails when _multiprocessing is not
33        # available.
34        if _multiprocessing is None:
35            sys.modules["_multiprocessing"] = types.ModuleType("_multiprocessing")
36
37    def tearDown(self):
38        if _multiprocessing is None:
39            sys.modules.pop("_multiprocessing")
40
41    def check_all(self, modname):
42        names = {}
43        with warnings_helper.check_warnings(
44            (f".*{modname}", DeprecationWarning),
45            (".* (module|package)", DeprecationWarning),
46            (".* (module|package)", PendingDeprecationWarning),
47            ("", ResourceWarning),
48            quiet=True):
49            try:
50                exec("import %s" % modname, names)
51            except:
52                # Silent fail here seems the best route since some modules
53                # may not be available or not initialize properly in all
54                # environments.
55                raise FailedImport(modname)
56        if not hasattr(sys.modules[modname], "__all__"):
57            raise NoAll(modname)
58        names = {}
59        with self.subTest(module=modname):
60            with warnings_helper.check_warnings(
61                ("", DeprecationWarning),
62                ("", ResourceWarning),
63                quiet=True):
64                try:
65                    exec("from %s import *" % modname, names)
66                except Exception as e:
67                    # Include the module name in the exception string
68                    self.fail("__all__ failure in {}: {}: {}".format(
69                              modname, e.__class__.__name__, e))
70                if "__builtins__" in names:
71                    del names["__builtins__"]
72                if '__annotations__' in names:
73                    del names['__annotations__']
74                if "__warningregistry__" in names:
75                    del names["__warningregistry__"]
76                keys = set(names)
77                all_list = sys.modules[modname].__all__
78                all_set = set(all_list)
79                self.assertCountEqual(all_set, all_list, "in module {}".format(modname))
80                self.assertEqual(keys, all_set, "in module {}".format(modname))
81
82    def walk_modules(self, basedir, modpath):
83        for fn in sorted(os.listdir(basedir)):
84            path = os.path.join(basedir, fn)
85            if os.path.isdir(path):
86                pkg_init = os.path.join(path, '__init__.py')
87                if os.path.exists(pkg_init):
88                    yield pkg_init, modpath + fn
89                    for p, m in self.walk_modules(path, modpath + fn + "."):
90                        yield p, m
91                continue
92            if not fn.endswith('.py') or fn == '__init__.py':
93                continue
94            yield path, modpath + fn[:-3]
95
96    def test_all(self):
97        # List of denied modules and packages
98        denylist = set([
99            # Will raise a SyntaxError when compiling the exec statement
100            '__future__',
101        ])
102
103        if not sys.platform.startswith('java'):
104            # In case _socket fails to build, make this test fail more gracefully
105            # than an AttributeError somewhere deep in CGIHTTPServer.
106            import _socket
107
108        ignored = []
109        failed_imports = []
110        lib_dir = os.path.dirname(os.path.dirname(__file__))
111        for path, modname in self.walk_modules(lib_dir, ""):
112            m = modname
113            denied = False
114            while m:
115                if m in denylist:
116                    denied = True
117                    break
118                m = m.rpartition('.')[0]
119            if denied:
120                continue
121            if support.verbose:
122                print(modname)
123            try:
124                # This heuristic speeds up the process by removing, de facto,
125                # most test modules (and avoiding the auto-executing ones).
126                with open(path, "rb") as f:
127                    if b"__all__" not in f.read():
128                        raise NoAll(modname)
129                    self.check_all(modname)
130            except NoAll:
131                ignored.append(modname)
132            except FailedImport:
133                failed_imports.append(modname)
134
135        if support.verbose:
136            print('Following modules have no __all__ and have been ignored:',
137                  ignored)
138            print('Following modules failed to be imported:', failed_imports)
139
140
141if __name__ == "__main__":
142    unittest.main()
143