https://github.com/python/cpython/commit/656a64b37f817cc8fe36ee17f332100482185cce
commit: 656a64b37f817cc8fe36ee17f332100482185cce
branch: main
author: Stefano Rivera <[email protected]>
committer: gpshead <[email protected]>
date: 2025-11-27T19:17:59Z
summary:
gh-141930: Use the regular IO stack to write .pyc files for a better error
message on failure (GH-141931)
* Use open() to write the bytecode
* Convert to unittest style asserts
* Tweak news, thanks @vstinner
* Tidy
* reword NEWS, avoid word "retried"
files:
A
Misc/NEWS.d/next/Core_and_Builtins/2025-11-24-21-09-30.gh-issue-141930.hIIzSd.rst
M Lib/importlib/_bootstrap_external.py
M Lib/test/test_importlib/test_util.py
diff --git a/Lib/importlib/_bootstrap_external.py
b/Lib/importlib/_bootstrap_external.py
index 192c0261408ead..2f9307cba4f086 100644
--- 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 a77ce234deec58..0adab8d14e0452 100644
--- a/Lib/test/test_importlib/test_util.py
+++ b/Lib/test/test_importlib/test_util.py
@@ -788,31 +788,70 @@ def test_complete_multi_phase_init_module(self):
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
index 00000000000000..06a12f98224e88
--- /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.
_______________________________________________
Python-checkins mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3//lists/python-checkins.python.org
Member address: [email protected]
