|
| 1 | +import unittest |
| 2 | +from test import support |
| 3 | +from test.support import warnings_helper |
| 4 | +import os |
| 5 | +import sys |
| 6 | +import types |
| 7 | + |
| 8 | +try: |
| 9 | + import _multiprocessing |
| 10 | +except ModuleNotFoundError: |
| 11 | + _multiprocessing = None |
| 12 | + |
| 13 | + |
| 14 | +if 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 | + |
| 21 | +class NoAll(RuntimeError): |
| 22 | + pass |
| 23 | + |
| 24 | +class FailedImport(RuntimeError): |
| 25 | + pass |
| 26 | + |
| 27 | + |
| 28 | +class 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 | + # TODO: RUSTPYTHON |
| 97 | + @unittest.expectedFailure |
| 98 | + def test_all(self): |
| 99 | + # List of denied modules and packages |
| 100 | + denylist = set([ |
| 101 | + # Will raise a SyntaxError when compiling the exec statement |
| 102 | + '__future__', |
| 103 | + ]) |
| 104 | + |
| 105 | + # In case _socket fails to build, make this test fail more gracefully |
| 106 | + # than an AttributeError somewhere deep in CGIHTTPServer. |
| 107 | + import _socket |
| 108 | + |
| 109 | + ignored = [] |
| 110 | + failed_imports = [] |
| 111 | + lib_dir = os.path.dirname(os.path.dirname(__file__)) |
| 112 | + for path, modname in self.walk_modules(lib_dir, ""): |
| 113 | + m = modname |
| 114 | + denied = False |
| 115 | + while m: |
| 116 | + if m in denylist: |
| 117 | + denied = True |
| 118 | + break |
| 119 | + m = m.rpartition('.')[0] |
| 120 | + if denied: |
| 121 | + continue |
| 122 | + if support.verbose: |
| 123 | + print(modname) |
| 124 | + try: |
| 125 | + # This heuristic speeds up the process by removing, de facto, |
| 126 | + # most test modules (and avoiding the auto-executing ones). |
| 127 | + with open(path, "rb") as f: |
| 128 | + if b"__all__" not in f.read(): |
| 129 | + raise NoAll(modname) |
| 130 | + self.check_all(modname) |
| 131 | + except NoAll: |
| 132 | + ignored.append(modname) |
| 133 | + except FailedImport: |
| 134 | + failed_imports.append(modname) |
| 135 | + |
| 136 | + if support.verbose: |
| 137 | + print('Following modules have no __all__ and have been ignored:', |
| 138 | + ignored) |
| 139 | + print('Following modules failed to be imported:', failed_imports) |
| 140 | + |
| 141 | + |
| 142 | +if __name__ == "__main__": |
| 143 | + unittest.main() |
0 commit comments