refactor(file-ops): fold simplify-pass findings

- write_file: encode content once, share bytes between bytes_written and
  the sha256 verification (drops a second full-content encode per write)
- patch_parser: replace the except-TypeError retry around
  write_file(pre_content=...) with signature-based feature detection so a
  TypeError raised inside a capable implementation propagates instead of
  triggering a duplicate write; tests for both duck-typing contracts
- tests: real-ops V4A BOM round-trip + _file_has_bom disk-probe guard
  (the teknium1-review regression previously only covered by a fake)
- comment: document dirs_created's long-standing "parent ensured" meaning
This commit is contained in:
kshitijk4poor
2026-08-03 16:47:54 +05:30
committed by kshitij
parent fcae5ad49a
commit 8c19e29259
4 changed files with 126 additions and 14 deletions
+29
View File
@@ -308,6 +308,35 @@ class TestBomHandling:
raw = target.read_bytes()
assert raw == self.BOM.encode("utf-8") + b"import os, json\nimport sys\n"
def test_v4a_update_preserves_bom_real_ops(self, ops, tmp_path: Path):
# V4A UPDATE path against REAL ShellFileOperations. This is the one
# provider path whose pre_content is BOM-STRIPPED (read_file_raw
# strips before _apply_update forwards it), so it regresses if
# _file_has_bom ever trusts pre_content instead of probing disk.
# Regression for teknium1's review on PR #55661.
target = tmp_path / "bom_v4a.py"
target.write_bytes(self.BOM.encode("utf-8") + b"print('hello')\n")
patch = (
"*** Begin Patch\n"
f"*** Update File: {target}\n"
"@@\n"
"-print('hello')\n"
"+print('world')\n"
"*** End Patch"
)
res = ops.patch_v4a(patch)
assert res.success, res.error
raw = target.read_bytes()
assert raw.startswith(self.BOM.encode("utf-8")), "BOM lost on V4A update"
assert b"print('world')" in raw
def test_file_has_bom_ignores_stripped_pre_content(self, ops, tmp_path: Path):
# _file_has_bom must probe the DISK even when handed pre_content
# that (having been BOM-stripped upstream) claims there is no BOM.
target = tmp_path / "bom_probe.py"
target.write_bytes(self.BOM.encode("utf-8") + b"x = 1\n")
assert ops._file_has_bom(str(target), pre_content="x = 1\n") is True
if __name__ == "__main__":
pytest.main([__file__, "-v"])
+56
View File
@@ -692,6 +692,62 @@ class _DictFileOps:
return SimpleNamespace(error=None)
class TestDuckTypedWriteFileCompat:
"""V4A UPDATE must work with basic write_file(path, content) impls.
apply_v4a_operations is duck-typed (file_ops: Any); external callers may
only implement the two-argument contract. The signature-based feature
detection must route them to the 2-arg call — and must NOT swallow a
TypeError raised INSIDE a pre_content-capable write_file (which would
trigger a duplicate write).
"""
PATCH = (
"*** Begin Patch\n"
"*** Update File: f.py\n"
"@@\n"
"-x = 1\n"
"+x = 2\n"
"*** End Patch"
)
def test_two_arg_write_file_still_supported(self):
calls = []
class BasicOps(_DictFileOps):
def write_file(self, path, content): # no pre_content
calls.append(path)
self.files[path] = content
return SimpleNamespace(error=None)
ops, err = parse_v4a_patch(self.PATCH)
assert err is None
fo = BasicOps({"f.py": "x = 1\n"})
result = apply_v4a_operations(ops, fo)
assert result.success is True, getattr(result, "error", None)
assert fo.files["f.py"] == "x = 2\n"
assert calls == ["f.py"] # exactly one write, no double invocation
def test_internal_typeerror_not_silently_retried(self):
# A TypeError raised INSIDE a pre_content-capable write_file must not
# trigger a second 2-arg write. (The op-loop's blanket except turns it
# into a failed result — the key contract is: ONE call, error surfaced.)
calls = []
class ExplodingOps(_DictFileOps):
def write_file(self, path, content, pre_content=None):
calls.append(path)
raise TypeError("bug inside a pre_content-capable impl")
ops, err = parse_v4a_patch(self.PATCH)
assert err is None
fo = ExplodingOps({"f.py": "x = 1\n"})
result = apply_v4a_operations(ops, fo)
assert result.success is False
assert "bug inside" in result.error
assert calls == ["f.py"] # not silently retried with 2 args
class TestMoveThenUpdateSameFile:
"""A rename-then-edit patch must validate and apply (was rejected).
+16 -11
View File
@@ -1564,10 +1564,12 @@ class ShellFileOperations(FileOperations):
self._snapshot_lsp_baseline(path)
# Write atomically. ``mkdir -p`` is folded into _atomic_write
# (one fewer subprocess vs. a separate mkdir call). Report
# dirs_created as True when the parent wasn't obviously present;
# we don't stat to avoid an extra syscall — if the mkdir succeeds
# or was already there, _atomic_write handles it.
# (one fewer subprocess vs. a separate mkdir call).
# ``dirs_created`` has always meant "parent dirs ensured" —
# ``mkdir -p`` exits 0 even when the dirs pre-exist, so the old
# separate-mkdir code reported True in exactly the same cases.
# A mkdir failure now surfaces as the atomic-write error return
# below, before this field is ever emitted.
parent = os.path.dirname(path)
dirs_created = bool(parent)
@@ -1592,12 +1594,15 @@ class ShellFileOperations(FileOperations):
return WriteResult(error=f"Failed to write file: {write_result.stdout}")
# Get bytes written — compute from the content we just wrote
# (len(content.encode('utf-8')) matches wc -c for UTF-8) instead
# of spawning a ``wc -c`` subprocess. ``surrogatepass`` mirrors the
# sha256 verification block below: content that flowed through a
# surrogateescape decode (backend output via patch_replace) may
# carry lone surrogates a strict encode would reject.
bytes_written = len(content.encode('utf-8', 'surrogatepass'))
# (len of the UTF-8 encoding matches wc -c) instead of spawning a
# ``wc -c`` subprocess. ``surrogatepass`` matches the sha256
# verification below: content that flowed through a surrogateescape
# decode (backend output via patch_replace) may carry lone
# surrogates a strict encode would reject. Encode ONCE and share
# the bytes with the sha256 block — a second full encode of a
# multi-MB file is measurable.
content_bytes = content.encode('utf-8', 'surrogatepass')
bytes_written = len(content_bytes)
# Post-write content verification (cheap, one shell call): compare
# the on-disk sha256 to the intended content's hash. Production
@@ -1612,7 +1617,7 @@ class ShellFileOperations(FileOperations):
hash_result = self._exec(hash_cmd)
if hash_result.exit_code == 0 and hash_result.stdout.strip():
disk_sha = hash_result.stdout.strip().split()[0]
expected_sha = hashlib.sha256(content.encode("utf-8", "surrogatepass")).hexdigest()
expected_sha = hashlib.sha256(content_bytes).hexdigest()
content_verified = disk_sha == expected_sha
if not content_verified:
return WriteResult(
+25 -3
View File
@@ -29,6 +29,7 @@ Usage:
"""
import difflib
import inspect
import re
from dataclasses import dataclass, field
from typing import List, Optional, Tuple, Any
@@ -517,6 +518,24 @@ def apply_v4a_operations(operations: List[PatchOperation],
)
def _write_file_accepts_pre_content(file_ops: Any) -> bool:
"""True when ``file_ops.write_file`` accepts a ``pre_content`` kwarg.
Decided from the signature (not by catching TypeError around the call)
so a TypeError raised *inside* a capable ``write_file`` propagates
instead of triggering a second, duplicate write. Unintrospectable
callables (some C-implemented ones) conservatively get the basic
two-argument form.
"""
try:
params = inspect.signature(file_ops.write_file).parameters
except (TypeError, ValueError):
return False
return "pre_content" in params or any(
p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()
)
def _apply_add(op: PatchOperation, file_ops: Any) -> Tuple[bool, str, Optional[str], Optional[dict]]:
"""Apply an add file operation.
@@ -686,11 +705,14 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> Tuple[bool, str, Optiona
# a redundant cat subprocess inside write_file. Fall back to the
# two-argument form when the file_ops implementation doesn't accept
# ``pre_content`` (duck-typed callers that only implement the basic
# ``write_file(path, content)`` contract).
try:
# ``write_file(path, content)`` contract). Feature-detect via the
# signature instead of catching TypeError around the call: a TypeError
# raised *inside* a pre_content-capable write_file must propagate, not
# trigger a second (double) write.
if _write_file_accepts_pre_content(file_ops):
write_result = file_ops.write_file(op.file_path, new_content,
pre_content=current_content)
except TypeError:
else:
write_result = file_ops.write_file(op.file_path, new_content)
if write_result.error:
return False, write_result.error, None, None