mirror of
https://github.com/python/cpython.git
synced 2025-12-08 06:10:17 +00:00
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"
This commit is contained in:
parent
69f54ce452
commit
656a64b37f
3 changed files with 59 additions and 22 deletions
|
|
@ -208,12 +208,8 @@ def _write_atomic(path, data, mode=0o666):
|
||||||
try:
|
try:
|
||||||
# We first write data to a temporary file, and then use os.replace() to
|
# We first write data to a temporary file, and then use os.replace() to
|
||||||
# perform an atomic rename.
|
# perform an atomic rename.
|
||||||
with _io.FileIO(fd, 'wb') as file:
|
with _io.open(fd, 'wb') as file:
|
||||||
bytes_written = file.write(data)
|
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")
|
|
||||||
_os.replace(path_tmp, path)
|
_os.replace(path_tmp, path)
|
||||||
except OSError:
|
except OSError:
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
|
|
@ -788,31 +788,70 @@ def test_complete_multi_phase_init_module(self):
|
||||||
self.run_with_own_gil(script)
|
self.run_with_own_gil(script)
|
||||||
|
|
||||||
|
|
||||||
class MiscTests(unittest.TestCase):
|
class PatchAtomicWrites:
|
||||||
def test_atomic_write_should_notice_incomplete_writes(self):
|
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
|
import _pyio
|
||||||
|
|
||||||
oldwrite = os.write
|
oldwrite = os.write
|
||||||
seen_write = False
|
|
||||||
|
|
||||||
truncate_at_length = 100
|
|
||||||
|
|
||||||
# Emulate an os.write that only writes partial data.
|
# Emulate an os.write that only writes partial data.
|
||||||
def write(fd, data):
|
def write(fd, data):
|
||||||
nonlocal seen_write
|
if self.seen_write and self.never_complete:
|
||||||
seen_write = True
|
return None
|
||||||
return oldwrite(fd, data[:truncate_at_length])
|
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
|
# Need to patch _io to be _pyio, so that io.FileIO is affected by the
|
||||||
# os.write patch.
|
# os.write patch.
|
||||||
with (support.swap_attr(_bootstrap_external, '_io', _pyio),
|
self.children = [
|
||||||
support.swap_attr(os, 'write', write)):
|
support.swap_attr(_bootstrap_external, '_io', _pyio),
|
||||||
with self.assertRaises(OSError):
|
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
|
# Make sure we write something longer than the point where we
|
||||||
# truncate.
|
# truncate.
|
||||||
content = b'x' * (truncate_at_length * 2)
|
content = b'x' * (truncate_at_length * 2)
|
||||||
_bootstrap_external._write_atomic(os_helper.TESTFN, content)
|
_bootstrap_external._write_atomic(os_helper.TESTFN, content)
|
||||||
assert seen_write
|
self.assertTrue(cm.seen_write)
|
||||||
|
|
||||||
with self.assertRaises(OSError):
|
with self.assertRaises(OSError):
|
||||||
os.stat(support.os_helper.TESTFN) # Check that the file did not get written.
|
os.stat(support.os_helper.TESTFN) # Check that the file did not get written.
|
||||||
|
|
|
||||||
|
|
@ -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.
|
||||||
Loading…
Add table
Add a link
Reference in a new issue