From 8c19e29259db433c5633214572648b4cb7b243c3 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Mon, 3 Aug 2026 16:47:54 +0530 Subject: [PATCH] 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 --- tests/tools/test_file_write_safety.py | 29 ++++++++++++++ tests/tools/test_patch_parser.py | 56 +++++++++++++++++++++++++++ tools/file_operations.py | 27 +++++++------ tools/patch_parser.py | 28 ++++++++++++-- 4 files changed, 126 insertions(+), 14 deletions(-) diff --git a/tests/tools/test_file_write_safety.py b/tests/tools/test_file_write_safety.py index d59dce7b21..ee1296e495 100644 --- a/tests/tools/test_file_write_safety.py +++ b/tests/tools/test_file_write_safety.py @@ -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"]) diff --git a/tests/tools/test_patch_parser.py b/tests/tools/test_patch_parser.py index 3487edee39..ea6c56257c 100644 --- a/tests/tools/test_patch_parser.py +++ b/tests/tools/test_patch_parser.py @@ -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). diff --git a/tools/file_operations.py b/tools/file_operations.py index bd4a5f95de..d22d12bb8c 100644 --- a/tools/file_operations.py +++ b/tools/file_operations.py @@ -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( diff --git a/tools/patch_parser.py b/tools/patch_parser.py index b0b3458867..37412336ee 100644 --- a/tools/patch_parser.py +++ b/tools/patch_parser.py @@ -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