1481 lines
57 KiB
Python
1481 lines
57 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
|
|
|
|
|
|
# ===========================================================================
|
|
# Lifecycle deletion ops: `delete` and `extract-then-delete` (ADR-0039)
|
|
# ===========================================================================
|
|
#
|
|
# These two kinds are NOT drawn from KIND_TABLE (owned by report_builder.py /
|
|
# validate_report.py, group 3's file) — they are constructed directly here so
|
|
# these tests do not depend on that concurrent work landing first.
|
|
|
|
class RecordingGitState:
|
|
"""Injectable git-state double: fixed tracked/dirty answers per test."""
|
|
|
|
def __init__(self, tracked: bool = True, dirty: bool = False) -> None:
|
|
self.tracked = tracked
|
|
self.dirty = dirty
|
|
self.is_tracked_calls: list[tuple[str, str]] = []
|
|
self.is_dirty_calls: list[tuple[str, str]] = []
|
|
|
|
def is_tracked(self, path: str, cwd: Path) -> bool:
|
|
self.is_tracked_calls.append((path, str(cwd)))
|
|
return self.tracked
|
|
|
|
def is_dirty(self, path: str, cwd: Path) -> bool:
|
|
self.is_dirty_calls.append((path, str(cwd)))
|
|
return self.dirty
|
|
|
|
|
|
def _lifecycle_entry(
|
|
path: str,
|
|
kind: str,
|
|
safety_tier: str = "auto",
|
|
is_directory: bool = False,
|
|
extraction_complete: bool | None = None,
|
|
lifecycle: dict | None = None,
|
|
extraction_target: str | None = None,
|
|
extraction_pointer: dict | None = None,
|
|
) -> dict:
|
|
"""Build a minimal `delete` / `extract-then-delete` report entry.
|
|
|
|
Deliberately independent of KIND_TABLE (see module docstring above).
|
|
`extraction_target` (when given) mirrors the pre-existing frozen-schema
|
|
field validate_report.py already requires on extract-then-delete entries
|
|
(`"repo-durable"` or `"cross-repo"`, the vault case).
|
|
"""
|
|
exact_edit: dict[str, Any] = {"kind": kind}
|
|
if is_directory:
|
|
exact_edit["is_directory"] = True
|
|
if extraction_target is not None:
|
|
exact_edit["extraction_target"] = extraction_target
|
|
if extraction_complete is not None:
|
|
exact_edit["extraction_complete"] = extraction_complete
|
|
entry_extraction_complete = extraction_complete
|
|
entry: dict[str, Any] = {
|
|
"path": path,
|
|
"op": f"test op {kind}",
|
|
"op_type": "deterministic",
|
|
"is_destructive": True,
|
|
"is_reversible": False,
|
|
"safety_tier": safety_tier,
|
|
"exact_edit": exact_edit,
|
|
"lifecycle": lifecycle or {
|
|
"rule_ref": "temporary/test-rule",
|
|
"lifetime": "temporary",
|
|
},
|
|
}
|
|
if extraction_complete is not None:
|
|
entry["extraction_complete"] = extraction_complete
|
|
if extraction_pointer is not None:
|
|
entry["extraction_pointer"] = extraction_pointer
|
|
return entry
|
|
|
|
|
|
def _applier_ls(root: Path, fs=None, git=None, git_state=None) -> PatchApplier:
|
|
return PatchApplier(project_root=root, fs=fs, git=git, git_state=git_state)
|
|
|
|
|
|
class RecordingGitRm:
|
|
"""Records git rm calls (and optionally mv, unused here)."""
|
|
|
|
def __init__(self, fail: bool = False) -> None:
|
|
self.rm_calls: list[tuple[str, str, bool]] = []
|
|
self.mv_calls: list[tuple[str, str]] = []
|
|
self._fail = fail
|
|
|
|
def mv(self, src: Path, dest: Path, cwd: Path) -> None:
|
|
self.mv_calls.append((str(src), str(dest)))
|
|
|
|
def rm(self, path: Path, cwd: Path, recursive: bool = False) -> None:
|
|
if self._fail:
|
|
raise RuntimeError("git rm simulated failure")
|
|
self.rm_calls.append((str(path), str(cwd), recursive))
|
|
|
|
|
|
class TestLifecycleDeleteApplyTimeReverification:
|
|
"""Task 4.1(a): apply-time re-verification — never trust cached tier."""
|
|
|
|
def test_auto_tier_tracked_clean_applies_via_git_rm(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry("docs/old.md", "delete", safety_tier="auto")
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "delete"
|
|
assert result["skipped"] == []
|
|
assert len(git.rm_calls) == 1
|
|
rm_path, rm_cwd, recursive = git.rm_calls[0]
|
|
assert rm_path == str(tmp_path / "docs" / "old.md")
|
|
assert recursive is False
|
|
assert "docs/old.md" in result["staged_paths"]
|
|
# Re-verification actually consulted live git state for this path.
|
|
assert git_state.is_tracked_calls == [("docs/old.md", str(tmp_path))]
|
|
|
|
def test_auto_tier_downgraded_to_skip_when_untracked_since_check(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=False)
|
|
entry = _lifecycle_entry("docs/old.md", "delete", safety_tier="auto")
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "git-state-changed-since-check"
|
|
assert git.rm_calls == []
|
|
|
|
def test_auto_tier_downgraded_to_skip_when_dirty_since_check(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=True)
|
|
entry = _lifecycle_entry("docs/old.md", "delete", safety_tier="auto")
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "git-state-changed-since-check"
|
|
assert git.rm_calls == []
|
|
|
|
def test_other_entries_on_other_paths_continue_after_a_skip(self, tmp_path):
|
|
"""One entry's guard-skip must not block a sibling path's delete."""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=False) # everything looks untracked
|
|
entries = [
|
|
_lifecycle_entry("docs/a.md", "delete", safety_tier="auto"),
|
|
_lifecycle_entry("docs/b.md", "delete", safety_tier="auto"),
|
|
]
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply(entries)
|
|
|
|
assert result["applied"] == []
|
|
assert len(result["skipped"]) == 2
|
|
reasons = {s["reason"] for s in result["skipped"]}
|
|
assert reasons == {"git-state-changed-since-check"}
|
|
|
|
def test_confirm_tier_lifecycle_delete_is_not_re_verified(self, tmp_path):
|
|
"""
|
|
A confirm-tier entry was already gated through explicit human
|
|
approval upstream (batch-confirm). The applier must not silently
|
|
downgrade/skip it just because the path is untracked or dirty —
|
|
that dirty/untracked state is precisely why it was confirm-tier.
|
|
"""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=False, dirty=True)
|
|
entry = _lifecycle_entry("docs/old.md", "delete", safety_tier="confirm")
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "delete"
|
|
assert len(git.rm_calls) == 1
|
|
# No git-state query needed at all for an already-confirmed entry.
|
|
assert git_state.is_tracked_calls == []
|
|
|
|
|
|
class TestLifecycleDeleteGitRm:
|
|
"""Task 4.1(b): `delete` performs true git rm, staged precisely."""
|
|
|
|
def test_directory_aggregate_entry_uses_recursive_git_rm(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"autoresearch/run-042", "delete", safety_tier="auto", is_directory=True
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "delete"
|
|
rm_path, rm_cwd, recursive = git.rm_calls[0]
|
|
assert rm_path == str(tmp_path / "autoresearch" / "run-042")
|
|
assert recursive is True
|
|
|
|
def test_directory_delete_bypasses_content_hash_guard(self, tmp_path):
|
|
"""
|
|
A directory aggregate entry carries no expected_sha256 (no single
|
|
file to hash) — it must still apply cleanly via git rm, never
|
|
rejected for "missing" hash fields.
|
|
"""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"autoresearch/run-001", "delete", safety_tier="auto", is_directory=True
|
|
)
|
|
assert "expected_sha256" not in entry["exact_edit"]
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "delete"
|
|
assert result["failed"] == []
|
|
|
|
def test_git_rm_failure_reported_as_failed(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm(fail=True)
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry("docs/old.md", "delete", safety_tier="auto")
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["failed"][0]["kind"] == "delete"
|
|
assert "git rm simulated failure" in result["failed"][0]["error"]
|
|
|
|
def test_real_git_rm_stages_deletion(self, tmp_path):
|
|
"""Integration: real git rm in a real repo removes + stages the file."""
|
|
repo = _init_git_repo(tmp_path)
|
|
target = repo / "old.md"
|
|
target.write_bytes(b"stale 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,
|
|
)
|
|
|
|
entry = _lifecycle_entry("old.md", "delete", safety_tier="auto")
|
|
result = PatchApplier(project_root=repo).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "delete"
|
|
assert not target.exists(), "file should be gone after git rm"
|
|
|
|
status = subprocess.run(
|
|
["git", "status", "--porcelain"],
|
|
cwd=str(repo), capture_output=True, text=True,
|
|
).stdout
|
|
# Staged deletion shows as "D old.md" (space, not "D " in worktree column).
|
|
assert "D old.md" in status
|
|
|
|
|
|
class TestExtractThenDeleteFailsClosed:
|
|
"""Task 4.1(c): extract-then-delete only deletes once extraction is done."""
|
|
|
|
def test_skipped_when_extraction_not_marked_complete(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/old-guide.md", "extract-then-delete", safety_tier="auto",
|
|
extraction_complete=False,
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "extraction-not-confirmed"
|
|
assert git.rm_calls == []
|
|
|
|
def test_skipped_when_extraction_flag_absent(self, tmp_path):
|
|
"""No extraction_complete field at all → treated as not confirmed."""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/old-guide.md", "extract-then-delete", safety_tier="auto",
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "extraction-not-confirmed"
|
|
assert git.rm_calls == []
|
|
|
|
def test_applies_git_rm_once_extraction_confirmed(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/old-guide.md", "extract-then-delete", safety_tier="auto",
|
|
extraction_complete=True,
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "extract-then-delete"
|
|
assert len(git.rm_calls) == 1
|
|
|
|
def test_failed_extraction_is_per_entry_skip_not_run_level_failure(self, tmp_path):
|
|
"""
|
|
A failed extraction for one entry must not prevent a sibling path's
|
|
entry (independent extraction) from applying in the same run.
|
|
"""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entries = [
|
|
_lifecycle_entry(
|
|
"docs/not-extracted.md", "extract-then-delete", safety_tier="auto",
|
|
extraction_complete=False,
|
|
),
|
|
_lifecycle_entry(
|
|
"docs/extracted.md", "extract-then-delete", safety_tier="auto",
|
|
extraction_complete=True,
|
|
),
|
|
]
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply(entries)
|
|
|
|
assert len(result["applied"]) == 1
|
|
assert result["applied"][0]["path"] == "docs/extracted.md"
|
|
assert len(result["skipped"]) == 1
|
|
assert result["skipped"][0]["reason"] == "extraction-not-confirmed"
|
|
|
|
def test_confirm_tier_extract_then_delete_still_fails_closed_on_extraction(self, tmp_path):
|
|
"""
|
|
Even for an already-approved confirm-tier entry, extraction
|
|
confirmation is unconditional — the applier itself never performs
|
|
the generative extraction, so it cannot proceed without the flag.
|
|
"""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=False, dirty=True)
|
|
entry = _lifecycle_entry(
|
|
"docs/old-guide.md", "extract-then-delete", safety_tier="confirm",
|
|
extraction_complete=False,
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert result["skipped"][0]["reason"] == "extraction-not-confirmed"
|
|
assert git.rm_calls == []
|
|
|
|
|
|
class WriteFailsFS(MemFS):
|
|
"""MemFS that raises on write_bytes for a chosen path (simulates an
|
|
unwritable extracted.md — e.g. its directory was removed)."""
|
|
|
|
def __init__(self, files: dict[str, bytes] | None = None, fail_path: str | None = None) -> None:
|
|
super().__init__(files)
|
|
self._fail_path = fail_path
|
|
|
|
def write_bytes(self, path: Path, data: bytes) -> None:
|
|
if self._fail_path is not None and str(path) == self._fail_path:
|
|
raise OSError("simulated write failure: directory removed")
|
|
super().write_bytes(path, data)
|
|
|
|
|
|
class TestExtractThenDeleteVaultPointerAppend:
|
|
"""Task 4.2: vault-destination extractions append an extracted.md pointer,
|
|
staged into the same transaction as the git rm; repo-durable destinations
|
|
get no index entry; a failed append (or missing/invalid pointer
|
|
metadata) skips the delete (fail closed, never gone-but-unindexed)."""
|
|
|
|
def _pointer(self, **overrides) -> dict:
|
|
base = {
|
|
"vault_note": "tool/graphify-clustering-behavior",
|
|
"reason": "why HDBSCAN over-merges doc clusters; read before "
|
|
"tuning any graphify clustering params",
|
|
"date": "2026-07-15",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
def test_vault_extraction_appends_pointer_and_creates_extracted_md(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/graphify-clustering-notes.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
extraction_target="cross-repo", extraction_pointer=self._pointer(),
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "extract-then-delete"
|
|
assert result["skipped"] == []
|
|
assert len(git.rm_calls) == 1
|
|
|
|
extracted_path = str(tmp_path / "docs" / "extracted.md")
|
|
assert extracted_path in fs.writes
|
|
content = fs.writes[extracted_path].decode("utf-8")
|
|
expected_line = (
|
|
"- [[vault: tool/graphify-clustering-behavior]] — why HDBSCAN "
|
|
"over-merges doc clusters; read before tuning any graphify "
|
|
"clustering params (extracted from `graphify-clustering-notes.md`, "
|
|
"2026-07-15)"
|
|
)
|
|
assert expected_line in content
|
|
assert "docs/extracted.md" in result["staged_paths"]
|
|
assert "docs/graphify-clustering-notes.md" in result["staged_paths"]
|
|
|
|
def test_vault_extraction_appends_to_existing_extracted_md(self, tmp_path):
|
|
existing_path = str(tmp_path / "docs" / "extracted.md")
|
|
existing_content = (
|
|
"- [[vault: tool/old-note]] — prior entry "
|
|
"(extracted from `old.md`, 2026-07-01)\n"
|
|
)
|
|
fs = MemFS({existing_path: existing_content.encode("utf-8")})
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/graphify-clustering-notes.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
extraction_target="cross-repo", extraction_pointer=self._pointer(),
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "extract-then-delete"
|
|
content = fs.writes[existing_path].decode("utf-8")
|
|
assert "tool/old-note" in content
|
|
assert "tool/graphify-clustering-behavior" in content
|
|
# Prior entry preserved before the newly appended one.
|
|
assert content.index("tool/old-note") < content.index("tool/graphify-clustering-behavior")
|
|
|
|
def test_repo_durable_extraction_writes_no_extracted_md(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/old-guide.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
extraction_target="repo-durable",
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "extract-then-delete"
|
|
assert fs.writes == {}
|
|
assert result["staged_paths"] == ["docs/old-guide.md"]
|
|
|
|
def test_no_extraction_target_writes_no_extracted_md(self, tmp_path):
|
|
"""Absence of extraction_target (or any value other than
|
|
"cross-repo") is treated as repo-durable — no pointer append."""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/old-guide.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "extract-then-delete"
|
|
assert fs.writes == {}
|
|
|
|
def test_failed_append_skips_the_delete(self, tmp_path):
|
|
extracted_path = str(tmp_path / "docs" / "extracted.md")
|
|
fs = WriteFailsFS(fail_path=extracted_path)
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/graphify-clustering-notes.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
extraction_target="cross-repo", extraction_pointer=self._pointer(),
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert git.rm_calls == []
|
|
assert result["skipped"][0]["reason"] == "extracted-md-append-failed"
|
|
|
|
def test_missing_pointer_metadata_skips_the_delete(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/graphify-clustering-notes.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
extraction_target="cross-repo",
|
|
# No extraction_pointer at all.
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert git.rm_calls == []
|
|
assert result["skipped"][0]["reason"] == "extraction-pointer-invalid"
|
|
assert fs.writes == {}
|
|
|
|
def test_incomplete_pointer_metadata_skips_the_delete(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/graphify-clustering-notes.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
extraction_target="cross-repo",
|
|
extraction_pointer={"vault_note": "tool/x"}, # missing reason/date
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert git.rm_calls == []
|
|
assert result["skipped"][0]["reason"] == "extraction-pointer-invalid"
|
|
assert fs.writes == {}
|
|
|
|
def test_extraction_complete_false_skips_before_any_pointer_logic(self, tmp_path):
|
|
"""extraction_complete: false still skips everything, even with a
|
|
fully-formed vault pointer present."""
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"docs/graphify-clustering-notes.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=False,
|
|
extraction_target="cross-repo", extraction_pointer=self._pointer(),
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"] == []
|
|
assert git.rm_calls == []
|
|
assert result["skipped"][0]["reason"] == "extraction-not-confirmed"
|
|
assert fs.writes == {}
|
|
|
|
def test_root_level_file_uses_root_extracted_md(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry(
|
|
"graphify-clustering-notes.md", "extract-then-delete",
|
|
safety_tier="auto", extraction_complete=True,
|
|
extraction_target="cross-repo", extraction_pointer=self._pointer(),
|
|
)
|
|
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply([entry])
|
|
|
|
assert result["applied"][0]["kind"] == "extract-then-delete"
|
|
extracted_path = str(tmp_path / "extracted.md")
|
|
assert extracted_path in fs.writes
|
|
assert "extracted.md" in result["staged_paths"]
|
|
|
|
|
|
class TestLifecycleConfirmTierGating:
|
|
"""Task 4.1(d): confirm-tier lifecycle entries never auto-apply silently."""
|
|
|
|
def test_confirm_tier_entry_only_applies_when_explicitly_indexed(self, tmp_path):
|
|
"""
|
|
Mirrors the existing --apply-indices gating convention: the applier
|
|
only ever touches entries the caller explicitly selected. A
|
|
confirm-tier lifecycle entry sitting in report.entries but NOT
|
|
passed via --apply-indices must never be applied.
|
|
"""
|
|
content = "line1\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
|
|
report_entries = [
|
|
_entry("doc.md", "delete-range", sha=sha, start=1, end=1), # index 0, auto
|
|
_lifecycle_entry("docs/risky.md", "delete", safety_tier="confirm"), # index 1
|
|
]
|
|
report = {
|
|
"schema_version": "1.0", "tool_version": "0.1.0",
|
|
"generated_at": "2026-07-14T00:00:00+00:00",
|
|
"scan": {"project_root": str(tmp_path), "scope_globs": [], "excluded_dirs": [], "files_scanned": 0},
|
|
"shortlist": [], "entries": report_entries,
|
|
}
|
|
report_path = tmp_path / "report.json"
|
|
report_path.write_text(json.dumps(report))
|
|
|
|
# Caller (skill) only approved index 0 — the confirm-tier delete was
|
|
# never approved and must not be passed.
|
|
applier = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state)
|
|
indexed = [(0, report_entries[0])]
|
|
result = applier.apply_indexed(indexed)
|
|
|
|
assert result["applied"][0]["entry_index"] == 0
|
|
assert git.rm_calls == [] # confirm-tier delete never touched
|
|
|
|
def test_confirm_tier_entry_applies_once_explicitly_approved_and_indexed(self, tmp_path):
|
|
fs = MemFS()
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
entry = _lifecycle_entry("docs/risky.md", "delete", safety_tier="confirm")
|
|
|
|
# Caller explicitly included this index after the user approved it
|
|
# in the batch-confirm prompt.
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply_indexed(
|
|
[(1, entry)]
|
|
)
|
|
|
|
assert result["applied"][0]["entry_index"] == 1
|
|
assert result["applied"][0]["kind"] == "delete"
|
|
assert len(git.rm_calls) == 1
|
|
|
|
|
|
class TestLifecycleDeleteIncompatibleWithOtherOps:
|
|
"""A lifecycle delete cannot be combined with any other op on the same path."""
|
|
|
|
def test_delete_combined_with_another_op_on_same_path_is_incompatible(self, tmp_path):
|
|
content = "line1\n"
|
|
file_bytes, sha = _make_file(content)
|
|
fs = MemFS({str(tmp_path / "doc.md"): file_bytes})
|
|
git = RecordingGitRm()
|
|
git_state = RecordingGitState(tracked=True, dirty=False)
|
|
|
|
entries = [
|
|
_lifecycle_entry("doc.md", "delete", safety_tier="auto"),
|
|
_entry("doc.md", "insert-frontmatter", extra={"key": "hygiene", "value": "frozen"}),
|
|
]
|
|
result = _applier_ls(tmp_path, fs=fs, git=git, git_state=git_state).apply(entries)
|
|
|
|
assert result["applied"] == []
|
|
reasons = {s["reason"] for s in result["skipped"]}
|
|
assert reasons == {"incompatible-ops-on-file"}
|
|
assert git.rm_calls == []
|
|
|
|
|
|
class TestUnknownKindStillRejectedAlongsideLifecycleKinds:
|
|
"""Regression: adding the lifecycle carve-out must not swallow real unknowns."""
|
|
|
|
def test_truly_unknown_kind_still_reported(self, tmp_path):
|
|
fs = MemFS()
|
|
entry = {
|
|
"path": "doc.md",
|
|
"op": "nuke it",
|
|
"op_type": "deterministic",
|
|
"exact_edit": {"kind": "nuke-it", "anchor": {"start_line": 1, "end_line": 1}},
|
|
}
|
|
result = _applier(tmp_path, fs=fs).apply([entry])
|
|
|
|
assert result["skipped"][0]["reason"] == "unknown-kind"
|