gh-141930: Use the regular IO stack to write .pyc files for a better error message...

* Use open() to write the bytecode
* Convert to unittest style asserts
* Tweak news, thanks @vstinner
* Tidy
* reword NEWS, avoid word "retried"

diff --git a/Lib/importlib/_bootstrap_external.py b/Lib/importlib/_bootstrap_external.py
index 192c0261408ead9df27c98b3035e1ddc442033a2..2f9307cba4f086e10bdc6938a2f162083bb8b713 100644 (file)
--- a/Lib/importlib/_bootstrap_external.py
+++ b/Lib/importlib/_bootstrap_external.py
@@ -208,12 +208,8 @@ def _write_atomic(path, data, mode=0o666):
     try:
         # We first write data to a temporary file, and then use os.replace() to
         # perform an atomic rename.
-        with _io.FileIO(fd, 'wb') as file:
-            bytes_written = file.write(data)
-        if bytes_written != len(data):
-            # Raise an OSError so the 'except' below cleans up the partially
-            # written file.
-            raise OSError("os.write() didn't write the full pyc file")
+        with _io.open(fd, 'wb') as file:
+            file.write(data)
         _os.replace(path_tmp, path)
     except OSError:
         try:

diff --git a/Lib/test/test_importlib/test_util.py b/Lib/test/test_importlib/test_util.py
index a77ce234deec5897f3a0647a70d4b954ba24e976..0adab8d14e04523a59d606b4300406440164e761 100644 (file)
--- a/Lib/test/test_importlib/test_util.py
+++ b/Lib/test/test_importlib/test_util.py
@@ -788,31 +788,70 @@ class IncompatibleExtensionModuleRestrictionsTests(unittest.TestCase):
             self.run_with_own_gil(script)
 
 
-class MiscTests(unittest.TestCase):
-    def test_atomic_write_should_notice_incomplete_writes(self):
+class PatchAtomicWrites:
+    def __init__(self, truncate_at_length, never_complete=False):
+        self.truncate_at_length = truncate_at_length
+        self.never_complete = never_complete
+        self.seen_write = False
+        self._children = []
+
+    def __enter__(self):
         import _pyio
 
         oldwrite = os.write
-        seen_write = False
-
-        truncate_at_length = 100
 
         # Emulate an os.write that only writes partial data.
         def write(fd, data):
-            nonlocal seen_write
-            seen_write = True
-            return oldwrite(fd, data[:truncate_at_length])
+            if self.seen_write and self.never_complete:
+                return None
+            self.seen_write = True
+            return oldwrite(fd, data[:self.truncate_at_length])
 
         # Need to patch _io to be _pyio, so that io.FileIO is affected by the
         # os.write patch.
-        with (support.swap_attr(_bootstrap_external, '_io', _pyio),
-              support.swap_attr(os, 'write', write)):
-            with self.assertRaises(OSError):
-                # Make sure we write something longer than the point where we
-                # truncate.
-                content = b'x' * (truncate_at_length * 2)
-                _bootstrap_external._write_atomic(os_helper.TESTFN, content)
-        assert seen_write
+        self.children = [
+            support.swap_attr(_bootstrap_external, '_io', _pyio),
+            support.swap_attr(os, 'write', write)
+        ]
+        for child in self.children:
+            child.__enter__()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        for child in self.children:
+            child.__exit__(exc_type, exc_val, exc_tb)
+
+
+class MiscTests(unittest.TestCase):
+
+    def test_atomic_write_retries_incomplete_writes(self):
+        truncate_at_length = 100
+        length = truncate_at_length * 2
+
+        with PatchAtomicWrites(truncate_at_length=truncate_at_length) as cm:
+            # Make sure we write something longer than the point where we
+            # truncate.
+            content = b'x' * length
+            _bootstrap_external._write_atomic(os_helper.TESTFN, content)
+        self.assertTrue(cm.seen_write)
+
+        self.assertEqual(os.stat(support.os_helper.TESTFN).st_size, length)
+        os.unlink(support.os_helper.TESTFN)
+
+    def test_atomic_write_errors_if_unable_to_complete(self):
+        truncate_at_length = 100
+
+        with (
+            PatchAtomicWrites(
+                truncate_at_length=truncate_at_length, never_complete=True,
+            ) as cm,
+            self.assertRaises(OSError)
+        ):
+            # Make sure we write something longer than the point where we
+            # truncate.
+            content = b'x' * (truncate_at_length * 2)
+            _bootstrap_external._write_atomic(os_helper.TESTFN, content)
+        self.assertTrue(cm.seen_write)
 
         with self.assertRaises(OSError):
             os.stat(support.os_helper.TESTFN) # Check that the file did not get written.

diff --git a/Misc/NEWS.d/next/Core_and_Builtins/2025-11-24-21-09-30.gh-issue-141930.hIIzSd.rst b/Misc/NEWS.d/next/Core_and_Builtins/2025-11-24-21-09-30.gh-issue-141930.hIIzSd.rst
new file mode 100644 (file)
index 0000000..06a12f9
--- /dev/null
+++ b/Misc/NEWS.d/next/Core_and_Builtins/2025-11-24-21-09-30.gh-issue-141930.hIIzSd.rst
@@ -0,0 +1,2 @@
+When importing a module, use Python's regular file object to ensure that
+writes to ``.pyc`` files are complete or an appropriate error is raised.