mirror of
https://github.com/python/cpython
synced 2026-11-14 15:27:20 +03:00
ae86660de3
gh-161229: Accept `None` for `globals` and `locals` in `__lazy_import__()` (GH-161231)
(cherry picked from commit 1941319efb)
Co-authored-by: Lazizbek Ergashev <lazerg2@gmail.com>
2924 lines
108 KiB
Python
2924 lines
108 KiB
Python
"""Tests for PEP 810 lazy imports."""
|
||
|
||
import io
|
||
import dis
|
||
import subprocess
|
||
import sys
|
||
import textwrap
|
||
import threading
|
||
import types
|
||
import unittest
|
||
import tempfile
|
||
import os
|
||
import contextlib
|
||
|
||
from test import support
|
||
from test.support.script_helper import assert_python_ok, assert_python_failure
|
||
|
||
try:
|
||
import _testcapi
|
||
except ImportError:
|
||
_testcapi = None
|
||
|
||
|
||
class LazyImportTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.lazy_imports_filter = sys.get_lazy_imports_filter()
|
||
self.lazy_imports = sys.get_lazy_imports()
|
||
|
||
def tearDown(self):
|
||
"""Clean up any test modules from sys.modules."""
|
||
for key in list(sys.modules.keys()):
|
||
if key.startswith('test.test_lazy_import.data'):
|
||
del sys.modules[key]
|
||
|
||
sys.set_lazy_imports_filter(self.lazy_imports_filter)
|
||
sys.set_lazy_imports(self.lazy_imports)
|
||
sys.lazy_modules.clear()
|
||
|
||
|
||
class LazyImportTests(LazyImportTestCase):
|
||
"""Tests for basic lazy import functionality."""
|
||
|
||
def test_basic_unused(self):
|
||
"""Lazy imported module should not be loaded if never accessed."""
|
||
import test.test_lazy_import.data.basic_unused
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.lazy_modules)
|
||
|
||
def test_sys_lazy_modules(self):
|
||
try:
|
||
import test.test_lazy_import.data.basic_from_unused
|
||
except ImportError as e:
|
||
self.fail('lazy import failed')
|
||
|
||
self.assertFalse("test.test_lazy_import.data.basic2" in sys.modules)
|
||
# The package is already loaded, so it is not a pending import.
|
||
self.assertNotIn("test.test_lazy_import.data", sys.lazy_modules)
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.lazy_modules)
|
||
test.test_lazy_import.data.basic_from_unused.basic2
|
||
self.assertNotIn("test.test_import.data", sys.lazy_modules)
|
||
|
||
def test_basic_unused_use_externally(self):
|
||
"""Lazy import should load module when accessed from outside."""
|
||
from test.test_lazy_import.data import basic_unused
|
||
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
x = basic_unused.test.test_lazy_import.data.basic2
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_basic_from_unused_use_externally(self):
|
||
"""Lazy 'from' import should load when accessed from outside."""
|
||
from test.test_lazy_import.data import basic_from_unused
|
||
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
x = basic_from_unused.basic2
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_basic_unused_dir(self):
|
||
"""dir() on module should not trigger lazy import reification."""
|
||
import test.test_lazy_import.data.basic_unused
|
||
|
||
x = dir(test.test_lazy_import.data.basic_unused)
|
||
self.assertIn("test", x)
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_basic_dir(self):
|
||
"""dir() at module scope should not trigger lazy import reification."""
|
||
from test.test_lazy_import.data import basic_dir
|
||
|
||
self.assertIn("test", basic_dir.x)
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_basic_used(self):
|
||
"""Lazy import should load when accessed within the module."""
|
||
import test.test_lazy_import.data.basic_used
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
@support.requires_subprocess()
|
||
def test_from_import_with_module_getattr(self):
|
||
"""Lazy from import should respect module-level __getattr__."""
|
||
code = textwrap.dedent("""
|
||
lazy from test.test_lazy_import.data.module_with_getattr import dynamic_attr
|
||
assert dynamic_attr == "from_getattr"
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_from_import_with_module_getattr_raising(self):
|
||
"""Lazy from import should respect module-level __getattr__."""
|
||
code = textwrap.dedent("""
|
||
lazy from test.test_lazy_import.data.module_with_getattr import raising_attr
|
||
|
||
try:
|
||
raising_attr
|
||
except ValueError as exc:
|
||
assert str(exc) == 'from_getattr', exc
|
||
else:
|
||
assert False, f'ValueError is not raised: {raising_attr}'
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_from_import_with_module_getattr_missing(self):
|
||
"""Lazy from import should respect module-level __getattr__."""
|
||
for attr in ("missing_attr", "import_error_attr"):
|
||
with self.subTest(attr=attr):
|
||
code = textwrap.dedent(f"""
|
||
lazy from test.test_lazy_import.data.module_with_getattr import {attr}
|
||
|
||
try:
|
||
{attr}
|
||
except ImportError as exc:
|
||
assert '{attr}' in str(exc), exc
|
||
assert exc.__cause__ is not None
|
||
else:
|
||
assert False, ('ImportError is not raised', {attr})
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_from_import_with_module_getattr_warning(self):
|
||
"""Lazy from import should respect module-level __getattr__."""
|
||
code = textwrap.dedent("""
|
||
import warnings
|
||
|
||
with warnings.catch_warnings(record=True) as log:
|
||
lazy from test.test_lazy_import.data.module_with_getattr import warning_attr
|
||
|
||
assert log == []
|
||
|
||
with warnings.catch_warnings(record=True) as log:
|
||
warning_attr
|
||
assert warning_attr == 'from_warning_attr', warning_attr
|
||
assert len(log) == 1, log
|
||
assert isinstance(log[0].message, UserWarning), log
|
||
assert str(log[0].message) == 'from_getattr', log
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_from_import_with_imported_module_getattr(self):
|
||
"""Lazy from import should not shadow an imported module's __getattr__."""
|
||
code = textwrap.dedent("""
|
||
import test.test_lazy_import.data.module_with_getattr as mod
|
||
lazy from test.test_lazy_import.data.module_with_getattr import dynamic_attr
|
||
assert dynamic_attr == "from_getattr"
|
||
assert mod.dynamic_attr == "from_getattr"
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_module_getattr_does_not_shadow_own_lazy_import(self):
|
||
"""Module __getattr__ should not shadow the module's own lazy imports."""
|
||
code = textwrap.dedent("""
|
||
import types
|
||
import test.test_lazy_import.data.module_with_lazy_import_and_getattr as mod
|
||
assert mod.basic2.__name__ == "test.test_lazy_import.data.basic2"
|
||
assert mod.f is mod.basic2.f
|
||
assert not isinstance(vars(mod)["basic2"], types.LazyImportType)
|
||
assert mod.missing == "from_getattr:missing"
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
class GlobalLazyImportModeTests(LazyImportTestCase):
|
||
"""Tests for sys.set_lazy_imports() global mode control."""
|
||
|
||
def tearDown(self):
|
||
for key in list(sys.modules.keys()):
|
||
if key.startswith('test.test_lazy_import.data'):
|
||
del sys.modules[key]
|
||
|
||
sys.set_lazy_imports_filter(None)
|
||
sys.set_lazy_imports("normal")
|
||
|
||
def test_global_off_rejected(self):
|
||
"""Mode 'none' is not supported."""
|
||
with self.assertRaises(ValueError):
|
||
sys.set_lazy_imports("none")
|
||
|
||
def test_global_on(self):
|
||
"""Mode 'all' should make regular imports lazy."""
|
||
import test.test_lazy_import.data.global_on
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_global_filter(self):
|
||
"""Filter returning False should prevent lazy loading."""
|
||
import test.test_lazy_import.data.global_filter
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_global_filter_true(self):
|
||
"""Filter returning True should allow lazy loading."""
|
||
import test.test_lazy_import.data.global_filter_true
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_global_filter_from(self):
|
||
"""Filter should work with 'from' imports."""
|
||
import test.test_lazy_import.data.global_filter_from
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_global_filter_from_true(self):
|
||
"""Filter returning True should allow lazy 'from' imports."""
|
||
import test.test_lazy_import.data.global_filter_from_true
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
|
||
class CompatibilityModeTests(LazyImportTestCase):
|
||
"""Tests for __lazy_modules__ compatibility mode."""
|
||
|
||
def test_compatibility_mode(self):
|
||
"""__lazy_modules__ should enable lazy imports for listed modules."""
|
||
import test.test_lazy_import.data.basic_compatibility_mode
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_compatibility_mode_used(self):
|
||
"""Using a lazy import from __lazy_modules__ should load the module."""
|
||
import test.test_lazy_import.data.basic_compatibility_mode_used
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_compatibility_mode_func(self):
|
||
"""Imports inside functions should be eager even in compatibility mode."""
|
||
import test.test_lazy_import.data.compatibility_mode_func
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_compatibility_mode_try_except(self):
|
||
"""Imports in try/except should be eager even in compatibility mode."""
|
||
import test.test_lazy_import.data.compatibility_mode_try_except
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_compatibility_mode_relative(self):
|
||
"""__lazy_modules__ should work with relative imports."""
|
||
import test.test_lazy_import.data.basic_compatibility_mode_relative
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
|
||
class ModuleIntrospectionTests(LazyImportTestCase):
|
||
"""Tests for module dict and getattr behavior with lazy imports."""
|
||
|
||
def test_modules_dict(self):
|
||
"""Accessing module.__dict__ should not trigger reification."""
|
||
import test.test_lazy_import.data.modules_dict
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_modules_getattr(self):
|
||
"""Module __getattr__ for lazy import name should trigger reification."""
|
||
import test.test_lazy_import.data.modules_getattr
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_modules_getattr_other(self):
|
||
"""Module __getattr__ for other names should not trigger reification."""
|
||
import test.test_lazy_import.data.modules_getattr_other
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
|
||
class LazyImportTypeTests(LazyImportTestCase):
|
||
"""Tests for the LazyImportType and its resolve() method."""
|
||
|
||
def test_lazy_value_resolve(self):
|
||
"""resolve() method should force the lazy import to load."""
|
||
import test.test_lazy_import.data.lazy_get_value
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_lazy_import_type_exposed(self):
|
||
"""LazyImportType should be exposed in types module."""
|
||
self.assertHasAttr(types, 'LazyImportType')
|
||
self.assertEqual(types.LazyImportType.__name__, 'lazy_import')
|
||
|
||
def test_lazy_import_type_cant_construct(self):
|
||
"""LazyImportType should not be directly constructible."""
|
||
self.assertRaises(TypeError, types.LazyImportType, {}, "module")
|
||
|
||
@support.requires_subprocess()
|
||
def test_lazy_import_type_attributes_accessible(self):
|
||
"""Check that static PyLazyImport_Type is initialized at startup."""
|
||
code = textwrap.dedent("""
|
||
lazy import json
|
||
print(globals()["json"].resolve)
|
||
""")
|
||
proc = assert_python_ok("-c", code)
|
||
self.assertIn(b"<built-in method resolve of lazy_import object at", proc.out)
|
||
|
||
@support.requires_subprocess()
|
||
def test_lazy_import_type_attribute_error_message(self):
|
||
"""Check that LazyImportType attribute error message is helpful."""
|
||
code = textwrap.dedent("""
|
||
lazy import asyncio
|
||
try:
|
||
globals()["asyncio"].Task
|
||
except AttributeError as exc:
|
||
assert str(exc) == (
|
||
"cannot access attribute 'Task' "
|
||
"on unresolved lazy import 'asyncio'"
|
||
), repr(str(exc))
|
||
else:
|
||
assert False, 'AttributeError is not raised'
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.run_with_limited_c_stack(depth=10_000, size=256 * 1024)
|
||
def test_deeply_dotted_name(self):
|
||
"""A long dotted name must not overflow the C stack."""
|
||
m = types.SimpleNamespace()
|
||
# Every lookup in m.m.m... succeeds, so resolve() only fails if the
|
||
# C stack runs out.
|
||
m.m = m
|
||
ns = {"__builtins__": {"__lazy_import__": __lazy_import__,
|
||
"__import__": lambda *args: m}}
|
||
# Free the chain on the main thread: a thread with a small C stack
|
||
# postpones deep deallocations and never gets back to them.
|
||
self.addCleanup(ns.clear)
|
||
exec("lazy import " + ".".join(["m"] * 10_000) + " as x", ns)
|
||
with contextlib.suppress(RecursionError):
|
||
ns["x"].resolve()
|
||
|
||
|
||
class SyntaxRestrictionTests(LazyImportTestCase):
|
||
"""Tests for syntax restrictions on lazy imports."""
|
||
|
||
def test_lazy_try_except(self):
|
||
"""lazy import inside try/except should raise SyntaxError."""
|
||
with self.assertRaises(SyntaxError):
|
||
import test.test_lazy_import.data.badsyntax.lazy_try_except
|
||
|
||
def test_lazy_try_except_from(self):
|
||
"""lazy from import inside try/except should raise SyntaxError."""
|
||
with self.assertRaises(SyntaxError):
|
||
import test.test_lazy_import.data.badsyntax.lazy_try_except_from
|
||
|
||
def test_lazy_try_except_from_star(self):
|
||
"""lazy from import * should raise SyntaxError."""
|
||
with self.assertRaises(SyntaxError):
|
||
import test.test_lazy_import.data.badsyntax.lazy_try_except_from_star
|
||
|
||
def test_lazy_future_import(self):
|
||
"""lazy from __future__ import should raise SyntaxError."""
|
||
with self.assertRaises(SyntaxError) as cm:
|
||
import test.test_lazy_import.data.badsyntax.lazy_future_import
|
||
# Check we highlight 'lazy' (column offset 0, end offset 4)
|
||
self.assertEqual(cm.exception.offset, 1)
|
||
self.assertEqual(cm.exception.end_offset, 5)
|
||
|
||
def test_lazy_import_func(self):
|
||
"""lazy import inside function should raise SyntaxError."""
|
||
with self.assertRaises(SyntaxError):
|
||
import test.test_lazy_import.data.badsyntax.lazy_import_func
|
||
|
||
def test_lazy_import_exec_in_function(self):
|
||
"""lazy import via exec() inside a function should raise SyntaxError."""
|
||
# exec() inside a function creates a non-module-level context
|
||
# where lazy imports are not allowed
|
||
def f():
|
||
exec("lazy import json")
|
||
|
||
with self.assertRaises(SyntaxError) as cm:
|
||
f()
|
||
self.assertIn("only allowed at module level", str(cm.exception))
|
||
|
||
def test_lazy_import_exec_in_class(self):
|
||
"""lazy import via exec() inside a class should raise SyntaxError."""
|
||
# exec() inside a class body also has non-module-level locals.
|
||
with self.assertRaises(SyntaxError) as cm:
|
||
class C:
|
||
exec("lazy import json")
|
||
|
||
self.assertIn("only allowed at module level", str(cm.exception))
|
||
|
||
@support.requires_subprocess()
|
||
def test_lazy_import_exec_at_module_level(self):
|
||
"""lazy import via exec() at module level should work."""
|
||
# exec() at module level (globals == locals) should allow lazy imports
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
exec("lazy import json")
|
||
# Should be lazy - not loaded yet
|
||
assert 'json' not in sys.modules
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
class EagerImportInLazyModeTests(LazyImportTestCase):
|
||
"""Tests for imports that should remain eager even in lazy mode."""
|
||
|
||
def test_try_except_eager(self):
|
||
"""Imports in try/except should be eager even with mode='all'."""
|
||
sys.set_lazy_imports("all")
|
||
import test.test_lazy_import.data.try_except_eager
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_try_except_eager_from(self):
|
||
"""From imports in try/except should be eager even with mode='all'."""
|
||
sys.set_lazy_imports("all")
|
||
import test.test_lazy_import.data.try_except_eager_from
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_eager_import_func(self):
|
||
"""Imports inside functions should return modules, not proxies."""
|
||
sys.set_lazy_imports("all")
|
||
import test.test_lazy_import.data.eager_import_func
|
||
|
||
f = test.test_lazy_import.data.eager_import_func.f
|
||
self.assertEqual(type(f()), type(sys))
|
||
|
||
def test_exec_import_func(self):
|
||
"""Implicit lazy imports via exec() inside functions should be eager."""
|
||
sys.set_lazy_imports("all")
|
||
|
||
def f():
|
||
exec("import test.test_lazy_import.data.basic2")
|
||
|
||
f()
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_exec_import_func_with_lazy_modules(self):
|
||
"""__lazy_modules__ should not make exec() imports lazy inside functions."""
|
||
globals()["__lazy_modules__"] = ["test.test_lazy_import.data.basic2"]
|
||
try:
|
||
def f():
|
||
exec("import test.test_lazy_import.data.basic2")
|
||
|
||
f()
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
finally:
|
||
del globals()["__lazy_modules__"]
|
||
|
||
def test_exec_import_class(self):
|
||
"""Implicit lazy imports via exec() inside classes should be eager."""
|
||
sys.set_lazy_imports("all")
|
||
|
||
class C:
|
||
exec("import test.test_lazy_import.data.basic2")
|
||
|
||
self.assertIsNotNone(C)
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_exec_import_class_with_lazy_modules(self):
|
||
"""__lazy_modules__ should not make exec() imports lazy inside classes."""
|
||
globals()["__lazy_modules__"] = ["test.test_lazy_import.data.basic2"]
|
||
try:
|
||
class C:
|
||
exec("import test.test_lazy_import.data.basic2")
|
||
|
||
self.assertIsNotNone(C)
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
finally:
|
||
del globals()["__lazy_modules__"]
|
||
|
||
|
||
class WithStatementTests(LazyImportTestCase):
|
||
"""Tests for lazy imports in with statement context."""
|
||
|
||
def test_lazy_with(self):
|
||
"""lazy import with 'with' statement should work."""
|
||
import test.test_lazy_import.data.lazy_with
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_lazy_with_from(self):
|
||
"""lazy from import with 'with' statement should work."""
|
||
import test.test_lazy_import.data.lazy_with_from
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class IndependentSubmoduleTests(LazyImportTestCase):
|
||
def check(self, code):
|
||
assert_python_ok('-c', textwrap.dedent(code))
|
||
|
||
def test_siblings(self):
|
||
for package, children in (
|
||
('test.test_lazy_import.data.pkg', ('b', 'bar')),
|
||
('test.test_lazy_import.data.metasyntactic.foo', ('ack', 'bar')),
|
||
):
|
||
for first, second in (children, children[::-1]):
|
||
with self.subTest(package=package, first=first):
|
||
self.check(f"""
|
||
import sys
|
||
lazy import {package}.{first}
|
||
lazy import {package}.{second}
|
||
assert {package}.{first} is sys.modules['{package}.{first}']
|
||
assert '{package}.{second}' not in sys.modules
|
||
assert {package}.{second} is sys.modules['{package}.{second}']
|
||
""")
|
||
|
||
def test_star_import(self):
|
||
self.check("""
|
||
lazy import urllib.nonexistent
|
||
lazy import urllib
|
||
assert urllib.__name__ == 'urllib'
|
||
from urllib import *
|
||
""")
|
||
|
||
def test_deleted_child(self):
|
||
self.check("""
|
||
lazy from xml import dom
|
||
assert dom.__name__ == 'xml.dom'
|
||
import xml
|
||
del xml.dom
|
||
assert not hasattr(xml, 'dom')
|
||
lazy from xml import dom
|
||
assert hasattr(xml, 'dom')
|
||
""")
|
||
|
||
def test_cached_import_releases_builtins(self):
|
||
for statement in ('lazy from xml.dom import Node',
|
||
'lazy import xml.dom.minidom'):
|
||
with self.subTest(statement=statement):
|
||
self.check(f"""
|
||
import builtins, gc, weakref
|
||
import xml.dom.minidom
|
||
class Payload:
|
||
pass
|
||
payload = Payload()
|
||
reference = weakref.ref(payload)
|
||
namespace = {{'__builtins__': dict(vars(builtins), payload=payload)}}
|
||
exec({statement!r}, namespace)
|
||
exec('Node' if 'from' in {statement!r} else 'xml.dom.minidom', namespace)
|
||
del namespace, payload
|
||
gc.collect()
|
||
assert reference() is None
|
||
""")
|
||
|
||
def test_invalid_descendant_retries(self):
|
||
self.check("""
|
||
lazy import test.test_lazy_import.data.pkg.b.foo
|
||
for _ in range(2):
|
||
try:
|
||
test.test_lazy_import.data.pkg.b.foo
|
||
except ModuleNotFoundError:
|
||
pass
|
||
else:
|
||
raise AssertionError('imported an attribute as a module')
|
||
import test.test_lazy_import.data.pkg.b
|
||
assert callable(test.test_lazy_import.data.pkg.b.foo)
|
||
""")
|
||
|
||
def test_namespace_import_hooks(self):
|
||
self.check("""
|
||
import builtins
|
||
first = {'__builtins__': vars(builtins).copy()}
|
||
second = {'__builtins__': vars(builtins).copy()}
|
||
exec('lazy import xml.dom', first)
|
||
exec('lazy import xml.dom', second)
|
||
def denied(*args):
|
||
raise AssertionError('used another namespace import hook')
|
||
second['__builtins__']['__import__'] = denied
|
||
exec('assert xml.dom.__name__ == "xml.dom"', first)
|
||
""")
|
||
|
||
def test_retry_after_recursive_access(self):
|
||
self.check("""
|
||
import sys, xml
|
||
from importlib.machinery import ModuleSpec
|
||
class Loader:
|
||
calls = 0
|
||
def find_spec(self, name, path=None, target=None):
|
||
if name == 'xml.broken':
|
||
return ModuleSpec(name, self)
|
||
def create_module(self, spec):
|
||
return None
|
||
def exec_module(self, module):
|
||
self.calls += 1
|
||
xml.broken
|
||
raise RuntimeError('failed initialization')
|
||
loader = Loader()
|
||
sys.meta_path.insert(0, loader)
|
||
lazy import xml.broken
|
||
for _ in range(2):
|
||
try:
|
||
xml.broken
|
||
except RuntimeError as exc:
|
||
assert str(exc) == 'failed initialization'
|
||
else:
|
||
raise AssertionError('returned a failed partial module')
|
||
assert loader.calls == 2
|
||
""")
|
||
|
||
|
||
class PackageTests(LazyImportTestCase):
|
||
"""Tests for lazy imports with packages."""
|
||
|
||
def test_lazy_import_pkg(self):
|
||
"""lazy import of package submodule should load the package."""
|
||
out = io.StringIO()
|
||
|
||
with contextlib.redirect_stdout(out):
|
||
import test.test_lazy_import.data.lazy_import_pkg
|
||
|
||
self.assertIn("test.test_lazy_import.data.pkg", sys.modules)
|
||
self.assertIn("test.test_lazy_import.data.pkg.bar", sys.modules)
|
||
self.assertIn("BAR_MODULE_LOADED", out.getvalue())
|
||
|
||
def test_lazy_submodule_stored_in_parent_dict(self):
|
||
"""Accessing a lazy submodule should store it in the parent's __dict__."""
|
||
out = io.StringIO()
|
||
|
||
with contextlib.redirect_stdout(out):
|
||
import test.test_lazy_import.data.lazy_import_pkg
|
||
|
||
pkg = sys.modules["test.test_lazy_import.data.pkg"]
|
||
self.assertIn("bar", pkg.__dict__)
|
||
self.assertIs(pkg.__dict__["bar"], sys.modules["test.test_lazy_import.data.pkg.bar"])
|
||
self.assertIn("BAR_MODULE_LOADED", out.getvalue())
|
||
|
||
def test_lazy_import_pkg_cross_import(self):
|
||
"""Cross-imports within package should preserve lazy imports."""
|
||
import test.test_lazy_import.data.pkg.c
|
||
|
||
self.assertIn("test.test_lazy_import.data.pkg", sys.modules)
|
||
self.assertIn("test.test_lazy_import.data.pkg.c", sys.modules)
|
||
self.assertNotIn("test.test_lazy_import.data.pkg.b", sys.modules)
|
||
|
||
g = test.test_lazy_import.data.pkg.c.get_globals()
|
||
self.assertEqual(type(g["x"]), int)
|
||
self.assertEqual(type(g["b"]), types.LazyImportType)
|
||
|
||
@support.requires_subprocess()
|
||
def test_lazy_from_import_does_not_pollute_parent(self):
|
||
"""Lazy from import should not add the name to the parent module's dict."""
|
||
code = textwrap.dedent("""
|
||
lazy from json import nonexistent_attr
|
||
import json
|
||
assert "nonexistent_attr" not in json.__dict__, (
|
||
"lazy from import should not publish attributes on the parent module"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_package_from_import_with_module_getattr_raising(self):
|
||
"""Lazy from import should respect a package's __getattr__."""
|
||
code = textwrap.dedent("""
|
||
lazy from test.test_lazy_import.data.pkg import raising_attr
|
||
|
||
try:
|
||
raising_attr
|
||
except ValueError as exc:
|
||
assert str(exc) == 'from_getattr', exc
|
||
else:
|
||
assert False, f'ValueError is not raised: {raising_attr}'
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_package_from_import_with_module_getattr_missing(self):
|
||
"""Lazy from import should respect package's __getattr__."""
|
||
for attr in ("missing_attr", "import_error_attr"):
|
||
with self.subTest(attr=attr):
|
||
code = textwrap.dedent(f"""
|
||
lazy from test.test_lazy_import.data.pkg import {attr}
|
||
|
||
try:
|
||
{attr}
|
||
except ImportError as exc:
|
||
assert '{attr}' in str(exc), exc
|
||
assert exc.__cause__ is not None
|
||
else:
|
||
assert False, ('ImportError is not raised', {attr})
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_from_import_with_module_getattr_warning(self):
|
||
"""Lazy from import should respect package's __getattr__."""
|
||
code = textwrap.dedent("""
|
||
import warnings
|
||
|
||
with warnings.catch_warnings(record=True) as log:
|
||
lazy from test.test_lazy_import.data.pkg import warning_attr
|
||
|
||
assert log == []
|
||
|
||
with warnings.catch_warnings(record=True) as log:
|
||
warning_attr
|
||
assert warning_attr == 'from_warning_attr', warning_attr
|
||
assert len(log) == 1, log
|
||
assert isinstance(log[0].message, UserWarning), log
|
||
assert str(log[0].message) == 'from_getattr', log
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.requires_subprocess()
|
||
def test_package_from_import_with_module_getattr(self):
|
||
"""Lazy from import should respect a package's __getattr__."""
|
||
code = textwrap.dedent("""
|
||
import test.test_lazy_import.data.pkg as pkg
|
||
lazy from test.test_lazy_import.data.pkg import dynamic_attr
|
||
assert dynamic_attr == "from_getattr"
|
||
assert pkg.dynamic_attr == "from_getattr"
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
class DunderLazyImportTests(LazyImportTestCase):
|
||
"""Tests for __lazy_import__ builtin function."""
|
||
|
||
def test_dunder_lazy_import(self):
|
||
"""__lazy_import__ should create lazy import proxy."""
|
||
import test.test_lazy_import.data.dunder_lazy_import
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_dunder_lazy_import_with_custom_filter(self):
|
||
sys.set_lazy_imports_filter(lambda importer, imported, fromlist: False)
|
||
import test.test_lazy_import.data.dunder_lazy_import
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_dunder_lazy_import_used(self):
|
||
"""Using __lazy_import__ result should trigger module load."""
|
||
import test.test_lazy_import.data.dunder_lazy_import_used
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
@support.requires_subprocess()
|
||
def test_dunder_lazy_import_fromlist_resolves_to_module(self):
|
||
for fromlist in ["basic2", ("basic2",)]:
|
||
with self.subTest(fromlist=fromlist):
|
||
code = textwrap.dedent(f"""
|
||
import sys
|
||
import types
|
||
|
||
lazy = __lazy_import__("test.test_lazy_import.data", fromlist={fromlist!r})
|
||
|
||
def check():
|
||
lazy_obj = globals()["lazy"]
|
||
assert type(lazy_obj) is types.LazyImportType, lazy_obj
|
||
assert "test.test_lazy_import.data.basic2" not in sys.modules
|
||
|
||
resolved = lazy_obj.resolve()
|
||
assert type(resolved) is types.ModuleType, resolved
|
||
assert "test.test_lazy_import.data.basic2" in sys.modules
|
||
assert resolved.__name__ == "test.test_lazy_import.data"
|
||
assert resolved.basic2.x == 42
|
||
|
||
check()
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_dunder_lazy_import_invalid_arguments(self):
|
||
"""__lazy_import__ should reject invalid arguments."""
|
||
for invalid_name in (b"", 123, None):
|
||
with self.assertRaises(TypeError):
|
||
__lazy_import__(invalid_name)
|
||
|
||
with self.assertRaises(ValueError):
|
||
__lazy_import__("sys", level=-1)
|
||
with self.assertRaises(TypeError):
|
||
__lazy_import__("sys", globals=1)
|
||
|
||
code = textwrap.dedent("""
|
||
__lazy_import__("sys", fromlist=(1, 2, 3))
|
||
""")
|
||
result = assert_python_failure("-c", code, NO_COLOR='y')
|
||
self.assertIn(
|
||
b"TypeError: Item in ``from list'' must be str, not int",
|
||
result.err)
|
||
|
||
@support.requires_subprocess()
|
||
def test_dunder_lazy_import_explicit_none(self):
|
||
"""__lazy_import__ should accept None for globals and locals."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
lazy = __lazy_import__("test.test_lazy_import.data.basic2",
|
||
globals=None, locals=None)
|
||
assert type(globals()["lazy"]) is types.LazyImportType
|
||
assert "test.test_lazy_import.data.basic2" not in sys.modules
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_dunder_lazy_import_builtins(self):
|
||
"""__lazy_import__ should use module's __builtins__ for __import__."""
|
||
from test.test_lazy_import.data import dunder_lazy_import_builtins
|
||
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
self.assertEqual(dunder_lazy_import_builtins.basic.basic2, 42)
|
||
|
||
|
||
class SysLazyImportsAPITests(LazyImportTestCase):
|
||
"""Tests for sys lazy imports API functions."""
|
||
|
||
def test_set_lazy_imports_requires_string(self):
|
||
"""set_lazy_imports should reject non-string arguments."""
|
||
with self.assertRaises(TypeError):
|
||
sys.set_lazy_imports(True)
|
||
with self.assertRaises(TypeError):
|
||
sys.set_lazy_imports(None)
|
||
with self.assertRaises(TypeError):
|
||
sys.set_lazy_imports(1)
|
||
|
||
def test_set_lazy_imports_rejects_invalid_mode(self):
|
||
"""set_lazy_imports should reject invalid mode strings."""
|
||
with self.assertRaises(ValueError):
|
||
sys.set_lazy_imports("invalid")
|
||
with self.assertRaises(ValueError):
|
||
sys.set_lazy_imports("on")
|
||
with self.assertRaises(ValueError):
|
||
sys.set_lazy_imports("off")
|
||
|
||
def test_get_lazy_imports_returns_string(self):
|
||
"""get_lazy_imports should return string modes."""
|
||
sys.set_lazy_imports("normal")
|
||
self.assertEqual(sys.get_lazy_imports(), "normal")
|
||
|
||
sys.set_lazy_imports("all")
|
||
self.assertEqual(sys.get_lazy_imports(), "all")
|
||
|
||
def test_get_lazy_imports_filter_default(self):
|
||
"""get_lazy_imports_filter should return None by default."""
|
||
sys.set_lazy_imports_filter(None)
|
||
self.assertIsNone(sys.get_lazy_imports_filter())
|
||
|
||
def test_set_and_get_lazy_imports_filter(self):
|
||
"""set/get_lazy_imports_filter should round-trip filter function."""
|
||
def my_filter(name):
|
||
return name.startswith("test.")
|
||
|
||
sys.set_lazy_imports_filter(my_filter)
|
||
self.assertIs(sys.get_lazy_imports_filter(), my_filter)
|
||
|
||
def test_lazy_modules_attribute_is_dict(self):
|
||
"""sys.lazy_modules should be a set per PEP 810."""
|
||
self.assertIsInstance(sys.lazy_modules, set)
|
||
|
||
@support.requires_subprocess()
|
||
def test_lazy_modules_tracks_lazy_imports(self):
|
||
"""sys.lazy_modules should track lazily imported module names."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
initial_count = len(sys.lazy_modules)
|
||
import test.test_lazy_import.data.basic_unused
|
||
assert "test.test_lazy_import.data.basic2" in sys.lazy_modules
|
||
assert len(sys.lazy_modules) > initial_count
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class ErrorHandlingTests(LazyImportTestCase):
|
||
"""Tests for error handling during lazy import reification."""
|
||
|
||
def test_missing_lazy_submodule_raises_module_not_found_error(self):
|
||
"""Accessing a nonexistent lazy submodule via parent attr raises ModuleNotFoundError."""
|
||
code = textwrap.dedent("""
|
||
lazy import test.test_lazy_import.data.nonexistent_module
|
||
|
||
try:
|
||
_ = test.test_lazy_import.data.nonexistent_module
|
||
except ModuleNotFoundError:
|
||
pass
|
||
else:
|
||
raise AssertionError("ModuleNotFoundError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_non_package_lazily_imported(self):
|
||
"""Accessing a nonexistent lazy name via parent attr raises ModuleNotFoundError."""
|
||
code = textwrap.dedent("""
|
||
lazy import math.pi
|
||
|
||
try:
|
||
_ = math.pi
|
||
except ModuleNotFoundError:
|
||
pass
|
||
else:
|
||
raise AssertionError("ModuleNotFoundError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_non_package_lazily_imported_as(self):
|
||
"""A dotted lazy import as raises when the name is not a module."""
|
||
# gh-157757: the eager statement raises, so the lazy one raises too.
|
||
code = textwrap.dedent("""
|
||
lazy import math.pi as pi
|
||
|
||
try:
|
||
pi
|
||
except ModuleNotFoundError:
|
||
pass
|
||
else:
|
||
raise AssertionError("ModuleNotFoundError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_missing_attribute_raises_import_error(self):
|
||
"""Accessing a nonexistent lazy name via from import raises ImportError."""
|
||
code = textwrap.dedent("""
|
||
lazy from sys import doesnotexist
|
||
|
||
try:
|
||
_ = doesnotexist
|
||
except ImportError:
|
||
pass
|
||
else:
|
||
raise AssertionError("ImportError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_missing_lazy_from_import_shows_chained_traceback(self):
|
||
"""Accessing missing attribute from lazy from-import should chain errors."""
|
||
code = textwrap.dedent("""
|
||
lazy from test.test_lazy_import.data.basic2 import nonexistent_name
|
||
|
||
try:
|
||
_ = nonexistent_name
|
||
except ImportError as e:
|
||
assert e.__cause__ is not None, "Expected chained exception"
|
||
else:
|
||
raise AssertionError("ImportError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
@support.subTests('name', (
|
||
'test.test_lazy_import.data.broken_module_chained_cause',
|
||
'test.test_lazy_import.data.broken_module_chained_context',
|
||
'test.test_lazy_import.data.broken_module_chained_suppressed',
|
||
))
|
||
def test_chained_exception_import_shows_notes(self, name):
|
||
"""Accessing missing attribute from lazy from-import should chain errors."""
|
||
code = textwrap.dedent(f"""
|
||
lazy import {name}
|
||
|
||
try:
|
||
_ = {name}
|
||
except ValueError as e:
|
||
assert any(
|
||
note.startswith("lazy import of '{name}' declared in ")
|
||
for note in e.__notes__
|
||
), e.__notes__
|
||
else:
|
||
raise AssertionError("ImportError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_reification_retries_on_failure(self):
|
||
"""Failed reification should allow retry on subsequent access.
|
||
|
||
PEP 810: "If reification fails, the lazy object is not reified or replaced.
|
||
Subsequent uses of the lazy object will re-try the reification."
|
||
"""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
lazy import test.test_lazy_import.data.broken_module
|
||
|
||
try:
|
||
_ = test.test_lazy_import.data.broken_module
|
||
except ValueError as exc:
|
||
assert str(exc) == "This module always fails to import", exc
|
||
else:
|
||
raise AssertionError("ValueError was not raised")
|
||
|
||
assert "test.test_lazy_import.data.broken_module" not in sys.modules
|
||
|
||
try:
|
||
_ = test.test_lazy_import.data.broken_module
|
||
except ValueError as exc:
|
||
assert str(exc) == "This module always fails to import", exc
|
||
else:
|
||
raise AssertionError("ValueError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_lazy_submodule_traceback_hides_importlib_frames(self):
|
||
code = textwrap.dedent("""
|
||
import traceback
|
||
|
||
lazy import test.test_lazy_import.data.broken_module
|
||
|
||
try:
|
||
_ = test.test_lazy_import.data.broken_module
|
||
except ValueError as exc:
|
||
frames = traceback.extract_tb(exc.__traceback__)
|
||
assert [frame.name for frame in frames] == ["<module>", "<module>"], frames
|
||
assert frames[0].filename == "<string>", frames
|
||
assert frames[1].filename.endswith("broken_module.py"), frames
|
||
else:
|
||
raise AssertionError("ValueError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_module_not_found_during_module_execution_propagates(self):
|
||
code = textwrap.dedent("""
|
||
lazy import test.test_lazy_import.data.missing_dependency
|
||
|
||
try:
|
||
_ = test.test_lazy_import.data.missing_dependency
|
||
except ModuleNotFoundError as exc:
|
||
assert exc.name == "missing_dependency_for_lazy_import_test", exc.name
|
||
assert str(exc) == "No module named 'missing_dependency_for_lazy_import_test'", exc
|
||
else:
|
||
raise AssertionError("ModuleNotFoundError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_self_named_module_not_found_during_module_execution_propagates(self):
|
||
code = textwrap.dedent("""
|
||
lazy import test.test_lazy_import.data.self_named_module_not_found
|
||
|
||
try:
|
||
_ = test.test_lazy_import.data.self_named_module_not_found
|
||
except ModuleNotFoundError as exc:
|
||
assert exc.name == "test.test_lazy_import.data.self_named_module_not_found", exc.name
|
||
assert str(exc) == "boom", exc
|
||
else:
|
||
raise AssertionError("ModuleNotFoundError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_none_in_sys_modules_during_submodule_resolution_propagates(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
sys.modules["test.test_lazy_import.data.blocked_module"] = None
|
||
lazy import test.test_lazy_import.data.blocked_module
|
||
|
||
try:
|
||
_ = test.test_lazy_import.data.blocked_module
|
||
except ModuleNotFoundError as exc:
|
||
assert exc.name == "test.test_lazy_import.data.blocked_module", exc.name
|
||
assert str(exc) == (
|
||
"import of test.test_lazy_import.data.blocked_module "
|
||
"halted; None in sys.modules"
|
||
), exc
|
||
else:
|
||
raise AssertionError("ModuleNotFoundError was not raised")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_circular_lazy_import_does_not_crash_for_gh_144727(self):
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
a_path = os.path.join(tmpdir, "a.py")
|
||
b_path = os.path.join(tmpdir, "b.py")
|
||
|
||
with open(a_path, "w") as f:
|
||
f.write(textwrap.dedent("""\
|
||
lazy import b
|
||
|
||
def something():
|
||
b.hello()
|
||
|
||
something()
|
||
"""))
|
||
|
||
with open(b_path, "w") as f:
|
||
f.write(textwrap.dedent("""\
|
||
lazy import a
|
||
|
||
def hello():
|
||
print(a)
|
||
"""))
|
||
|
||
result = subprocess.run(
|
||
[sys.executable, a_path],
|
||
capture_output=True,
|
||
text=True,
|
||
cwd=tmpdir,
|
||
)
|
||
# Should get a proper Python error, not a crash
|
||
self.assertEqual(result.returncode, 1)
|
||
self.assertIn("Error", result.stderr)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class GlobalsAndDictTests(LazyImportTestCase):
|
||
"""Tests for globals() and __dict__ behavior with lazy imports.
|
||
|
||
PEP 810: "Calling globals() or accessing a module's __dict__ does not trigger
|
||
reification – they return the module's dictionary, and accessing lazy objects
|
||
through that dictionary still returns lazy proxy objects."
|
||
"""
|
||
|
||
def test_globals_returns_lazy_proxy_when_accessed_from_function(self):
|
||
"""globals() accessed from a function should return lazy proxy without reification.
|
||
|
||
Note: At module level, accessing globals()['name'] triggers LOAD_NAME which
|
||
automatically resolves lazy imports. Inside a function, accessing globals()['name']
|
||
uses BINARY_SUBSCR which returns the lazy proxy without resolution.
|
||
"""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
lazy from test.test_lazy_import.data.basic2 import x
|
||
|
||
# Check that module is not yet loaded
|
||
assert 'test.test_lazy_import.data.basic2' not in sys.modules
|
||
|
||
def check_lazy():
|
||
# Access through globals() from inside a function
|
||
g = globals()
|
||
lazy_obj = g['x']
|
||
return type(lazy_obj) is types.LazyImportType
|
||
|
||
# Inside function, should get lazy proxy
|
||
is_lazy = check_lazy()
|
||
assert is_lazy, "Expected LazyImportType from function scope"
|
||
|
||
# Module should STILL not be loaded
|
||
assert 'test.test_lazy_import.data.basic2' not in sys.modules
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_globals_dict_access_returns_lazy_proxy_inline(self):
|
||
"""Accessing globals()['name'] inline should return lazy proxy.
|
||
|
||
Note: Assigning g['name'] to a local variable at module level triggers
|
||
reification due to STORE_NAME bytecode. Inline access preserves laziness.
|
||
"""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
lazy import json
|
||
# Inline access without assignment to local variable preserves lazy proxy
|
||
assert type(globals()['json']) is types.LazyImportType
|
||
assert 'json' not in sys.modules
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_module_dict_returns_lazy_proxy_without_reifying(self):
|
||
"""module.__dict__ access should not trigger reification."""
|
||
import test.test_lazy_import.data.globals_access
|
||
|
||
# Module not loaded yet via direct dict access
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
# Access via get_from_globals should return lazy proxy
|
||
lazy_obj = test.test_lazy_import.data.globals_access.get_from_globals()
|
||
self.assertEqual(type(lazy_obj), types.LazyImportType)
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_direct_access_triggers_reification(self):
|
||
"""Direct name access (not through globals()) should trigger reification."""
|
||
import test.test_lazy_import.data.globals_access
|
||
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
# Direct access should reify
|
||
result = test.test_lazy_import.data.globals_access.get_direct()
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_resolve_method_forces_reification(self):
|
||
"""Calling resolve() on lazy proxy should force reification.
|
||
|
||
Note: Must access lazy proxy from within a function to avoid automatic
|
||
reification by LOAD_NAME at module level.
|
||
"""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
lazy from test.test_lazy_import.data.basic2 import x
|
||
|
||
assert 'test.test_lazy_import.data.basic2' not in sys.modules
|
||
|
||
def test_resolve():
|
||
g = globals()
|
||
lazy_obj = g['x']
|
||
assert type(lazy_obj) is types.LazyImportType, f"Expected lazy proxy, got {type(lazy_obj)}"
|
||
|
||
resolved = lazy_obj.resolve()
|
||
|
||
# Now module should be loaded
|
||
assert 'test.test_lazy_import.data.basic2' in sys.modules
|
||
assert resolved == 42 # x is 42 in basic2.py
|
||
return True
|
||
|
||
assert test_resolve()
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_add_lazy_to_globals(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
lazy from test.test_lazy_import.data import basic2
|
||
|
||
assert 'test.test_lazy_import.data.basic2' not in sys.modules
|
||
|
||
class C: pass
|
||
sneaky = C()
|
||
sneaky.x = 1
|
||
|
||
def f():
|
||
t = 0
|
||
for _ in range(5):
|
||
t += sneaky.x
|
||
return t
|
||
|
||
f()
|
||
globals()["sneaky"] = globals()["basic2"]
|
||
assert f() == 210
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_add_lazy_to_exec_globals_after_specialization(self):
|
||
code = textwrap.dedent("""
|
||
source = '''
|
||
import sys
|
||
import types
|
||
|
||
lazy from test.test_lazy_import.data import basic2
|
||
|
||
assert 'test.test_lazy_import.data.basic2' not in sys.modules
|
||
|
||
class C: pass
|
||
sneaky = C()
|
||
sneaky.x = 1
|
||
|
||
def f():
|
||
t = 0
|
||
for _ in range(5):
|
||
t += sneaky.x
|
||
return t
|
||
|
||
f()
|
||
globals()["sneaky"] = globals()["basic2"]
|
||
assert f() == 210
|
||
print("OK")
|
||
'''
|
||
ns = {"__name__": "lazy_exec_globals"}
|
||
exec(source, ns)
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_add_lazy_to_exec_builtins_after_specialization(self):
|
||
code = textwrap.dedent("""
|
||
import builtins
|
||
source = '''
|
||
import sys
|
||
import types
|
||
|
||
lazy from test.test_lazy_import.data import basic2
|
||
|
||
assert 'test.test_lazy_import.data.basic2' not in sys.modules
|
||
|
||
class C: pass
|
||
sneaky = C()
|
||
sneaky.x = 1
|
||
__builtins__["sneaky"] = sneaky
|
||
del sneaky
|
||
|
||
def f():
|
||
t = 0
|
||
for _ in range(5):
|
||
t += sneaky.x
|
||
return t
|
||
|
||
f()
|
||
__builtins__["sneaky"] = globals()["basic2"]
|
||
assert f() == 210
|
||
print("OK")
|
||
'''
|
||
ns = {"__name__": "lazy_exec_builtins", "__builtins__": builtins.__dict__.copy()}
|
||
exec(source, ns)
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class MultipleNameFromImportTests(LazyImportTestCase):
|
||
"""Tests for lazy from ... import with multiple names.
|
||
|
||
PEP 810: "When using lazy from ... import, each imported name is bound to a
|
||
lazy proxy object. The first access to any of these names triggers loading
|
||
of the entire module and reifies only that specific name to its actual value.
|
||
Other names remain as lazy proxies until they are accessed."
|
||
"""
|
||
|
||
def test_accessing_one_name_leaves_others_as_proxies(self):
|
||
"""Accessing one name from multi-name import should leave others lazy."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
lazy from test.test_lazy_import.data.basic2 import f, x
|
||
|
||
# Neither should be loaded yet
|
||
assert 'test.test_lazy_import.data.basic2' not in sys.modules
|
||
|
||
g = globals()
|
||
assert type(g['f']) is types.LazyImportType
|
||
assert type(g['x']) is types.LazyImportType
|
||
|
||
# Access 'x' - this loads the module and reifies only 'x'
|
||
value = x
|
||
assert value == 42
|
||
|
||
# Module is now loaded
|
||
assert 'test.test_lazy_import.data.basic2' in sys.modules
|
||
|
||
# 'x' should be reified (int), 'f' should still be lazy proxy
|
||
assert type(g['x']) is int, f"Expected int, got {type(g['x'])}"
|
||
assert type(g['f']) is types.LazyImportType, f"Expected LazyImportType, got {type(g['f'])}"
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_accessing_one_name_imports_only_its_submodule(self):
|
||
"""Accessing one name should not import the other names' submodules."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
lazy from test.test_lazy_import.data.pkg import b, bar, broken
|
||
|
||
# Importing bar prints, and importing broken raises.
|
||
b.foo()
|
||
|
||
assert "test.test_lazy_import.data.pkg.bar" not in sys.modules
|
||
assert "test.test_lazy_import.data.pkg.broken" not in sys.modules
|
||
""")
|
||
rc, out, err = assert_python_ok("-c", code)
|
||
self.assertEqual(out, b"")
|
||
|
||
def test_all_names_reified_after_all_accessed(self):
|
||
"""All names should be reified after each is accessed."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
lazy from test.test_lazy_import.data.basic2 import f, x
|
||
|
||
g = globals()
|
||
|
||
# Access both
|
||
_ = x
|
||
_ = f
|
||
|
||
# Both should be reified now
|
||
assert type(g['x']) is int
|
||
assert callable(g['f'])
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class SysLazyModulesTrackingTests(LazyImportTestCase):
|
||
"""Tests for sys.lazy_modules tracking behavior.
|
||
|
||
PEP 810: "When the module is reified, it's removed from sys.lazy_modules"
|
||
"""
|
||
|
||
def test_module_added_to_lazy_modules_on_lazy_import(self):
|
||
"""Module should be added to sys.lazy_modules when lazily imported."""
|
||
# PEP 810 states lazy_modules tracks modules that have been lazily imported
|
||
# Note: The current implementation keeps modules in lazy_modules even after
|
||
# reification (primarily for diagnostics and introspection)
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
initial_count = len(sys.lazy_modules)
|
||
|
||
lazy import test.test_lazy_import.data.basic2
|
||
|
||
# Should be in lazy_modules after lazy import
|
||
assert "test.test_lazy_import.data.basic2" in sys.lazy_modules
|
||
assert len(sys.lazy_modules) > initial_count
|
||
|
||
# Trigger reification
|
||
_ = test.test_lazy_import.data.basic2.x
|
||
|
||
# Module should still be tracked (for diagnostics per PEP 810)
|
||
assert "test.test_lazy_import.data.basic2" not in sys.lazy_modules
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_lazy_modules_is_per_interpreter(self):
|
||
"""Each interpreter should have independent sys.lazy_modules."""
|
||
# Basic test that sys.lazy_modules exists and is a set
|
||
self.assertIsInstance(sys.lazy_modules, set)
|
||
|
||
def test_lazy_module_without_children_is_tracked(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
lazy import json
|
||
assert "json" in sys.lazy_modules, (
|
||
f"expected 'json' in sys.lazy_modules, got {set(sys.lazy_modules)}"
|
||
)
|
||
print("OK")
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_already_loaded_module_is_not_tracked(self):
|
||
"""A lazy import of an already loaded module should not be tracked."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
# Loaded by a regular import.
|
||
import json
|
||
lazy import json as lazy_json
|
||
assert "json" not in sys.lazy_modules, (
|
||
f"expected 'json' not in sys.lazy_modules, got {sys.lazy_modules}"
|
||
)
|
||
|
||
# Loaded by reifying an earlier lazy import.
|
||
lazy import base64
|
||
_ = base64.b64encode
|
||
lazy import base64 as lazy_base64
|
||
assert "base64" not in sys.lazy_modules, (
|
||
f"expected 'base64' not in sys.lazy_modules, got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_already_loaded_submodule_is_not_tracked(self):
|
||
"""`lazy from` a loaded submodule should not be tracked either."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import test.test_lazy_import.data.pkg.b
|
||
lazy from test.test_lazy_import.data.pkg import b
|
||
assert "test.test_lazy_import.data.pkg.b" not in sys.lazy_modules, (
|
||
f"expected 'pkg.b' untracked, got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_attribute_entry_removed_on_reification(self):
|
||
"""`lazy from x import attr` should untrack "x.attr" once resolved."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
lazy from test.test_lazy_import.data.basic2 import x
|
||
assert "test.test_lazy_import.data.basic2.x" in sys.lazy_modules, (
|
||
f"expected 'basic2.x' tracked, got {sys.lazy_modules}"
|
||
)
|
||
_ = x
|
||
assert "test.test_lazy_import.data.basic2.x" not in sys.lazy_modules, (
|
||
f"expected 'basic2.x' untracked, got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_already_loaded_attribute_is_not_tracked(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import math
|
||
|
||
lazy from math import pi
|
||
assert "math.pi" not in sys.lazy_modules, sys.lazy_modules
|
||
assert pi == math.pi
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_cached_lazy_attribute_tracks_its_target(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import test.test_lazy_import.data.basic_from_unused
|
||
|
||
holder = "test.test_lazy_import.data.basic_from_unused"
|
||
target = "test.test_lazy_import.data.basic2"
|
||
|
||
namespace = {"__name__": "cached_import_test"}
|
||
exec(f"lazy from {holder} import basic2", namespace)
|
||
assert holder + ".basic2" not in sys.lazy_modules, sys.lazy_modules
|
||
assert target in sys.lazy_modules, sys.lazy_modules
|
||
exec("assert basic2.x == 42", namespace)
|
||
assert target not in sys.lazy_modules, sys.lazy_modules
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_cached_attribute_keeps_pending_module_tracked(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import test.test_lazy_import.data.pkg as pkg
|
||
pkg.b = 42
|
||
|
||
lazy import test.test_lazy_import.data.pkg.b as pending
|
||
lazy from test.test_lazy_import.data.pkg import b
|
||
name = "test.test_lazy_import.data.pkg.b"
|
||
assert b == 42, b
|
||
assert name not in sys.modules, sys.modules
|
||
assert name in sys.lazy_modules, sys.lazy_modules
|
||
assert pending.foo() == "foo"
|
||
assert name not in sys.lazy_modules, sys.lazy_modules
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_failed_reification_stays_tracked(self):
|
||
"""A lazy import that fails to resolve must stay tracked."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
lazy import test.test_lazy_import.data.broken_module
|
||
try:
|
||
_ = test.test_lazy_import.data.broken_module
|
||
except ValueError:
|
||
pass
|
||
else:
|
||
raise AssertionError("ValueError was not raised")
|
||
assert "test.test_lazy_import.data.broken_module" in sys.lazy_modules, (
|
||
f"failed reification must stay tracked, got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_blocked_module_is_still_tracked(self):
|
||
"""A ``None`` entry in sys.modules must not count as loaded."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
sys.modules['test.test_lazy_import.data.basic2'] = None
|
||
lazy import test.test_lazy_import.data.basic2
|
||
assert "test.test_lazy_import.data.basic2" in sys.lazy_modules, (
|
||
f"blocked module must stay tracked, got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_initializing_module_is_still_tracked(self):
|
||
"""A module that is still executing must not count as loaded."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
name = "test.test_lazy_import.data.init_fails"
|
||
try:
|
||
import test.test_lazy_import.data.init_fails
|
||
except ValueError:
|
||
pass
|
||
else:
|
||
raise AssertionError("ValueError was not raised")
|
||
assert name not in sys.modules, "failed import left a module behind"
|
||
assert name in sys.lazy_modules, (
|
||
f"expected {name!r} tracked, got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_module_spec_descriptor_is_not_run(self):
|
||
"""Checking whether a module is loaded must not run its descriptors."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
class RaisingSpec(types.ModuleType):
|
||
@property
|
||
def __spec__(self):
|
||
raise RuntimeError("__spec__ descriptor was run")
|
||
|
||
sys.modules["raising_spec"] = RaisingSpec("raising_spec")
|
||
lazy import raising_spec
|
||
assert "raising_spec" not in sys.lazy_modules, (
|
||
f"expected 'raising_spec' untracked, got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_spec_initializing_descriptor_is_not_run(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
class Spec:
|
||
@property
|
||
def _initializing(self):
|
||
raise RuntimeError("_initializing descriptor was run")
|
||
|
||
@property
|
||
def __dict__(self):
|
||
raise RuntimeError("__dict__ descriptor was run")
|
||
|
||
class SlottedSpec:
|
||
__slots__ = ()
|
||
|
||
@property
|
||
def _initializing(self):
|
||
raise RuntimeError("_initializing descriptor was run")
|
||
|
||
for spec in (Spec(), SlottedSpec()):
|
||
module = types.ModuleType("custom_spec")
|
||
module.__spec__ = spec
|
||
sys.modules["custom_spec"] = module
|
||
lazy import custom_spec
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_cached_attribute_does_not_check_spec_twice(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
|
||
class Spec:
|
||
def __init__(self):
|
||
self.calls = 0
|
||
|
||
@property
|
||
def _initializing(self):
|
||
self.calls += 1
|
||
if self.calls == 2:
|
||
raise RuntimeError("_initializing was read twice")
|
||
return False
|
||
|
||
module = types.ModuleType("cached_spec")
|
||
module.__spec__ = Spec()
|
||
module.attr = 1
|
||
sys.modules["cached_spec"] = module
|
||
lazy from cached_spec import attr
|
||
assert attr == 1, attr
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_spec_initializing_truth_conversion_is_not_run(self):
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import types
|
||
from importlib.machinery import ModuleSpec
|
||
|
||
class Flag:
|
||
def __bool__(self):
|
||
raise RuntimeError("_initializing truth conversion was run")
|
||
|
||
released = memoryview(b"")
|
||
released.release()
|
||
for flag in (Flag(), released):
|
||
sys.lazy_modules.discard("custom_spec")
|
||
module = types.ModuleType("custom_spec")
|
||
module.__spec__ = ModuleSpec("custom_spec", None)
|
||
module.__spec__._initializing = flag
|
||
sys.modules["custom_spec"] = module
|
||
lazy import custom_spec
|
||
assert "custom_spec" in sys.lazy_modules, sys.lazy_modules
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
def test_pending_submodule_is_still_tracked(self):
|
||
"""`lazy from` a submodule that is not loaded must stay tracked."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
lazy from test.test_lazy_import.data.pkg import b
|
||
assert "test.test_lazy_import.data.pkg.b" in sys.lazy_modules, (
|
||
f"expected 'pkg.b' tracked, got {sys.lazy_modules}"
|
||
)
|
||
import test.test_lazy_import.data.pkg
|
||
assert "test.test_lazy_import.data.pkg.b" not in sys.modules, (
|
||
"loading the package must not load the submodule"
|
||
)
|
||
assert "test.test_lazy_import.data.pkg.b" in sys.lazy_modules, (
|
||
f"loading the package must not untrack the submodule, "
|
||
f"got {sys.lazy_modules}"
|
||
)
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class CommandLineAndEnvVarTests(unittest.TestCase):
|
||
"""Tests for command-line and environment variable control.
|
||
|
||
PEP 810: The global lazy imports flag can be controlled through:
|
||
- The -X lazy_imports=<mode> command-line option
|
||
- The PYTHON_LAZY_IMPORTS=<mode> environment variable
|
||
"""
|
||
|
||
def test_cli_lazy_imports_all_makes_regular_imports_lazy(self):
|
||
"""-X lazy_imports=all should make all imports potentially lazy."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
# In 'all' mode, regular imports become lazy
|
||
import json
|
||
# json should not be in sys.modules yet (lazy)
|
||
# Actually accessing it triggers reification
|
||
if 'json' not in sys.modules:
|
||
print("LAZY")
|
||
else:
|
||
print("EAGER")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-X", "lazy_imports=all", "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
||
self.assertIn("LAZY", result.stdout)
|
||
|
||
def test_cli_lazy_imports_none_is_rejected(self):
|
||
"""-X lazy_imports=none should be rejected."""
|
||
result = subprocess.run(
|
||
[sys.executable, "-X", "lazy_imports=none", "-c", "pass"],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("-X lazy_imports: invalid value", result.stderr)
|
||
self.assertIn("expected 'all' or 'normal'", result.stderr)
|
||
|
||
def test_cli_lazy_imports_normal_respects_lazy_keyword_only(self):
|
||
"""-X lazy_imports=normal should respect lazy keyword only."""
|
||
# Note: Use test modules instead of stdlib modules to avoid
|
||
# modules already loaded by the interpreter startup
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import test.test_lazy_import.data.basic2 # Should be eager
|
||
lazy import test.test_lazy_import.data.pkg.b # Should be lazy
|
||
|
||
eager_loaded = 'test.test_lazy_import.data.basic2' in sys.modules
|
||
lazy_loaded = 'test.test_lazy_import.data.pkg.b' in sys.modules
|
||
|
||
if eager_loaded and not lazy_loaded:
|
||
print("OK")
|
||
else:
|
||
print(f"FAIL: eager={eager_loaded}, lazy={lazy_loaded}")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-X", "lazy_imports=normal", "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_env_var_lazy_imports_all_enables_global_lazy(self):
|
||
"""PYTHON_LAZY_IMPORTS=all should enable global lazy imports."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import json
|
||
if 'json' not in sys.modules:
|
||
print("LAZY")
|
||
else:
|
||
print("EAGER")
|
||
""")
|
||
import os
|
||
env = os.environ.copy()
|
||
env["PYTHON_LAZY_IMPORTS"] = "all"
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True,
|
||
env=env
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
||
self.assertIn("LAZY", result.stdout)
|
||
|
||
def test_env_var_lazy_imports_none_is_rejected(self):
|
||
"""PYTHON_LAZY_IMPORTS=none should be rejected."""
|
||
import os
|
||
env = os.environ.copy()
|
||
env["PYTHON_LAZY_IMPORTS"] = "none"
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", "pass"],
|
||
capture_output=True,
|
||
text=True,
|
||
env=env
|
||
)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("PYTHON_LAZY_IMPORTS: invalid value", result.stderr)
|
||
self.assertIn("expected 'all' or 'normal'", result.stderr)
|
||
|
||
def test_cli_overrides_env_var(self):
|
||
"""Command-line option should take precedence over environment variable."""
|
||
# PEP 810: -X lazy_imports takes precedence over PYTHON_LAZY_IMPORTS
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import json
|
||
if 'json' in sys.modules:
|
||
print("EAGER")
|
||
else:
|
||
print("LAZY")
|
||
""")
|
||
import os
|
||
env = os.environ.copy()
|
||
env["PYTHON_LAZY_IMPORTS"] = "all" # env says all imports are lazy
|
||
result = subprocess.run(
|
||
[sys.executable, "-X", "lazy_imports=normal", "-c", code],
|
||
capture_output=True,
|
||
text=True,
|
||
env=env
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
||
# CLI should win, so a regular import should stay eager.
|
||
self.assertIn("EAGER", result.stdout)
|
||
|
||
def test_sys_set_lazy_imports_overrides_cli(self):
|
||
"""sys.set_lazy_imports() should take precedence over CLI option."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
sys.set_lazy_imports("normal") # Override CLI
|
||
import json
|
||
if 'json' in sys.modules:
|
||
print("EAGER")
|
||
else:
|
||
print("LAZY")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-X", "lazy_imports=all", "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
||
self.assertIn("EAGER", result.stdout)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class FilterFunctionSignatureTests(LazyImportTestCase):
|
||
"""Tests for the filter function signature per PEP 810.
|
||
|
||
PEP 810: func(importer: str, name: str, fromlist: tuple[str, ...] | None) -> bool
|
||
"""
|
||
|
||
def _run_subprocess_with_modules(self, code, files):
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
for relpath, contents in files.items():
|
||
path = os.path.join(tmpdir, relpath)
|
||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||
with open(path, "w", encoding="utf-8") as file:
|
||
file.write(textwrap.dedent(contents))
|
||
|
||
env = os.environ.copy()
|
||
env["PYTHONPATH"] = os.pathsep.join(
|
||
entry for entry in (tmpdir, env.get("PYTHONPATH")) if entry
|
||
)
|
||
env["PYTHON_LAZY_IMPORTS"] = "normal"
|
||
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", textwrap.dedent(code)],
|
||
capture_output=True,
|
||
cwd=tmpdir,
|
||
env=env,
|
||
text=True,
|
||
)
|
||
return result
|
||
|
||
def _assert_subprocess_ok(self, code, files):
|
||
result = self._run_subprocess_with_modules(code, files)
|
||
self.assertEqual(
|
||
result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}"
|
||
)
|
||
return result
|
||
|
||
def test_filter_receives_correct_arguments_for_import(self):
|
||
"""Filter should receive (importer, name, fromlist=None) for 'import x'."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
received_args = []
|
||
|
||
def my_filter(importer, name, fromlist):
|
||
received_args.append((importer, name, fromlist))
|
||
return True
|
||
|
||
sys.set_lazy_imports_filter(my_filter)
|
||
|
||
lazy import json
|
||
|
||
assert len(received_args) == 1, f"Expected 1 call, got {len(received_args)}"
|
||
importer, name, fromlist = received_args[0]
|
||
assert name == "json", f"Expected name='json', got {name!r}"
|
||
assert fromlist is None, f"Expected fromlist=None, got {fromlist!r}"
|
||
assert isinstance(importer, str), f"Expected str importer, got {type(importer)}"
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_filter_receives_fromlist_for_from_import(self):
|
||
"""Filter should receive fromlist tuple for 'from x import y, z'."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
received_args = []
|
||
|
||
def my_filter(importer, name, fromlist):
|
||
received_args.append((importer, name, fromlist))
|
||
return True
|
||
|
||
sys.set_lazy_imports_filter(my_filter)
|
||
|
||
lazy from json import dumps, loads
|
||
|
||
assert len(received_args) == 1, f"Expected 1 call, got {len(received_args)}"
|
||
importer, name, fromlist = received_args[0]
|
||
assert name == "json", f"Expected name='json', got {name!r}"
|
||
assert fromlist == ("dumps", "loads"), f"Expected ('dumps', 'loads'), got {fromlist!r}"
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_filter_returning_false_forces_eager_import(self):
|
||
"""Filter returning False should make import eager."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
def deny_filter(importer, name, fromlist):
|
||
return False
|
||
|
||
sys.set_lazy_imports_filter(deny_filter)
|
||
|
||
lazy import json
|
||
|
||
# Should be eager due to filter
|
||
if 'json' in sys.modules:
|
||
print("EAGER")
|
||
else:
|
||
print("LAZY")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stderr: {result.stderr}")
|
||
self.assertIn("EAGER", result.stdout)
|
||
|
||
def test_filter_distinguishes_absolute_and_relative_from_imports(self):
|
||
"""Relative imports should pass resolved module names to the filter."""
|
||
files = {
|
||
"target.py": """
|
||
VALUE = "absolute"
|
||
""",
|
||
"pkg/__init__.py": "",
|
||
"pkg/target.py": """
|
||
VALUE = "relative"
|
||
""",
|
||
"pkg/runner.py": """
|
||
import sys
|
||
|
||
seen = []
|
||
|
||
def my_filter(importer, name, fromlist):
|
||
seen.append((importer, name, fromlist))
|
||
return True
|
||
|
||
sys.set_lazy_imports_filter(my_filter)
|
||
|
||
lazy from target import VALUE as absolute_value
|
||
lazy from .target import VALUE as relative_value
|
||
|
||
assert seen == [
|
||
(__name__, "target", ("VALUE",)),
|
||
(__name__, "pkg.target", ("VALUE",)),
|
||
], seen
|
||
""",
|
||
}
|
||
|
||
result = self._assert_subprocess_ok(
|
||
"""
|
||
import pkg.runner
|
||
print("OK")
|
||
""",
|
||
files,
|
||
)
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_filter_receives_resolved_name_for_relative_package_import(self):
|
||
"""'lazy from . import x' should report the resolved package name."""
|
||
files = {
|
||
"pkg/__init__.py": "",
|
||
"pkg/sibling.py": """
|
||
VALUE = 1
|
||
""",
|
||
"pkg/runner.py": """
|
||
import sys
|
||
|
||
seen = []
|
||
|
||
def my_filter(importer, name, fromlist):
|
||
seen.append((importer, name, fromlist))
|
||
return True
|
||
|
||
sys.set_lazy_imports_filter(my_filter)
|
||
|
||
lazy from . import sibling
|
||
|
||
assert seen == [
|
||
(__name__, "pkg", ("sibling",)),
|
||
], seen
|
||
""",
|
||
}
|
||
|
||
result = self._assert_subprocess_ok(
|
||
"""
|
||
import pkg.runner
|
||
print("OK")
|
||
""",
|
||
files,
|
||
)
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_filter_receives_resolved_name_for_parent_relative_import(self):
|
||
"""Parent relative imports should also use the resolved module name."""
|
||
files = {
|
||
"pkg/__init__.py": "",
|
||
"pkg/target.py": """
|
||
VALUE = 1
|
||
""",
|
||
"pkg/sub/__init__.py": "",
|
||
"pkg/sub/runner.py": """
|
||
import sys
|
||
|
||
seen = []
|
||
|
||
def my_filter(importer, name, fromlist):
|
||
seen.append((importer, name, fromlist))
|
||
return True
|
||
|
||
sys.set_lazy_imports_filter(my_filter)
|
||
|
||
lazy from ..target import VALUE
|
||
|
||
assert seen == [
|
||
(__name__, "pkg.target", ("VALUE",)),
|
||
], seen
|
||
""",
|
||
}
|
||
|
||
result = self._assert_subprocess_ok(
|
||
"""
|
||
import pkg.sub.runner
|
||
print("OK")
|
||
""",
|
||
files,
|
||
)
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_filter_can_force_eager_only_for_resolved_relative_target(self):
|
||
"""Resolved names should let filters treat relative and absolute imports differently."""
|
||
files = {
|
||
"target.py": """
|
||
VALUE = "absolute"
|
||
""",
|
||
"pkg/__init__.py": "",
|
||
"pkg/target.py": """
|
||
VALUE = "relative"
|
||
""",
|
||
"pkg/runner.py": """
|
||
import sys
|
||
|
||
def my_filter(importer, name, fromlist):
|
||
return name != "pkg.target"
|
||
|
||
sys.set_lazy_imports_filter(my_filter)
|
||
|
||
lazy from target import VALUE as absolute_value
|
||
lazy from .target import VALUE as relative_value
|
||
|
||
assert "pkg.target" in sys.modules, sorted(
|
||
name for name in sys.modules
|
||
if name in {"target", "pkg.target"}
|
||
)
|
||
assert "target" not in sys.modules, sorted(
|
||
name for name in sys.modules
|
||
if name in {"target", "pkg.target"}
|
||
)
|
||
assert relative_value == "relative", relative_value
|
||
""",
|
||
}
|
||
|
||
result = self._assert_subprocess_ok(
|
||
"""
|
||
import pkg.runner
|
||
print("OK")
|
||
""",
|
||
files,
|
||
)
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
class AdditionalSyntaxRestrictionTests(LazyImportTestCase):
|
||
"""Additional syntax restriction tests per PEP 810."""
|
||
|
||
def test_lazy_import_inside_class_raises_syntax_error(self):
|
||
"""lazy import inside class body should raise SyntaxError."""
|
||
# PEP 810: "The soft keyword is only allowed at the global (module) level,
|
||
# not inside functions, class bodies, try blocks, or import *"
|
||
with self.assertRaises(SyntaxError):
|
||
import test.test_lazy_import.data.badsyntax.lazy_class_body
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class MixedLazyEagerImportTests(LazyImportTestCase):
|
||
"""Tests for mixing lazy and eager imports of the same module.
|
||
|
||
PEP 810: "If module foo is imported both lazily and eagerly in the same
|
||
program, the eager import takes precedence and both bindings resolve to
|
||
the same module object."
|
||
"""
|
||
|
||
def test_eager_import_before_lazy_resolves_to_same_module(self):
|
||
"""Eager import before lazy should make lazy resolve to same module."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import json # Eager import first
|
||
|
||
lazy import json as lazy_json # Lazy import same module
|
||
|
||
# lazy_json should resolve to the same object
|
||
assert json is lazy_json, "Lazy and eager imports should resolve to same module"
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_lazy_import_before_eager_resolves_to_same_module(self):
|
||
"""Lazy import followed by eager should both point to same module."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
|
||
lazy import json as lazy_json
|
||
|
||
# Lazy not loaded yet
|
||
assert 'json' not in sys.modules
|
||
|
||
import json # Eager import triggers load
|
||
|
||
# Both should be the same object
|
||
assert json is lazy_json
|
||
print("OK")
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
class RelativeImportTests(LazyImportTestCase):
|
||
"""Tests for relative imports with lazy keyword."""
|
||
|
||
def test_relative_lazy_import(self):
|
||
"""lazy from . import submodule should work."""
|
||
from test.test_lazy_import.data import relative_lazy
|
||
|
||
# basic2 should not be loaded yet
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
# Access triggers reification
|
||
result = relative_lazy.get_basic2()
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
def test_relative_lazy_from_import(self):
|
||
"""lazy from .module import name should work."""
|
||
from test.test_lazy_import.data import relative_lazy_from
|
||
|
||
# basic2 should not be loaded yet
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
# Access triggers reification
|
||
result = relative_lazy_from.get_x()
|
||
self.assertEqual(result, 42)
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
|
||
class LazyModulesCompatibilityFromImportTests(LazyImportTestCase):
|
||
"""Tests for __lazy_modules__ with from imports.
|
||
|
||
PEP 810: "When a module is made lazy this way, from-imports using that
|
||
module are also lazy"
|
||
"""
|
||
|
||
def test_lazy_modules_makes_from_imports_lazy(self):
|
||
"""__lazy_modules__ should make from imports of listed modules lazy."""
|
||
from test.test_lazy_import.data import lazy_compat_from
|
||
|
||
# basic2 should not be loaded yet because it's in __lazy_modules__
|
||
self.assertNotIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
# Access triggers reification
|
||
result = lazy_compat_from.get_x()
|
||
self.assertEqual(result, 42)
|
||
self.assertIn("test.test_lazy_import.data.basic2", sys.modules)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class ImportStateAtReificationTests(LazyImportTestCase):
|
||
"""Tests for import system state at reification time.
|
||
|
||
PEP 810: "Reification still calls __import__ to resolve the import, which uses
|
||
the state of the import system (e.g. sys.path, sys.meta_path, sys.path_hooks
|
||
and __import__) at reification time, not the state when the lazy import
|
||
statement was evaluated."
|
||
"""
|
||
|
||
def test_sys_path_at_reification_time_is_used(self):
|
||
"""sys.path changes after lazy import should affect reification."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import tempfile
|
||
import os
|
||
|
||
# Create a temporary module
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
mod_path = os.path.join(tmpdir, "dynamic_test_module.py")
|
||
with open(mod_path, "w") as f:
|
||
f.write("VALUE = 'from_temp_dir'\\n")
|
||
|
||
# Lazy import before adding to path
|
||
lazy import dynamic_test_module
|
||
|
||
# Module cannot be found yet
|
||
try:
|
||
_ = dynamic_test_module
|
||
print("FAIL - should not find module")
|
||
except ModuleNotFoundError:
|
||
pass
|
||
|
||
# Now add temp dir to path
|
||
sys.path.insert(0, tmpdir)
|
||
|
||
# Now reification should succeed using current sys.path
|
||
assert dynamic_test_module.VALUE == 'from_temp_dir'
|
||
print("OK")
|
||
|
||
sys.path.remove(tmpdir)
|
||
""")
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class ThreadSafetyTests(LazyImportTestCase):
|
||
"""Tests for thread-safety of lazy imports."""
|
||
|
||
def test_concurrent_lazy_import_reification(self):
|
||
"""Multiple threads racing to reify the same lazy import should succeed."""
|
||
from test.test_lazy_import.data import basic_unused
|
||
|
||
num_threads = 10
|
||
results = [None] * num_threads
|
||
errors = []
|
||
barrier = threading.Barrier(num_threads)
|
||
|
||
def access_lazy_import(idx):
|
||
try:
|
||
barrier.wait()
|
||
module = basic_unused.test.test_lazy_import.data.basic2
|
||
results[idx] = module
|
||
except Exception as e:
|
||
errors.append((idx, e))
|
||
|
||
threads = [
|
||
threading.Thread(target=access_lazy_import, args=(i,))
|
||
for i in range(num_threads)
|
||
]
|
||
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
|
||
self.assertEqual(errors, [], f"Errors occurred: {errors}")
|
||
self.assertTrue(all(r is not None for r in results))
|
||
first_module = results[0]
|
||
for r in results[1:]:
|
||
self.assertIs(r, first_module)
|
||
|
||
def test_concurrent_reification_multiple_modules(self):
|
||
"""Multiple threads reifying different lazy imports concurrently."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import threading
|
||
|
||
sys.set_lazy_imports("all")
|
||
|
||
lazy import json
|
||
lazy import os
|
||
lazy import io
|
||
lazy import re
|
||
|
||
num_threads = 8
|
||
results = {}
|
||
errors = []
|
||
barrier = threading.Barrier(num_threads)
|
||
|
||
def access_modules(idx):
|
||
try:
|
||
barrier.wait()
|
||
mods = [json, os, io, re]
|
||
results[idx] = [type(m).__name__ for m in mods]
|
||
except Exception as e:
|
||
errors.append((idx, e))
|
||
|
||
threads = [
|
||
threading.Thread(target=access_modules, args=(i,))
|
||
for i in range(num_threads)
|
||
]
|
||
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
|
||
assert not errors, f"Errors: {errors}"
|
||
for idx, mods in results.items():
|
||
assert all(m == 'module' for m in mods), f"Thread {idx} got: {mods}"
|
||
|
||
print("OK")
|
||
""")
|
||
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_concurrent_lazy_modules_dict_updates(self):
|
||
"""Multiple threads creating lazy imports should safely update sys.lazy_modules."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import threading
|
||
|
||
num_threads = 16
|
||
iterations = 50
|
||
errors = []
|
||
barrier = threading.Barrier(num_threads)
|
||
|
||
def create_lazy_imports(idx):
|
||
try:
|
||
barrier.wait()
|
||
for i in range(iterations):
|
||
exec(f"lazy import json as json_{idx}_{i}", globals())
|
||
exec(f"lazy import os as os_{idx}_{i}", globals())
|
||
except Exception as e:
|
||
errors.append((idx, e))
|
||
|
||
threads = [
|
||
threading.Thread(target=create_lazy_imports, args=(i,))
|
||
for i in range(num_threads)
|
||
]
|
||
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
|
||
assert not errors, f"Errors: {errors}"
|
||
assert isinstance(sys.lazy_modules, set), "sys.lazy_modules is not a dict"
|
||
print("OK")
|
||
""")
|
||
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_concurrent_reification_same_module_high_contention(self):
|
||
"""High contention: many threads reifying the exact same lazy import."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import threading
|
||
import types
|
||
|
||
sys.set_lazy_imports("all")
|
||
|
||
lazy import json
|
||
|
||
num_threads = 20
|
||
results = [None] * num_threads
|
||
errors = []
|
||
barrier = threading.Barrier(num_threads)
|
||
|
||
def access_json(idx):
|
||
try:
|
||
barrier.wait()
|
||
for _ in range(100):
|
||
_ = json.dumps
|
||
_ = json.loads
|
||
results[idx] = json
|
||
except Exception as e:
|
||
errors.append((idx, e))
|
||
|
||
threads = [
|
||
threading.Thread(target=access_json, args=(i,))
|
||
for i in range(num_threads)
|
||
]
|
||
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
|
||
assert not errors, f"Errors: {errors}"
|
||
assert all(r is not None for r in results), "Some threads got None"
|
||
first = results[0]
|
||
assert all(r is first for r in results), "Inconsistent module objects"
|
||
assert not isinstance(first, types.LazyImportType), "Got lazy import instead of module"
|
||
print("OK")
|
||
""")
|
||
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
def test_concurrent_reification_with_module_attribute_access(self):
|
||
"""Threads racing to reify and immediately access module attributes."""
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import threading
|
||
|
||
sys.set_lazy_imports("all")
|
||
|
||
lazy import collections
|
||
lazy import functools
|
||
lazy import itertools
|
||
|
||
num_threads = 12
|
||
results = {}
|
||
errors = []
|
||
barrier = threading.Barrier(num_threads)
|
||
|
||
def stress_lazy_imports(idx):
|
||
try:
|
||
barrier.wait()
|
||
for _ in range(50):
|
||
_ = collections.OrderedDict
|
||
_ = functools.partial
|
||
_ = itertools.chain
|
||
_ = collections.defaultdict
|
||
_ = functools.lru_cache
|
||
_ = itertools.islice
|
||
results[idx] = (
|
||
type(collections).__name__,
|
||
type(functools).__name__,
|
||
type(itertools).__name__,
|
||
)
|
||
except Exception as e:
|
||
errors.append((idx, e))
|
||
|
||
threads = [
|
||
threading.Thread(target=stress_lazy_imports, args=(i,))
|
||
for i in range(num_threads)
|
||
]
|
||
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join()
|
||
|
||
assert not errors, f"Errors: {errors}"
|
||
for idx, types_tuple in results.items():
|
||
assert all(t == 'module' for t in types_tuple), f"Thread {idx}: {types_tuple}"
|
||
print("OK")
|
||
""")
|
||
|
||
result = subprocess.run(
|
||
[sys.executable, "-c", code],
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
self.assertEqual(result.returncode, 0, f"stdout: {result.stdout}, stderr: {result.stderr}")
|
||
self.assertIn("OK", result.stdout)
|
||
|
||
|
||
class LazyImportDisTests(unittest.TestCase):
|
||
def test_lazy_import_dis(self):
|
||
"""dis should properly show lazy import"""
|
||
code = compile("lazy import foo", "exec", "exec")
|
||
f = io.StringIO()
|
||
dis.dis(code, file=f)
|
||
self.assertIn("foo + lazy", f.getvalue())
|
||
|
||
def test_normal_import_dis(self):
|
||
"""non lazy imports should just show the name"""
|
||
code = compile("import foo", "exec", "exec")
|
||
f = io.StringIO()
|
||
dis.dis(code, file=f)
|
||
for line in f.getvalue().split('\n'):
|
||
if "IMPORT_NAME" in line:
|
||
self.assertIn("(foo)", line)
|
||
break
|
||
else:
|
||
self.assertFail("IMPORT_NAME not found")
|
||
|
||
|
||
@unittest.skipIf(_testcapi is None, 'need the _testcapi module')
|
||
class LazyCApiTests(LazyImportTestCase):
|
||
def test_set_matches_sys(self):
|
||
self.assertEqual(_testcapi.PyImport_GetLazyImportsMode(), sys.get_lazy_imports())
|
||
for mode in ("normal", "all"):
|
||
_testcapi.PyImport_SetLazyImportsMode(mode)
|
||
self.assertEqual(_testcapi.PyImport_GetLazyImportsMode(), sys.get_lazy_imports())
|
||
self.assertRaises(ValueError, _testcapi.PyImport_SetLazyImportsMode, "none")
|
||
|
||
def test_filter_matches_sys(self):
|
||
self.assertEqual(_testcapi.PyImport_GetLazyImportsFilter(), sys.get_lazy_imports_filter())
|
||
|
||
def filter(*args):
|
||
pass
|
||
|
||
_testcapi.PyImport_SetLazyImportsFilter(filter)
|
||
self.assertEqual(_testcapi.PyImport_GetLazyImportsFilter(), sys.get_lazy_imports_filter())
|
||
|
||
def test_set_bad_filter(self):
|
||
self.assertRaises(ValueError, _testcapi.PyImport_SetLazyImportsFilter, 42)
|
||
|
||
def test_dunder_lazy_import_without_frame(self):
|
||
# gh-151510: __lazy_import__() called with no globals and no running
|
||
# Python frame must raise TypeError instead of crashing.
|
||
with self.assertRaisesRegex(
|
||
TypeError,
|
||
r"__lazy_import__\(\) missing globals when called without a frame",
|
||
):
|
||
_testcapi.lazy_import_without_frame(
|
||
"test.test_lazy_import.data.basic2"
|
||
)
|
||
|
||
|
||
class DottedLazyImportTests(unittest.TestCase):
|
||
|
||
def test_lazy_import_as_wins_over_variable(self):
|
||
"""A dotted lazy import as imports the submodule the variable hides."""
|
||
# gh-157757: importing pkg.b rebinds pkg.b from the variable to the
|
||
# module, eagerly and lazily alike.
|
||
code = textwrap.dedent("""
|
||
import sys
|
||
import test.test_lazy_import.data.pkg as pkg
|
||
pkg.b = "hides the b submodule"
|
||
|
||
lazy import test.test_lazy_import.data.pkg.b as b
|
||
lazy import xml.dom.minidom as minidom
|
||
|
||
assert b is sys.modules["test.test_lazy_import.data.pkg.b"], b
|
||
assert minidom is sys.modules["xml.dom.minidom"], minidom
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
def test_dotted_as_of_loaded_module(self):
|
||
"""A dotted lazy import as binds the module, not a same-named attribute."""
|
||
# importlib.metadata is already loaded and has a `metadata` attribute.
|
||
code = textwrap.dedent("""
|
||
import importlib.metadata
|
||
import importlib.metadata as eager
|
||
|
||
lazy import importlib.metadata as lazily
|
||
|
||
assert lazily is eager, lazily
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
def test_dotted_as_replays_lookups_on_custom_placeholder(self):
|
||
"""A dotted lazy import as looks up its names on what the hook returned."""
|
||
code = textwrap.dedent("""
|
||
import builtins
|
||
import xml.dom
|
||
|
||
# In a list, so the hook reading it does not resolve it.
|
||
placeholder = [__lazy_import__("xml")]
|
||
default = builtins.__lazy_import__
|
||
builtins.__lazy_import__ = lambda *args: placeholder[0]
|
||
lazy import fake.dom as dom
|
||
builtins.__lazy_import__ = default
|
||
|
||
assert dom is xml.dom, dom
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
def test_empty_fromlist_placeholder_matches_no_fromlist(self):
|
||
"""An empty fromlist behaves like None."""
|
||
code = textwrap.dedent("""
|
||
expected = "<lazy_import 'xml.dom'>"
|
||
# In lists, so reading them does not resolve them.
|
||
for fromlist in (None, ()):
|
||
same = [__lazy_import__("xml.dom", fromlist=fromlist)]
|
||
assert repr(same[0]) == expected, (fromlist, repr(same[0]))
|
||
bare = [__lazy_import__("xml.dom")]
|
||
assert repr(bare[0]) == expected, repr(bare[0])
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
def test_empty_fromlist_preserved_for_custom_import(self):
|
||
code = textwrap.dedent("""
|
||
import builtins
|
||
import types
|
||
|
||
value = object()
|
||
module = types.SimpleNamespace(dom=value)
|
||
placeholder = [__lazy_import__("xml.dom", fromlist=())]
|
||
default_import = builtins.__import__
|
||
default_lazy_import = builtins.__lazy_import__
|
||
calls = []
|
||
|
||
def import_hook(name, globals, locals, fromlist, level):
|
||
assert name == "xml.dom", name
|
||
assert fromlist == (), fromlist
|
||
calls.append(fromlist)
|
||
return module
|
||
|
||
builtins.__import__ = import_hook
|
||
assert placeholder[0].resolve() is module
|
||
builtins.__lazy_import__ = lambda *args: placeholder[0]
|
||
lazy import fake.dom as dom
|
||
assert dom is value
|
||
builtins.__import__ = default_import
|
||
builtins.__lazy_import__ = default_lazy_import
|
||
|
||
assert calls == [(), ()], calls
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
def test_dotted_as_replays_lookups_on_dotted_placeholder(self):
|
||
"""A dotted lazy import as replays its names on the hook's package."""
|
||
# importlib.metadata has a `metadata` attribute of its own, which the
|
||
# placeholder for importlib must not answer with.
|
||
for target in ("xml.dom", "importlib.metadata"):
|
||
with self.subTest(target=target):
|
||
leaf = target.rpartition(".")[2]
|
||
code = textwrap.dedent(f"""
|
||
import builtins
|
||
import sys
|
||
import {target}
|
||
|
||
# In a list, so the hook reading it does not resolve it.
|
||
placeholder = [__lazy_import__("{target}", fromlist=())]
|
||
default = builtins.__lazy_import__
|
||
builtins.__lazy_import__ = lambda *args: placeholder[0]
|
||
lazy import fake.{leaf} as {leaf}
|
||
builtins.__lazy_import__ = default
|
||
|
||
name = repr(globals()["{leaf}"])
|
||
assert name == "<lazy_import '{target}'>", name
|
||
assert {leaf} is sys.modules["{target}"], {leaf}
|
||
""")
|
||
assert_python_ok("-c", code)
|
||
|
||
|
||
class SubmoduleLazinessTests(unittest.TestCase):
|
||
"""Tests that module-level lazy imports remain lazy until accessed."""
|
||
|
||
def tearDown(self):
|
||
for key in list(sys.modules.keys()):
|
||
if key.startswith('test.test_lazy_import.data'):
|
||
del sys.modules[key]
|
||
sys.set_lazy_imports_filter(None)
|
||
sys.set_lazy_imports("normal")
|
||
|
||
def test_unaccessed_imports_stay_lazy(self):
|
||
"""Imports in 'all' mode should stay lazy until accessed."""
|
||
sys.set_lazy_imports("all")
|
||
from test.test_lazy_import.data.metasyntactic import names
|
||
self.assertIsInstance(names.__dict__["Foo"], types.LazyImportType)
|
||
self.assertNotIn(
|
||
"test.test_lazy_import.data.metasyntactic.foo", sys.modules
|
||
)
|
||
_ = names.Foo
|
||
self.assertEqual(names.Foo, "Foo")
|
||
self.assertIn(
|
||
"test.test_lazy_import.data.metasyntactic.foo", sys.modules
|
||
)
|
||
self.assertIsInstance(names.__dict__["Ack"], types.LazyImportType)
|
||
self.assertNotIn(
|
||
"test.test_lazy_import.data.metasyntactic.foo.ack", sys.modules
|
||
)
|
||
|
||
|
||
class AttributeSideEffectTests(unittest.TestCase):
|
||
"""Tests that submodule imports don't overwrite parent attributes."""
|
||
|
||
def tearDown(self):
|
||
for key in list(sys.modules.keys()):
|
||
if key.startswith('test.test_lazy_import.data'):
|
||
del sys.modules[key]
|
||
sys.set_lazy_imports_filter(None)
|
||
sys.set_lazy_imports("normal")
|
||
|
||
def test_version_submodule_does_not_overwrite(self):
|
||
"""A __version__ submodule should not overwrite the parent's
|
||
__version__ attribute imported in __init__.py."""
|
||
import test.test_lazy_import.data.versioned as versioned
|
||
self.assertEqual(versioned.__version__, "1.0")
|
||
self.assertEqual(
|
||
versioned.__copyright__,
|
||
"Copyright (c) 2001-2022 Python Software Foundation.",
|
||
)
|
||
|
||
|
||
class ModuleVariableNameCollisionTests(unittest.TestCase):
|
||
"""Tests for name collision between a submodule and a variable."""
|
||
|
||
def tearDown(self):
|
||
for key in list(sys.modules.keys()):
|
||
if key.startswith('test.test_lazy_import.data'):
|
||
del sys.modules[key]
|
||
sys.set_lazy_imports_filter(None)
|
||
sys.set_lazy_imports("normal")
|
||
|
||
def test_variable_after_import_wins(self):
|
||
"""Variable assigned after import should overwrite the submodule."""
|
||
from test.test_lazy_import.data import module_same_name_var_order1
|
||
self.assertEqual(module_same_name_var_order1.bar, "Blah")
|
||
|
||
def test_import_after_variable_wins(self):
|
||
"""Import after variable assignment should overwrite the variable."""
|
||
from test.test_lazy_import.data import module_same_name_var_order2
|
||
bar_mod = sys.modules[
|
||
"test.test_lazy_import.data.module_same_name_var_order2.bar"
|
||
]
|
||
self.assertIs(module_same_name_var_order2.bar, bar_mod)
|
||
|
||
|
||
class DeletedModuleReimportTests(unittest.TestCase):
|
||
"""Tests for reimporting after module deletion from sys.modules."""
|
||
|
||
def tearDown(self):
|
||
for key in list(sys.modules.keys()):
|
||
if key.startswith('test.test_lazy_import.data'):
|
||
del sys.modules[key]
|
||
sys.set_lazy_imports_filter(None)
|
||
sys.set_lazy_imports("normal")
|
||
|
||
def test_reimport_creates_new_module(self):
|
||
"""Deleting and reimporting should create a new module object."""
|
||
import test.test_lazy_import.data.metasyntactic.foo
|
||
import test.test_lazy_import.data.metasyntactic.foo.bar.baz
|
||
|
||
first_bar = test.test_lazy_import.data.metasyntactic.foo.bar
|
||
|
||
del sys.modules[
|
||
"test.test_lazy_import.data.metasyntactic.foo.bar"
|
||
]
|
||
|
||
import test.test_lazy_import.data.metasyntactic.foo.bar.thud
|
||
|
||
second_bar = test.test_lazy_import.data.metasyntactic.foo.bar
|
||
|
||
self.assertIsNot(first_bar, second_bar)
|
||
self.assertIn("baz", dir(first_bar))
|
||
self.assertNotIn("thud", dir(first_bar))
|
||
self.assertIn("thud", dir(second_bar))
|
||
self.assertNotIn("baz", dir(second_bar))
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class CircularImportLazyTests(unittest.TestCase):
|
||
"""Tests that lazy imports can break circular import patterns."""
|
||
|
||
def test_succeeds_with_lazy(self):
|
||
"""Same-level circular imports should succeed with lazy mode."""
|
||
proc = assert_python_ok(
|
||
"-X", "lazy_imports=all", "-c",
|
||
"import test.test_lazy_import.data.circular_import_pkg.main;"
|
||
"print('OK')",
|
||
)
|
||
self.assertIn(b"OK", proc.out)
|
||
|
||
def test_fails_without_lazy(self):
|
||
"""Same-level circular imports should fail without lazy mode."""
|
||
result = subprocess.run(
|
||
[sys.executable, "-c",
|
||
"import test.test_lazy_import.data.circular_import_pkg.main"],
|
||
capture_output=True, text=True,
|
||
)
|
||
self.assertNotEqual(result.returncode, 0)
|
||
self.assertIn("ImportError", result.stderr)
|
||
|
||
|
||
@support.requires_subprocess()
|
||
class LazyImportFinalizationTests(unittest.TestCase):
|
||
"""Destructors that use lazy imports after finalization freed their state."""
|
||
|
||
def test_declare(self):
|
||
code = textwrap.dedent("""
|
||
import builtins, os
|
||
|
||
class Canary:
|
||
def __del__(self, write=os.write, exec=exec,
|
||
builtins={"__lazy_import__": builtins.__lazy_import__}):
|
||
exec("lazy import a.b", {"__builtins__": builtins})
|
||
write(1, b"ok")
|
||
|
||
# Only this placeholder keeps the canary alive.
|
||
exec("lazy import pkg.mod", {"__builtins__": {
|
||
"__lazy_import__": builtins.__lazy_import__, "c": Canary()}})
|
||
""")
|
||
self.assertEqual(assert_python_ok("-c", code).out, b"ok")
|
||
|
||
def test_resolve(self):
|
||
code = textwrap.dedent("""
|
||
import contextvars, json, os, types
|
||
|
||
def safe_import(name, *args, allowed={"json": json}):
|
||
return allowed[name]
|
||
|
||
# Not in sys.modules, with builtins that still work at shutdown.
|
||
ns = types.ModuleType("ns")
|
||
ns.json = __lazy_import__("json", {"__builtins__": {"__import__": safe_import}})
|
||
|
||
class Canary:
|
||
def __del__(self, write=os.write, type=type, ns=ns):
|
||
write(1, type(ns.json).__name__.encode())
|
||
|
||
contextvars.ContextVar("v").set(Canary())
|
||
""")
|
||
self.assertEqual(assert_python_ok("-c", code).out, b"module")
|
||
|
||
def test_set_lazy_attributes(self):
|
||
code = textwrap.dedent("""
|
||
import _imp, os, sys
|
||
|
||
class Canary:
|
||
def __del__(self, write=os.write,
|
||
set_lazy=_imp._set_lazy_attributes):
|
||
set_lazy(None, "mod")
|
||
write(1, b"ok")
|
||
|
||
# Freed while sys.lazy_modules is being cleared.
|
||
sys.lazy_modules.add(Canary())
|
||
""")
|
||
self.assertEqual(assert_python_ok("-c", code).out, b"ok")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|