841 lines
31 KiB
Python
841 lines
31 KiB
Python
"""
|
|
Tests for scripts/patch_applier.py — the deterministic file-patch applier.
|
|
|
|
Safety net covering:
|
|
- each kind's transformation on a fixture
|
|
- multi-entry-one-file descending-order correctness (two delete-ranges)
|
|
- invariant #8: content changed since check → skip-file, continue other files
|
|
- incompatible-ops detection: move-to-archive + other op → whole file skipped;
|
|
overlapping anchors → skipped
|
|
- insert-frontmatter idempotency + key-conflict-preserve
|
|
- move-to-archive via real `git mv` in a tmp git repo
|
|
- replace-text no-op when match absent
|
|
- unknown kind rejected
|
|
- exit codes for all-success / partial / usage-error (via CLI main())
|
|
|
|
sys.path is patched here (no shared conftest) matching repo test style.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Path bootstrap
|
|
# ---------------------------------------------------------------------------
|
|
_SCRIPTS_DIR = Path(__file__).parent.parent / "scripts"
|
|
if str(_SCRIPTS_DIR) not in sys.path:
|
|
sys.path.insert(0, str(_SCRIPTS_DIR))
|
|
|
|
from patch_applier import ( # noqa: E402
|
|
PatchApplier,
|
|
RealFileSystem,
|
|
_delete_lines,
|
|
_insert_frontmatter_key,
|
|
_parse_frontmatter,
|
|
_replace_in_span,
|
|
_ranges_overlap,
|
|
main as applier_main,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test doubles
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class MemFS:
|
|
"""In-memory filesystem for hermetic tests (no real disk I/O)."""
|
|
|
|
def __init__(self, files: dict[str, bytes] | None = None) -> None:
|
|
self._files: dict[str, bytes] = dict(files or {})
|
|
self.writes: dict[str, bytes] = {}
|
|
self.mkdirs: list[str] = []
|
|
|
|
def read_bytes(self, path: Path) -> bytes:
|
|
key = str(path)
|
|
if key not in self._files:
|
|
raise FileNotFoundError(f"MemFS: {key} not found")
|
|
return self._files[key]
|
|
|
|
def write_bytes(self, path: Path, data: bytes) -> None:
|
|
key = str(path)
|
|
self._files[key] = data
|
|
self.writes[key] = data
|
|
|
|
def exists(self, path: Path) -> bool:
|
|
return str(path) in self._files
|
|
|
|
def mkdir(self, path: Path) -> None:
|
|
self.mkdirs.append(str(path))
|
|
|
|
|
|
class RecordingGit:
|
|
"""Records git mv calls without touching disk."""
|
|
|
|
def __init__(self, fail: bool = False) -> None:
|
|
self.calls: list[tuple[str, str]] = []
|
|
self._fail = fail
|
|
|
|
def mv(self, src: Path, dest: Path, cwd: Path) -> None:
|
|
if self._fail:
|
|
raise RuntimeError("git mv simulated failure")
|
|
self.calls.append((str(src), str(dest)))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _sha256(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def _make_file(content: str) -> tuple[bytes, str]:
|
|
"""Return (bytes, sha256_hex) for a text content."""
|
|
b = content.encode("utf-8")
|
|
return b, _sha256(b)
|
|
|
|
|
|
def _entry(
|
|
path: str,
|
|
kind: str,
|
|
extra: dict | None = None,
|
|
sha: str = "",
|
|
start: int = 1,
|
|
end: int = 1,
|
|
) -> dict:
|
|
"""Build a minimal deterministic entry dict."""
|
|
from validate_report import KIND_TABLE
|
|
spec = KIND_TABLE[kind]
|
|
exact_edit: dict[str, Any] = {"kind": kind}
|
|
if spec.has_anchor:
|
|
exact_edit["anchor"] = {"start_line": start, "end_line": end}
|
|
exact_edit["expected_sha256"] = sha
|
|
if extra:
|
|
exact_edit.update(extra)
|
|
return {
|
|
"path": path,
|
|
"op": f"test op {kind}",
|
|
"op_type": "deterministic",
|
|
"is_destructive": spec.is_destructive,
|
|
"is_reversible": spec.is_reversible,
|
|
"exact_edit": exact_edit,
|
|
}
|
|
|
|
|
|
def _applier(root: Path, fs=None, git=None) -> PatchApplier:
|
|
return PatchApplier(project_root=root, fs=fs, git=git)
|
|
|
|
|
|
def _init_git_repo(tmp_path: Path) -> Path:
|
|
"""Create a real git repo in tmp_path; configure minimal identity."""
|
|
subprocess.run(["git", "init", str(tmp_path)], check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "config", "user.email", "test@test.com"],
|
|
cwd=str(tmp_path), check=True, capture_output=True,
|
|
)
|
|
subprocess.run(
|
|
["git", "config", "user.name", "Test"],
|
|
cwd=str(tmp_path), check=True, capture_output=True,
|
|
)
|
|
return tmp_path
|
|
|
|
|
|
# ===========================================================================
|
|
# Unit: pure helper functions
|
|
# ===========================================================================
|
|
|
|
class TestHelpers:
|
|
|
|
def test_delete_lines_removes_inclusive_range(self):
|
|
lines = ["a\n", "b\n", "c\n", "d\n", "e\n"]
|
|
result = _delete_lines(lines, 2, 4)
|
|
assert result == ["a\n", "e\n"]
|
|
|
|
def test_delete_lines_whole_file(self):
|
|
lines = ["a\n", "b\n"]
|
|
assert _delete_lines(lines, 1, 2) == []
|
|
|
|
def test_delete_lines_single_line(self):
|
|
lines = ["a\n", "b\n", "c\n"]
|
|
assert _delete_lines(lines, 2, 2) == ["a\n", "c\n"]
|
|
|
|
def test_replace_in_span_finds_first_occurrence(self):
|
|
lines = ["foo bar\n", "foo baz\n", "other\n"]
|
|
out, found = _replace_in_span(lines, 1, 2, "foo", "qux")
|
|
assert found is True
|
|
assert out[0] == "qux bar\n" # first occurrence replaced
|
|
assert out[1] == "foo baz\n" # second unchanged (first only)
|
|
|
|
def test_replace_in_span_not_found(self):
|
|
lines = ["a\n", "b\n"]
|
|
out, found = _replace_in_span(lines, 1, 2, "missing", "x")
|
|
assert found is False
|
|
assert out == ["a\n", "b\n"]
|
|
|
|
def test_ranges_overlap_adjacent_not_overlapping(self):
|
|
assert not _ranges_overlap(1, 3, 4, 6)
|
|
|
|
def test_ranges_overlap_touching_end(self):
|
|
assert _ranges_overlap(1, 4, 4, 6)
|
|
|
|
def test_ranges_overlap_contained(self):
|
|
assert _ranges_overlap(1, 10, 3, 5)
|
|
|
|
def test_parse_frontmatter_present(self):
|
|
text = "---\nhygiene: frozen\nauthor: jared\n---\nbody\n"
|
|
kv, end = _parse_frontmatter(text)
|
|
assert kv == {"hygiene": "frozen", "author": "jared"}
|
|
assert end == 4 # line index after closing ---
|
|
|
|
def test_parse_frontmatter_absent(self):
|
|
kv, end = _parse_frontmatter("# heading\nbody\n")
|
|
assert kv == {}
|
|
assert end == 0
|
|
|
|
def test_insert_frontmatter_key_creates_block(self):
|
|
text = "# heading\nbody\n"
|
|
result = _insert_frontmatter_key(text, "hygiene", "frozen")
|
|
assert result.startswith("---\nhygiene: frozen\n---\n")
|
|
|
|
def test_insert_frontmatter_key_appends_to_existing(self):
|
|
text = "---\nauthor: jared\n---\nbody\n"
|
|
result = _insert_frontmatter_key(text, "hygiene", "frozen")
|
|
assert "hygiene: frozen" in result
|
|
assert "author: jared" in result
|
|
# closing --- must still appear after new key
|
|
lines = result.splitlines()
|
|
fm_end = lines.index("---", 1)
|
|
assert "hygiene: frozen" in lines[fm_end - 1]
|
|
|
|
|
|
# ===========================================================================
|
|
# Each kind's transformation
|
|
# ===========================================================================
|
|
|
|
class TestDeleteRange:
|
|
|
|
def test_removes_specified_lines(self, tmp_path):
|
|
content = "line1\nline2\nline3\nline4\nline5\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
applier = _applier(tmp_path, fs=fs)
|
|
|
|
entry = _entry("doc.md", "delete-range", sha=sha, start=2, end=4)
|
|
result = applier.apply([entry])
|
|
|
|
assert len(result["applied"]) == 1
|
|
assert result["applied"][0]["kind"] == "delete-range"
|
|
written = fs.writes[str(tmp_path / "doc.md")].decode()
|
|
assert written == "line1\nline5\n"
|
|
|
|
def test_records_staged_path(self, tmp_path):
|
|
content = "a\nb\nc\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "x.md"): file_bytes})
|
|
result = _applier(tmp_path, fs=fs).apply([
|
|
_entry("x.md", "delete-range", sha=sha, start=2, end=2)
|
|
])
|
|
assert "x.md" in result["staged_paths"]
|
|
|
|
|
|
class TestReplaceText:
|
|
|
|
def test_replaces_match_in_span(self, tmp_path):
|
|
content = "see old.md here\nother line\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
entry = _entry("doc.md", "replace-text", sha=sha, start=1, end=1,
|
|
extra={"match": "old.md", "replacement": "new.md"})
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "replace-text"
|
|
written = fs.writes[str(tmp_path / "doc.md")].decode()
|
|
assert "new.md" in written
|
|
assert "old.md" not in written
|
|
|
|
def test_match_absent_is_noop_success(self, tmp_path):
|
|
"""replace-text: match absent in span → no-op success, no write."""
|
|
content = "nothing relevant here\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
entry = _entry("doc.md", "replace-text", sha=sha, start=1, end=1,
|
|
extra={"match": "old.md", "replacement": "new.md"})
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "replace-text"
|
|
assert result["skipped"] == []
|
|
assert result["failed"] == []
|
|
# No write needed (content unchanged).
|
|
assert str(tmp_path / "doc.md") not in fs.writes
|
|
|
|
|
|
class TestDedupe:
|
|
|
|
def test_removes_duplicate_span(self, tmp_path):
|
|
content = "header\ndupe1\ndupe2\nfooter\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
entry = _entry("doc.md", "dedupe", sha=sha, start=2, end=3,
|
|
extra={"canonical_ref": "other.md#section"})
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "dedupe"
|
|
written = fs.writes[str(tmp_path / "doc.md")].decode()
|
|
assert "dupe1" not in written
|
|
assert "header" in written
|
|
assert "footer" in written
|
|
|
|
|
|
class TestInsertFrontmatter:
|
|
|
|
def test_inserts_key_into_existing_block(self, tmp_path):
|
|
content = "---\nauthor: jared\n---\nbody\n"
|
|
file_bytes, _ = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
entry = _entry("doc.md", "insert-frontmatter",
|
|
extra={"key": "hygiene", "value": "frozen"})
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "insert-frontmatter"
|
|
written = fs.writes[str(tmp_path / "doc.md")].decode()
|
|
assert "hygiene: frozen" in written
|
|
assert "author: jared" in written
|
|
|
|
def test_creates_frontmatter_block_when_absent(self, tmp_path):
|
|
content = "# heading\nbody\n"
|
|
file_bytes, _ = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
entry = _entry("doc.md", "insert-frontmatter",
|
|
extra={"key": "hygiene", "value": "frozen"})
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
written = fs.writes[str(tmp_path / "doc.md")].decode()
|
|
assert written.startswith("---\nhygiene: frozen\n---\n")
|
|
|
|
def test_idempotent_when_key_already_correct(self, tmp_path):
|
|
"""Key present with target value → success, no write."""
|
|
content = "---\nhygiene: frozen\n---\nbody\n"
|
|
file_bytes, _ = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
entry = _entry("doc.md", "insert-frontmatter",
|
|
extra={"key": "hygiene", "value": "frozen"})
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "insert-frontmatter"
|
|
assert result["skipped"] == []
|
|
# Idempotent → content unchanged → no write.
|
|
assert str(tmp_path / "doc.md") not in fs.writes
|
|
|
|
def test_key_conflict_preserves_existing(self, tmp_path):
|
|
"""Key present with DIFFERENT value → skipped, existing value untouched."""
|
|
content = "---\nhygiene: ignored\n---\nbody\n"
|
|
file_bytes, _ = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
entry = _entry("doc.md", "insert-frontmatter",
|
|
extra={"key": "hygiene", "value": "frozen"})
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "frontmatter-key-conflict"
|
|
# File must NOT have been modified.
|
|
assert str(tmp_path / "doc.md") not in fs.writes
|
|
|
|
|
|
class TestMoveToArchive:
|
|
|
|
def test_git_mv_called_with_correct_args(self, tmp_path):
|
|
content = "# old plan\nbody\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({
|
|
str(tmp_path / "docs" / "old.md"): file_bytes,
|
|
str(tmp_path / "archive" / "docs" / "old.md"): b"", # simulate post-mv
|
|
})
|
|
git = RecordingGit()
|
|
entry = _entry("docs/old.md", "move-to-archive", sha=sha, start=1, end=2,
|
|
extra={"dest_path": "archive/docs/old.md"})
|
|
result = _applier(tmp_path, fs=fs, git=git).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "move-to-archive"
|
|
assert len(git.calls) == 1
|
|
src, dest = git.calls[0]
|
|
assert src == str(tmp_path / "docs" / "old.md")
|
|
assert dest == str(tmp_path / "archive" / "docs" / "old.md")
|
|
|
|
def test_staged_paths_contains_dest_path(self, tmp_path):
|
|
content = "x\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "old.md"): file_bytes})
|
|
git = RecordingGit()
|
|
entry = _entry("old.md", "move-to-archive", sha=sha, start=1, end=1,
|
|
extra={"dest_path": "archive/old.md"})
|
|
result = _applier(tmp_path, fs=fs, git=git).apply([entry])
|
|
|
|
assert "archive/old.md" in result["staged_paths"]
|
|
|
|
def test_real_git_mv_stages_both_sides(self, tmp_path):
|
|
"""Integration: real git mv in a real repo; file is moved + staged."""
|
|
repo = _init_git_repo(tmp_path)
|
|
src = repo / "old.md"
|
|
src.write_bytes(b"old content\n")
|
|
subprocess.run(["git", "add", "old.md"], cwd=str(repo), check=True, capture_output=True)
|
|
subprocess.run(
|
|
["git", "commit", "-m", "add file"],
|
|
cwd=str(repo), check=True, capture_output=True
|
|
)
|
|
|
|
file_bytes = src.read_bytes()
|
|
sha = _sha256(file_bytes)
|
|
|
|
entry = _entry("old.md", "move-to-archive", sha=sha, start=1, end=1,
|
|
extra={"dest_path": "archive/old.md"})
|
|
result = PatchApplier(project_root=repo).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "move-to-archive"
|
|
assert not src.exists(), "source should be gone after git mv"
|
|
assert (repo / "archive" / "old.md").exists(), "dest should exist"
|
|
|
|
# Verify git staging.
|
|
status = subprocess.run(
|
|
["git", "status", "--porcelain"],
|
|
cwd=str(repo), capture_output=True, text=True
|
|
).stdout
|
|
# Expect a rename staging entry (R or D+A lines).
|
|
assert "old.md" in status
|
|
|
|
|
|
# ===========================================================================
|
|
# Per-file transaction: descending-order correctness (multi-entry)
|
|
# ===========================================================================
|
|
|
|
class TestMultiEntryDescendingOrder:
|
|
|
|
def test_two_delete_ranges_on_one_file_applied_correctly(self, tmp_path):
|
|
"""
|
|
Deleting lines 5-6 then 2-3 (descending) must not shift line numbers.
|
|
|
|
File (1-indexed):
|
|
1: header
|
|
2: remove-me-a
|
|
3: remove-me-b
|
|
4: keep-middle
|
|
5: remove-me-c
|
|
6: remove-me-d
|
|
7: footer
|
|
Expected after both deletes:
|
|
header
|
|
keep-middle
|
|
footer
|
|
"""
|
|
content = "header\nremove-me-a\nremove-me-b\nkeep-middle\nremove-me-c\nremove-me-d\nfooter\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
|
|
entries = [
|
|
_entry("doc.md", "delete-range", sha=sha, start=2, end=3), # lower range
|
|
_entry("doc.md", "delete-range", sha=sha, start=5, end=6), # higher range
|
|
]
|
|
result = _applier(tmp_path, fs=fs).apply(entries)
|
|
|
|
assert len(result["applied"]) == 2
|
|
assert result["skipped"] == []
|
|
written = fs.writes[str(tmp_path / "doc.md")].decode()
|
|
assert written == "header\nkeep-middle\nfooter\n"
|
|
|
|
def test_delete_range_then_insert_frontmatter_correct_order(self, tmp_path):
|
|
"""
|
|
insert-frontmatter must apply AFTER anchored edits.
|
|
File: 3 content lines. Delete line 3, then insert frontmatter.
|
|
Result: frontmatter block + line1 + line2.
|
|
"""
|
|
content = "line1\nline2\nline3\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
|
|
entries = [
|
|
_entry("doc.md", "insert-frontmatter",
|
|
extra={"key": "hygiene", "value": "frozen"}),
|
|
_entry("doc.md", "delete-range", sha=sha, start=3, end=3),
|
|
]
|
|
result = _applier(tmp_path, fs=fs).apply(entries)
|
|
|
|
assert len(result["applied"]) == 2
|
|
written = fs.writes[str(tmp_path / "doc.md")].decode()
|
|
assert written.startswith("---\nhygiene: frozen\n---\n")
|
|
assert "line1" in written
|
|
assert "line2" in written
|
|
assert "line3" not in written
|
|
|
|
|
|
# ===========================================================================
|
|
# Invariant #8: content changed since check → skip file, continue others
|
|
# ===========================================================================
|
|
|
|
class TestContentChangedGuard:
|
|
|
|
def test_changed_file_skipped_others_continue(self, tmp_path):
|
|
"""
|
|
Two files: doc-a.md has changed since check, doc-b.md is unchanged.
|
|
doc-a.md must be skipped; doc-b.md must be applied.
|
|
"""
|
|
# doc-a: content is different from the sha in the entry.
|
|
doc_a_bytes = b"new content not matching sha\n"
|
|
doc_a_stale_sha = _sha256(b"original content\n") # sha from check time
|
|
|
|
# doc-b: content matches.
|
|
doc_b_content = "line1\nline2\nline3\n"
|
|
doc_b_bytes, doc_b_sha = _make_file(doc_b_content)
|
|
|
|
fs = MemFS({
|
|
str(tmp_path / "doc-a.md"): doc_a_bytes,
|
|
str(tmp_path / "doc-b.md"): doc_b_bytes,
|
|
})
|
|
|
|
entries = [
|
|
_entry("doc-a.md", "delete-range", sha=doc_a_stale_sha, start=1, end=1),
|
|
_entry("doc-b.md", "delete-range", sha=doc_b_sha, start=2, end=2),
|
|
]
|
|
result = _applier(tmp_path, fs=fs).apply(entries)
|
|
|
|
# doc-a skipped
|
|
assert any(s["path"] == "doc-a.md" and s["reason"] == "content-changed-since-check"
|
|
for s in result["skipped"])
|
|
# doc-b applied
|
|
assert any(a["path"] == "doc-b.md" for a in result["applied"])
|
|
# doc-a NOT written
|
|
assert str(tmp_path / "doc-a.md") not in fs.writes
|
|
# doc-b IS written
|
|
assert str(tmp_path / "doc-b.md") in fs.writes
|
|
|
|
|
|
# ===========================================================================
|
|
# Incompatible-ops detection
|
|
# ===========================================================================
|
|
|
|
class TestIncompatibleOps:
|
|
|
|
def test_move_to_archive_plus_delete_range_skips_whole_file(self, tmp_path):
|
|
content = "content\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
git = RecordingGit()
|
|
|
|
entries = [
|
|
_entry("doc.md", "move-to-archive", sha=sha, start=1, end=1,
|
|
extra={"dest_path": "archive/doc.md"}),
|
|
_entry("doc.md", "delete-range", sha=sha, start=1, end=1),
|
|
]
|
|
result = _applier(tmp_path, fs=fs, git=git).apply(entries)
|
|
|
|
assert result["applied"] == []
|
|
assert all(s["reason"] == "incompatible-ops-on-file" for s in result["skipped"])
|
|
assert len(result["skipped"]) == 2 # both entries skipped
|
|
# git mv must NOT have been called.
|
|
assert git.calls == []
|
|
|
|
def test_overlapping_anchors_skips_whole_file(self, tmp_path):
|
|
content = "a\nb\nc\nd\ne\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
|
|
# Ranges 2-4 and 3-5 overlap.
|
|
entries = [
|
|
_entry("doc.md", "delete-range", sha=sha, start=2, end=4),
|
|
_entry("doc.md", "dedupe", sha=sha, start=3, end=5,
|
|
extra={"canonical_ref": "other.md"}),
|
|
]
|
|
result = _applier(tmp_path, fs=fs).apply(entries)
|
|
|
|
assert result["applied"] == []
|
|
assert all(s["reason"] == "incompatible-ops-on-file" for s in result["skipped"])
|
|
# File must not be written.
|
|
assert str(tmp_path / "doc.md") not in fs.writes
|
|
|
|
def test_non_overlapping_anchors_allowed(self, tmp_path):
|
|
content = "a\nb\nc\nd\ne\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
|
|
# Ranges 1-2 and 4-5 do NOT overlap.
|
|
entries = [
|
|
_entry("doc.md", "delete-range", sha=sha, start=1, end=2),
|
|
_entry("doc.md", "delete-range", sha=sha, start=4, end=5),
|
|
]
|
|
result = _applier(tmp_path, fs=fs).apply(entries)
|
|
|
|
assert result["skipped"] == []
|
|
assert len(result["applied"]) == 2
|
|
|
|
|
|
# ===========================================================================
|
|
# Unknown kind rejected
|
|
# ===========================================================================
|
|
|
|
class TestUnknownKind:
|
|
|
|
def test_unknown_kind_skipped_with_structured_reason(self, tmp_path):
|
|
fs = MemFS()
|
|
entry = {
|
|
"path": "doc.md",
|
|
"op": "nuke it",
|
|
"op_type": "deterministic",
|
|
"is_destructive": True,
|
|
"is_reversible": False,
|
|
"exact_edit": {"kind": "nuke-it", "anchor": {"start_line": 1, "end_line": 1}},
|
|
}
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "unknown-kind"
|
|
|
|
def test_generative_entry_rejected(self, tmp_path):
|
|
fs = MemFS()
|
|
entry = {
|
|
"path": "doc.md",
|
|
"op": "distill",
|
|
"op_type": "generative",
|
|
"is_destructive": False,
|
|
"is_reversible": True,
|
|
# No exact_edit (as per spec for generative).
|
|
}
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "generative-entry-rejected"
|
|
|
|
def test_unknown_kind_mixed_with_valid_sibling_reports_all(self, tmp_path):
|
|
"""
|
|
One invalid entry + one valid entry on the same file: neither must be
|
|
applied, and both must appear in skipped (valid sibling must not vanish).
|
|
"""
|
|
content = "line1\nline2\nline3\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
|
|
entries = [
|
|
{ # valid delete-range
|
|
"path": "doc.md",
|
|
"op": "delete line 2",
|
|
"op_type": "deterministic",
|
|
"exact_edit": {
|
|
"kind": "delete-range",
|
|
"anchor": {"start_line": 2, "end_line": 2},
|
|
"expected_sha256": sha,
|
|
},
|
|
},
|
|
{ # invalid kind
|
|
"path": "doc.md",
|
|
"op": "nuke",
|
|
"op_type": "deterministic",
|
|
"exact_edit": {"kind": "nuke-it", "anchor": {"start_line": 3, "end_line": 3}},
|
|
},
|
|
]
|
|
result = _applier(tmp_path, fs=fs).apply(entries)
|
|
|
|
assert result["applied"] == []
|
|
# Both entries must appear in skipped.
|
|
skipped_indices = {s["entry_index"] for s in result["skipped"]}
|
|
assert {0, 1} == skipped_indices
|
|
# File must NOT have been written.
|
|
assert str(tmp_path / "doc.md") not in fs.writes
|
|
|
|
|
|
class TestNonUtf8Content:
|
|
|
|
def test_non_utf8_file_skipped_not_corrupted(self, tmp_path):
|
|
"""
|
|
File with non-UTF-8 bytes: skip with reason, do not write a corrupted
|
|
version. The hash guard passes (we hash raw bytes) — this is the
|
|
edge case that errors="replace" would silently corrupt.
|
|
"""
|
|
# Craft bytes that are valid Latin-1 but invalid UTF-8.
|
|
raw_bytes = b"line1\nline2 with \xff non-utf8\nline3\n"
|
|
sha = _sha256(raw_bytes)
|
|
fs = MemFS({str(tmp_path / "doc.md"): raw_bytes})
|
|
|
|
entry = _entry("doc.md", "delete-range", sha=sha, start=2, end=2)
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "non-utf8-content"
|
|
# Must not have written anything.
|
|
assert str(tmp_path / "doc.md") not in fs.writes
|
|
|
|
|
|
# ===========================================================================
|
|
# move-to-archive idempotency (source already gone)
|
|
# ===========================================================================
|
|
|
|
class TestMoveToArchiveIdempotent:
|
|
|
|
def test_source_already_moved_is_success(self, tmp_path):
|
|
"""
|
|
If source is gone and dest exists, treat as idempotent success.
|
|
"""
|
|
dest_bytes = b"already archived\n"
|
|
sha = _sha256(b"original\n") # sha from check time (source is gone)
|
|
fs = MemFS({
|
|
# Source NOT in fs (already moved).
|
|
str(tmp_path / "archive" / "doc.md"): dest_bytes,
|
|
})
|
|
git = RecordingGit()
|
|
|
|
entry = _entry("doc.md", "move-to-archive", sha=sha, start=1, end=1,
|
|
extra={"dest_path": "archive/doc.md"})
|
|
result = _applier(tmp_path, fs=fs, git=git).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "move-to-archive"
|
|
assert result["failed"] == []
|
|
|
|
|
|
# ===========================================================================
|
|
# Exit codes via CLI main()
|
|
# ===========================================================================
|
|
|
|
class TestExitCodes:
|
|
|
|
def _write_report(self, path: Path, entries: list) -> None:
|
|
report = {
|
|
"schema_version": "1.0",
|
|
"tool_version": "0.1.0",
|
|
"generated_at": "2026-06-24T00:00:00+00:00",
|
|
"scan": {
|
|
"project_root": str(path.parent),
|
|
"scope_globs": ["**/*.md"],
|
|
"excluded_dirs": [],
|
|
"files_scanned": 1,
|
|
},
|
|
"shortlist": ["doc.md"],
|
|
"entries": entries,
|
|
}
|
|
path.write_text(json.dumps(report), encoding="utf-8")
|
|
|
|
def test_exit_0_all_applied(self, tmp_path):
|
|
content = "line1\nline2\nline3\n"
|
|
file_bytes, sha = _make_file(content)
|
|
(tmp_path / "doc.md").write_bytes(file_bytes)
|
|
|
|
report_path = tmp_path / "report.json"
|
|
self._write_report(report_path, [
|
|
{
|
|
"path": "doc.md",
|
|
"op": "delete line 2",
|
|
"op_type": "deterministic",
|
|
"is_destructive": True,
|
|
"is_reversible": False,
|
|
"safety_tier": "confirm",
|
|
"signals": [],
|
|
"category": {"class": "stale", "subtype": "contradicted"},
|
|
"token_estimate": {"raw_tokens": 10},
|
|
"exact_edit": {
|
|
"kind": "delete-range",
|
|
"anchor": {"start_line": 2, "end_line": 2},
|
|
"expected_sha256": sha,
|
|
},
|
|
}
|
|
])
|
|
|
|
rc = applier_main([
|
|
"--report", str(report_path),
|
|
"--apply-indices", "0",
|
|
"--project-root", str(tmp_path),
|
|
])
|
|
assert rc == 0
|
|
|
|
def test_exit_1_partial_skipped(self, tmp_path):
|
|
"""Content changed → skip → exit 1."""
|
|
actual_bytes = b"different content\n"
|
|
stale_sha = _sha256(b"original content\n")
|
|
(tmp_path / "doc.md").write_bytes(actual_bytes)
|
|
|
|
report_path = tmp_path / "report.json"
|
|
self._write_report(report_path, [
|
|
{
|
|
"path": "doc.md",
|
|
"op": "delete line 1",
|
|
"op_type": "deterministic",
|
|
"is_destructive": True,
|
|
"is_reversible": False,
|
|
"safety_tier": "confirm",
|
|
"signals": [],
|
|
"category": {"class": "stale", "subtype": "contradicted"},
|
|
"token_estimate": {"raw_tokens": 5},
|
|
"exact_edit": {
|
|
"kind": "delete-range",
|
|
"anchor": {"start_line": 1, "end_line": 1},
|
|
"expected_sha256": stale_sha,
|
|
},
|
|
}
|
|
])
|
|
|
|
rc = applier_main([
|
|
"--report", str(report_path),
|
|
"--apply-indices", "0",
|
|
"--project-root", str(tmp_path),
|
|
])
|
|
assert rc == 1
|
|
|
|
def test_exit_2_missing_report(self, tmp_path):
|
|
rc = applier_main([
|
|
"--report", str(tmp_path / "nonexistent.json"),
|
|
"--apply-indices", "0",
|
|
"--project-root", str(tmp_path),
|
|
])
|
|
assert rc == 2
|
|
|
|
def test_exit_2_out_of_range_index(self, tmp_path):
|
|
content = "x\n"
|
|
(tmp_path / "doc.md").write_bytes(content.encode())
|
|
report_path = tmp_path / "report.json"
|
|
self._write_report(report_path, []) # 0 entries
|
|
|
|
rc = applier_main([
|
|
"--report", str(report_path),
|
|
"--apply-indices", "5",
|
|
"--project-root", str(tmp_path),
|
|
])
|
|
assert rc == 2
|
|
|
|
def test_exit_2_invalid_indices_string(self, tmp_path):
|
|
# Write a minimal valid report.
|
|
report_path = tmp_path / "report.json"
|
|
report = {
|
|
"schema_version": "1.0", "tool_version": "0.1.0",
|
|
"generated_at": "2026-06-24T00:00:00+00:00",
|
|
"scan": {"project_root": str(tmp_path), "scope_globs": [], "excluded_dirs": [], "files_scanned": 0},
|
|
"shortlist": [], "entries": [],
|
|
}
|
|
report_path.write_text(json.dumps(report))
|
|
rc = applier_main([
|
|
"--report", str(report_path),
|
|
"--apply-indices", "abc",
|
|
"--project-root", str(tmp_path),
|
|
])
|
|
assert rc == 2
|
|
|
|
|
|
# ===========================================================================
|
|
# apply_indexed preserves original report indices
|
|
# ===========================================================================
|
|
|
|
class TestApplyIndexed:
|
|
|
|
def test_entry_index_matches_original_report_index(self, tmp_path):
|
|
content = "line1\nline2\nline3\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
|
|
entry = _entry("doc.md", "delete-range", sha=sha, start=2, end=2)
|
|
# original report index is 7 (not 0).
|
|
result = _applier(tmp_path, fs=fs).apply_indexed([(7, entry)])
|
|
|
|
assert result["applied"][0]["entry_index"] == 7
|