Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 32 additions & 3 deletions codespell_lib/_codespell.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,9 @@
import os
import re
import shlex
import stat
import sys
import tempfile
import textwrap
from collections.abc import Iterable, Sequence
from re import Match, Pattern
Expand Down Expand Up @@ -1309,13 +1311,40 @@ def parse_file(
f" {cfilename}:{cline}: {cwrongword} ==> {crightword}",
file=sys.stderr,
)
with open(filename, "w", encoding=encoding, newline="") as f:
for _, _, lines in fragments:
f.writelines(lines)
_write_file_atomically(filename, encoding, fragments)

return bad_count


def _write_file_atomically(
filename: str,
encoding: str,
fragments: Iterable[tuple[Any, Any, list[str]]],
) -> None:
"""Replace *filename* with a same-directory tempfile so a failed write
cannot leave the original file truncated to zero bytes.
"""
directory = os.path.dirname(os.path.abspath(filename))
try:
original_mode = stat.S_IMODE(os.stat(filename).st_mode)
except OSError:
original_mode = None
fd, tmp_path = tempfile.mkstemp(prefix=".codespell-", suffix=".tmp", dir=directory)
try:
with os.fdopen(fd, "w", encoding=encoding, newline="") as tmp:
for _, _, lines in fragments:
tmp.writelines(lines)
if original_mode is not None:
os.chmod(tmp_path, original_mode)
os.replace(tmp_path, filename)
except Exception:
try:
os.unlink(tmp_path)
except OSError:
pass
raise


def flatten_clean_comma_separated_arguments(
arguments: Iterable[str],
) -> list[str]:
Expand Down
35 changes: 35 additions & 0 deletions codespell_lib/tests/test_basic.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import os
import os.path as op
import re
import stat
import subprocess
import sys
from collections.abc import Generator
Expand Down Expand Up @@ -175,6 +176,40 @@ def test_basic(
assert cs.main(tmp_path) == 0


def test_write_changes_keeps_original_if_replace_fails(
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""An interrupted --write-changes must not empty the original file."""
fname = tmp_path / "a.txt"
original = "this file has an abandonned word\n"
fname.write_text(original)

def fail_replace(src: str, dst: str) -> None:
raise InterruptedError

monkeypatch.setattr(os, "replace", fail_replace)
with pytest.raises(InterruptedError):
cs.main("-q", "16", "-w", fname)
assert fname.read_text() == original
assert list(tmp_path.glob(".codespell-*.tmp")) == []


@pytest.mark.skipif(sys.platform == "win32", reason="POSIX permission bits")
def test_write_changes_preserves_file_mode(
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
) -> None:
"""Atomic --write-changes must not reset the original file mode."""
fname = tmp_path / "a.txt"
fname.write_text("this file has an abandonned word\n")
fname.chmod(0o640)
assert cs.main("-q", "16", "-w", fname) == 0
assert fname.read_text() == "this file has an abandoned word\n"
assert stat.S_IMODE(fname.stat().st_mode) == 0o640


def test_write_changes_lists_changes(
tmp_path: Path,
capsys: pytest.CaptureFixture[str],
Expand Down