Coverage for smith / infrastructure / filesystem.py: 100%

1"""Atomic file-system adapter — write files transactionally or remove them.""" 

2 

3import shutil 

4import tempfile 

5from pathlib import Path 

6 

7from smith.domain.value_objects import FileSpec 

8 

9 

10class FileSystemError(Exception): 

11 """Raised when an atomic file operation fails.""" 

12 

13 

14class AtomicFileSystem: 

15 """File-system adapter that writes files atomically via a staging directory.""" 

16 

17 def __init__(self, project_dir: Path) -> None: 

18 """Initialise with the project root directory.""" 

19 self._project_dir = project_dir 

20 

21 def check_conflicts(self, paths: list[Path]) -> list[Path]: 

22 """Return the subset of *paths* that already exist on disk.""" 

23 return [p for p in paths if (self._project_dir / p).exists()] 

24 

25 def write_atomic(self, specs: list[FileSpec]) -> None: 

26 """Write all specs atomically; roll back on failure.""" 

27 if not specs: 

28 return 

29 staging = Path(tempfile.mkdtemp(dir=self._project_dir)) 

30 try: 

31 for spec in specs: 

32 dest = staging / spec.relative_path 

33 dest.parent.mkdir(parents=True, exist_ok=True) 

34 dest.write_bytes(spec.content) 

35 for spec in specs: 

36 final = self._project_dir / spec.relative_path 

37 final.parent.mkdir(parents=True, exist_ok=True) 

38 Path(staging / spec.relative_path).replace(final) 

39 except Exception as err: 

40 shutil.rmtree(staging, ignore_errors=True) 

41 raise FileSystemError("Atomic write failed") from err 

42 else: 

43 shutil.rmtree(staging, ignore_errors=True) 

44 

45 def remove(self, paths: list[Path]) -> None: 

46 """Remove the given paths from the project directory.""" 

47 for p in paths: 

48 full = self._project_dir / p 

49 if full.exists(): 

50 full.unlink() 

51 

52 def exists(self, paths: list[Path]) -> dict[Path, bool]: 

53 """Return a mapping of each path to whether it exists on disk.""" 

54 return {p: (self._project_dir / p).exists() for p in paths}