Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion src/svrf/gate.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,18 @@ def _prepare(self, slot: Path, commit: str) -> None:
slot.parent.mkdir(parents=True, exist_ok=True)
subprocess.run(["git", "-C", str(self.clone), "worktree", "prune"], capture_output=True)
self._git("worktree", "add", "-q", "--detach", "-f", str(slot), commit)
self._git("checkout", "-q", "--detach", "-f", commit, cwd=slot)
else:
# A reused slot may carry flags that hide changed bytes. Clear each
# kind independently; skip-worktree also suppresses force-checkout.
entries = subprocess.run(["git", "-C", str(slot), "ls-files", "-v", "-z"],
check=True, capture_output=True).stdout.split(b"\0")
assumed = [e[2:] for e in entries if e[:1].islower()]
skipped = [e[2:] for e in entries if e[:1].upper() == b"S"]
for option, paths in (("--no-assume-unchanged", assumed), ("--no-skip-worktree", skipped)):
if paths:
subprocess.run(["git", "-C", str(slot), "update-index", option, "-z", "--stdin"],
input=b"\0".join(paths) + b"\0", check=True, capture_output=True)
self._git("checkout", "-q", "--detach", "-f", commit, cwd=slot)
self._git("clean", "-fdq", cwd=slot)

def _step(self, command: str, cwd: Path, log: Path, env: dict) -> tuple[int, str, float]:
Expand Down
73 changes: 73 additions & 0 deletions tests/test_gate_same_source.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
"""Gate preparation materializes a new slot once and restores reused source."""
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from fakes import git
from svrf.gate import CommandGate


class SameSourceGate(unittest.TestCase):
def test_initial_preparation_does_not_repeat_worktree_materialization(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / "repository"
root.mkdir()
git(root, "init", "-q", "-b", "main")
(root / "source.txt").write_text("source\n")
git(root, "add", ".")
git(root, "commit", "-qm", "source")
head = git(root, "rev-parse", "HEAD").strip()
gate = CommandGate(root, Path(tmp) / "slots", 1, ["true"], logs=Path(tmp) / "logs")
slot = Path(tmp) / "slots" / "slot-0"
original = gate._git
calls = []
def traced(*args, **kwargs):
calls.append(args)
return original(*args, **kwargs)
with patch.object(gate, "_git", side_effect=traced):
gate._prepare(slot, head)
self.assertEqual(sum(args[:2] == ("worktree", "add") for args in calls), 1)
self.assertFalse(any(args and args[0] == "checkout" for args in calls), calls)
calls.clear()
with patch.object(gate, "_git", side_effect=traced):
gate._prepare(slot, head)
self.assertEqual(sum(args and args[0] == "checkout" for args in calls), 1)
self.assertEqual((slot / "source.txt").read_text(), "source\n")

def test_dirty_tracked_source_is_restored_before_gate(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / "repository"
root.mkdir()
git(root, "init", "-q", "-b", "main")
(root / "source.txt").write_text("source\n")
git(root, "add", ".")
git(root, "commit", "-qm", "source")
head = git(root, "rev-parse", "HEAD").strip()
gate = CommandGate(root, Path(tmp) / "slots", 1, ["true"], logs=Path(tmp) / "logs")
slot = Path(tmp) / "slots" / "slot-0"
gate._prepare(slot, head)
(slot / "source.txt").write_text("tamper\n")
gate._prepare(slot, head)
self.assertEqual((slot / "source.txt").read_text(), "source\n")

def test_index_flags_cannot_hide_changed_source_from_gate(self):
for flag in ("--assume-unchanged", "--skip-worktree"):
with self.subTest(flag=flag), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp) / "repository"
root.mkdir()
git(root, "init", "-q", "-b", "main")
(root / "source.txt").write_text("source\n")
git(root, "add", ".")
git(root, "commit", "-qm", "source")
head = git(root, "rev-parse", "HEAD").strip()
gate = CommandGate(root, Path(tmp) / "slots", 1,
["test \"$(cat source.txt)\" = source"],
logs=Path(tmp) / "logs")
slot = Path(tmp) / "slots" / "slot-0"
gate._prepare(slot, head)
git(slot, "update-index", flag, "source.txt")
(slot / "source.txt").write_text("tamper\n")
self.assertEqual(git(slot, "status", "--porcelain"), "")
result = gate.run(head, head, flag)
self.assertTrue(result["green"], result)
self.assertEqual((slot / "source.txt").read_text(), "source\n")
Loading