Skip to content
Merged
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
16 changes: 16 additions & 0 deletions tests/test_file_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import hashlib
import os
import stat
import tempfile
from pathlib import Path

Expand Down Expand Up @@ -176,6 +177,21 @@ def test_custom_encoding(self, tmp_dir):
fu.atomic_write_text(p, "café", encoding="latin-1")
assert p.read_bytes() == "café".encode("latin-1")

@pytest.mark.skipif(os.name != "posix", reason="requires POSIX permission bits")
def test_new_file_uses_0644_mode(self, tmp_dir):
p = Path(tmp_dir) / "out.txt"
fu.atomic_write_text(p, "new")
assert stat.S_IMODE(p.stat().st_mode) == 0o644

@pytest.mark.skipif(os.name != "posix", reason="requires POSIX permission bits")
@pytest.mark.parametrize("mode", [0o600, 0o640, 0o755])
def test_overwrite_preserves_existing_mode(self, tmp_dir, mode):
p = Path(tmp_dir) / "out.txt"
p.write_text("old", encoding="utf-8")
p.chmod(mode)
fu.atomic_write_text(p, "new")
assert stat.S_IMODE(p.stat().st_mode) == mode


class TestBackupNameStem:
def test_contains_name_and_hash(self, tmp_dir):
Expand Down
10 changes: 10 additions & 0 deletions tests/test_multi_edit.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for multi_edit."""

import os
import stat
import tempfile
from pathlib import Path

Expand All @@ -27,6 +28,15 @@ def test_single_edit(self, tmp_file):
assert r["ok"] is True
assert Path(tmp_file).read_text(encoding="utf-8") == "x = 2\n"

@pytest.mark.skipif(os.name != "posix", reason="requires POSIX permission bits")
def test_preserves_existing_mode(self, tmp_file):
path = Path(tmp_file)
path.write_text("x = 1\n", encoding="utf-8")
path.chmod(0o640)
r = run([{"file": tmp_file, "old": "x = 1", "new": "x = 2"}])
assert r["ok"] is True
assert stat.S_IMODE(path.stat().st_mode) == 0o640

def test_replace_all(self, tmp_file):
Path(tmp_file).write_text("x = 1\nx = 1\n", encoding="utf-8")
r = run([{"file": tmp_file, "old": "x = 1", "new": "x = 2", "replace_all": True}])
Expand Down
14 changes: 14 additions & 0 deletions tools/_file_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -285,10 +285,21 @@ def human_size(n: int) -> str:
return s.replace(".0PB", "PB")


def _atomic_target_mode(path: str | Path) -> int | None:
"""返回原子替换后的 POSIX 权限位;非 POSIX 平台不做 mode 处理。"""
if os.name != "posix":
return None
try:
return Path(path).stat().st_mode & 0o777
except FileNotFoundError:
return 0o644


def atomic_write_text(path: str | Path, content: str, encoding: str = "utf-8") -> None:
"""原子写入文本文件:先写同目录临时文件,再 os.replace 替换目标文件。

保留原始换行符(调用方需确保 content 中的换行符已是期望形式)。
POSIX 下保留现有文件权限位;新文件默认 0644。
"""
import os as _os
import tempfile as _tmp
Expand All @@ -299,6 +310,9 @@ def atomic_write_text(path: str | Path, content: str, encoding: str = "utf-8") -
try:
with _os.fdopen(fd, "w", encoding=encoding, newline="") as f:
f.write(content)
final_mode = _atomic_target_mode(target)
if final_mode is not None:
_os.chmod(tmp, final_mode)
_os.replace(str(tmp), str(target))
except Exception:
try:
Expand Down
4 changes: 4 additions & 0 deletions tools/multi_edit.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from pathlib import Path

from ._file_utils import SAFE_EDIT_MAX_SIZE, read_file_with_encoding, find_closest_line, align_whitespace, backup_name_stem, check_path_allowed, prune_backups, strip_line_number_prefixes
from ._file_utils import _atomic_target_mode
from .safe_edit import _backup_dir
from .syntax_check import check as syntax_check_file

Expand Down Expand Up @@ -266,6 +267,9 @@ def run(edits: list, syntax_check: bool = True) -> dict:
f.write(final_content)
tmp_paths[path] = Path(tmp_name)
for path, tmp_path in tmp_paths.items():
final_mode = _atomic_target_mode(path)
if final_mode is not None:
os.chmod(tmp_path, final_mode)
os.replace(str(tmp_path), str(path))
except Exception as exc:
rollback_errors = []
Expand Down
Loading