Doni/CommonAutoRearsh
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve artifact line endings and restore deletions

Browse Source
This commit is contained in:
sladro 2026-04-02 11:52:10 +08:00
parent f261f0bf8f
commit 3172ce264b
2 changed files with 64 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
engine/artifact_manager.py
View File

@ -1,10 +1,10 @@
from __future__ import annotations
from dataclasses import dataclass
from difflib import unified_diff
from fnmatch import fnmatch
from hashlib import sha256
from pathlib import Path
from difflib import unified_diff
from engine.models import BaselineSnapshot, TaskSpec
@ -30,19 +30,28 @@ class ArtifactManager:
file_contents: dict[Path, str] = {}
file_hashes: dict[Path, str] = {}
for path in self.resolve_paths():
content = path.read_text(encoding="utf-8")
with path.open("r", encoding="utf-8", newline="") as handle:
content = handle.read()
file_contents[path] = content
file_hashes[path] = sha256(content.encode("utf-8")).hexdigest()
return BaselineSnapshot(file_contents=file_contents, file_hashes=file_hashes)
def restore(self, snapshot: BaselineSnapshot) -> None:
current_paths = set(self.resolve_paths())
snapshot_paths = set(snapshot.file_contents)
for path in current_paths - snapshot_paths:
path.unlink()
for path, content in snapshot.file_contents.items():
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
with path.open("w", encoding="utf-8", newline="") as handle:
handle.write(content)
def diff_summary(self, snapshot: BaselineSnapshot) -> str:
lines: list[str] = []
current_contents = {path: path.read_text(encoding="utf-8") for path in self.resolve_paths()}
current_contents: dict[Path, str] = {}
for path in self.resolve_paths():
with path.open("r", encoding="utf-8", newline="") as handle:
current_contents[path] = handle.read()
all_paths = sorted(set(snapshot.file_contents) | set(current_contents))
for path in all_paths:
before = snapshot.file_contents.get(path, "")

51
tests/test_artifact_manager.py
View File

@ -1,4 +1,5 @@
from pathlib import Path
from hashlib import sha256
import tempfile
import unittest
@ -42,6 +43,56 @@ class ArtifactManagerTest(unittest.TestCase):
manager.restore(snapshot)
self.assertEqual(target.read_text(encoding="utf-8"), "hello\n")
def test_restore_removes_newly_created_included_artifact(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "artifacts"
artifact_dir.mkdir()
target = artifact_dir / "sample.md"
target.write_text("hello\n", encoding="utf-8")
manager = ArtifactManager(make_task(root))
snapshot = manager.snapshot()
extra = artifact_dir / "new.md"
extra.write_text("new\n", encoding="utf-8")
manager.restore(snapshot)
self.assertFalse(extra.exists())
self.assertEqual(target.read_text(encoding="utf-8"), "hello\n")
def test_snapshot_and_restore_preserve_crlf_content(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "artifacts"
artifact_dir.mkdir()
target = artifact_dir / "sample.md"
original = "line1\r\nline2\r\n"
with target.open("w", encoding="utf-8", newline="") as handle:
handle.write(original)
manager = ArtifactManager(make_task(root))
snapshot = manager.snapshot()
self.assertEqual(snapshot.file_contents[target], original)
self.assertEqual(
snapshot.file_hashes[target],
sha256(original.encode("utf-8")).hexdigest(),
)
with target.open("w", encoding="utf-8", newline="") as handle:
handle.write("changed\r\n")
manager.restore(snapshot)
with target.open("r", encoding="utf-8", newline="") as handle:
restored = handle.read()
self.assertEqual(restored, original)
def test_resolve_paths_is_deterministic_and_respects_excludes(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "artifacts"
artifact_dir.mkdir()
(artifact_dir / "b.md").write_text("b\n", encoding="utf-8")
(artifact_dir / "ignore.md").write_text("ignore\n", encoding="utf-8")
(artifact_dir / "a.md").write_text("a\n", encoding="utf-8")
manager = ArtifactManager(make_task(root))
paths = manager.resolve_paths()
self.assertEqual(paths, [artifact_dir / "a.md", artifact_dir / "b.md"])
def test_diff_summary_contains_changed_line(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)