gh-152298: Fix LazyImportType resolve reification

Author: 1sgtpepper

Include/internal/pycore_lazyimportobject.h

@@ -19,6 +19,9 @@ typedef struct {
     PyObject *lz_builtins;
     PyObject *lz_from;
     PyObject *lz_attr;
+    PyObject *lz_owner_globals;
+    PyObject *lz_owner_key;
+    PyObject *lz_resolved;
     // Frame information for the original import location.
     PyCodeObject *lz_code;     // Code object where the lazy import was created.
     int lz_instr_offset;       // Instruction offset where the lazy import was created.
@@ -28,6 +31,21 @@ typedef struct {
 PyAPI_FUNC(PyObject *) _PyLazyImport_GetName(PyObject *lazy_import);
 PyAPI_FUNC(PyObject *) _PyLazyImport_New(
     struct _PyInterpreterFrame *frame, PyObject *import_func, PyObject *from, PyObject *attr);
+PyAPI_FUNC(void) _PyLazyImport_BindGlobal(
+    PyThreadState *tstate,
+    PyObject *lazy_import,
+    PyObject *globals,
+    PyObject *name);
+PyAPI_FUNC(int) _PyLazyImport_CommitIfCurrent(
+    PyThreadState *tstate,
+    PyObject *lazy_import,
+    PyObject *globals,
+    PyObject *name,
+    PyObject *resolved);
+PyAPI_FUNC(int) _PyLazyImport_CommitStoredIfCurrent(
+    PyThreadState *tstate,
+    PyObject *lazy_import,
+    PyObject *resolved);
 
 #ifdef __cplusplus
 }

Lib/test/test_lazy_import/__init__.py

@@ -277,6 +277,273 @@ def test_lazy_import_type_attributes_accessible(self):
         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_resolve_replaces_original_global(self):
+        code = textwrap.dedent("""
+            import builtins
+            import types
+
+            real_import = builtins.__import__
+            calls = []
+
+            lazy import target_module as target
+
+            def custom_import(name, globals=None, locals=None, fromlist=None,
+                              level=0):
+                if name == "target_module":
+                    index = len(calls) + 1
+                    calls.append((index, name, globals.get("__name__"), fromlist))
+                    module = types.ModuleType(name)
+                    module.VALUE = f"value-{index}"
+                    return module
+                return real_import(name, globals, locals, fromlist, level)
+
+            builtins.__import__ = custom_import
+            try:
+                def resolve_from_dict():
+                    lazy_obj = globals()["target"]
+                    resolved = lazy_obj.resolve()
+                    assert resolved.VALUE == "value-1"
+                    assert globals()["target"] is resolved
+                    assert target.VALUE == "value-1"
+
+                resolve_from_dict()
+                assert len(calls) == 1, calls
+            finally:
+                builtins.__import__ = real_import
+        """)
+        assert_python_ok("-c", code)
+
+    @support.requires_subprocess()
+    def test_resolve_is_cached_on_stale_proxy(self):
+        code = textwrap.dedent("""
+            import builtins
+            import types
+
+            real_import = builtins.__import__
+            calls = []
+
+            lazy import target_module as target
+
+            def custom_import(name, globals=None, locals=None, fromlist=None,
+                              level=0):
+                if name == "target_module":
+                    calls.append(name)
+                    module = types.ModuleType(name)
+                    module.VALUE = len(calls)
+                    return module
+                return real_import(name, globals, locals, fromlist, level)
+
+            builtins.__import__ = custom_import
+            try:
+                def resolve_twice():
+                    lazy_obj = globals()["target"]
+                    first = lazy_obj.resolve()
+                    second = lazy_obj.resolve()
+                    assert first is second
+                    assert globals()["target"] is first
+
+                resolve_twice()
+                assert len(calls) == 1, calls
+            finally:
+                builtins.__import__ = real_import
+        """)
+        assert_python_ok("-c", code)
+
+    @support.requires_subprocess()
+    def test_resolve_after_automatic_reification_uses_cache(self):
+        code = textwrap.dedent("""
+            import builtins
+            import types
+
+            real_import = builtins.__import__
+            calls = []
+
+            lazy import target_module as target
+            holder = [globals()["target"]]
+
+            def custom_import(name, globals=None, locals=None, fromlist=None,
+                              level=0):
+                if name == "target_module":
+                    calls.append(name)
+                    module = types.ModuleType(name)
+                    module.VALUE = f"value-{len(calls)}"
+                    return module
+                return real_import(name, globals, locals, fromlist, level)
+
+            builtins.__import__ = custom_import
+            try:
+                first = target.VALUE
+                second = holder[0].resolve()
+                assert second.VALUE == first == "value-1"
+                assert globals()["target"] is second
+                assert len(calls) == 1, calls
+            finally:
+                builtins.__import__ = real_import
+        """)
+        assert_python_ok("-c", code)
+
+    @support.requires_subprocess()
+    def test_resolve_uses_actual_alias_binding(self):
+        code = textwrap.dedent("""
+            import builtins
+            import types
+
+            real_import = builtins.__import__
+            calls = []
+
+            lazy import target_module as target
+            lazy from target_module import VALUE as alias
+
+            def custom_import(name, globals=None, locals=None, fromlist=None,
+                              level=0):
+                if name == "target_module":
+                    index = len(calls) + 1
+                    calls.append((name, fromlist))
+                    module = types.ModuleType(name)
+                    module.VALUE = f"value-{index}"
+                    return module
+                return real_import(name, globals, locals, fromlist, level)
+
+            builtins.__import__ = custom_import
+            try:
+                def resolve_aliases():
+                    target_lazy = globals()["target"]
+                    target_resolved = target_lazy.resolve()
+                    assert globals()["target"] is target_resolved
+                    assert globals().get("target_module") is not target_resolved
+
+                    alias_lazy = globals()["alias"]
+                    alias_resolved = alias_lazy.resolve()
+                    assert globals()["alias"] is alias_resolved
+                    assert alias_resolved == "value-2"
+
+                resolve_aliases()
+                assert len(calls) == 2, calls
+            finally:
+                builtins.__import__ = real_import
+        """)
+        assert_python_ok("-c", code)
+
+    @support.requires_subprocess()
+    def test_resolve_does_not_clobber_rebound_or_deleted_global(self):
+        code = textwrap.dedent("""
+            import builtins
+            import types
+
+            real_import = builtins.__import__
+            calls = []
+
+            lazy import target_module as rebound
+            lazy import target_module as deleted
+
+            def custom_import(name, globals=None, locals=None, fromlist=None,
+                              level=0):
+                if name == "target_module":
+                    calls.append(name)
+                    module = types.ModuleType(name)
+                    module.VALUE = len(calls)
+                    return module
+                return real_import(name, globals, locals, fromlist, level)
+
+            builtins.__import__ = custom_import
+            try:
+                def resolve_after_changes():
+                    rebound_lazy = globals()["rebound"]
+                    deleted_lazy = globals()["deleted"]
+                    sentinel = object()
+                    globals()["rebound"] = sentinel
+                    del globals()["deleted"]
+
+                    rebound_resolved = rebound_lazy.resolve()
+                    deleted_resolved = deleted_lazy.resolve()
+
+                    assert globals()["rebound"] is sentinel
+                    assert "deleted" not in globals()
+                    assert rebound_resolved.VALUE == 1
+                    assert deleted_resolved.VALUE == 2
+
+                resolve_after_changes()
+                assert len(calls) == 2, calls
+            finally:
+                builtins.__import__ = real_import
+        """)
+        assert_python_ok("-c", code)
+
+    @support.requires_subprocess()
+    def test_resolve_failure_preserves_lazy_global_for_retry(self):
+        code = textwrap.dedent("""
+            import builtins
+            import types
+
+            real_import = builtins.__import__
+            attempts = 0
+
+            lazy import target_module as target
+
+            def custom_import(name, globals=None, locals=None, fromlist=None,
+                              level=0):
+                global attempts
+                if name == "target_module":
+                    attempts += 1
+                    if attempts == 1:
+                        raise ImportError("not yet")
+                    module = types.ModuleType(name)
+                    module.VALUE = "ok"
+                    return module
+                return real_import(name, globals, locals, fromlist, level)
+
+            builtins.__import__ = custom_import
+            try:
+                def retry_resolve():
+                    lazy_obj = globals()["target"]
+                    try:
+                        lazy_obj.resolve()
+                    except ImportError:
+                        pass
+                    else:
+                        raise AssertionError("first resolve should fail")
+                    assert globals()["target"] is lazy_obj
+
+                    resolved = lazy_obj.resolve()
+                    assert globals()["target"] is resolved
+                    assert resolved.VALUE == "ok"
+
+                retry_resolve()
+                assert attempts == 2, attempts
+            finally:
+                builtins.__import__ = real_import
+        """)
+        assert_python_ok("-c", code)
+
+    @support.requires_subprocess()
+    def test_automatic_reification_does_not_clobber_changed_global(self):
+        code = textwrap.dedent("""
+            import builtins
+            import types
+
+            real_import = builtins.__import__
+
+            lazy import target_module as target
+
+            def custom_import(name, globals=None, locals=None, fromlist=None,
+                              level=0):
+                if name == "target_module":
+                    globals["target"] = "user value"
+                    module = types.ModuleType(name)
+                    module.VALUE = "resolved"
+                    return module
+                return real_import(name, globals, locals, fromlist, level)
+
+            builtins.__import__ = custom_import
+            try:
+                assert target.VALUE == "resolved"
+                assert globals()["target"] == "user value"
+            finally:
+                builtins.__import__ = real_import
+        """)
+        assert_python_ok("-c", code)
+
 
 class SyntaxRestrictionTests(LazyImportTestCase):
     """Tests for syntax restrictions on lazy imports."""

Misc/NEWS.d/next/Core_and_Builtins/2026-06-26-23-58-00.gh-issue-152298.5uT4tW.rst

@@ -0,0 +1,3 @@
+Make :meth:`types.LazyImportType.resolve` cache successful resolutions and
+replace the original module global only when it still references the same lazy
+import proxy.