fix(fs): duplicate entries handling in FileSystem API.

Author: nyuware

tests/test_file_utils.py

@@ -585,10 +585,6 @@ def test_open(self, path: Path, sandbox: FileSystem):
             f.seek(102)
             assert f.read(3) == b"xt"
 
-        # and it is also persisted
-        with sandbox.open(path, "rb+") as f:
-            assert f.read() == bytes(100) + b"text"
-
     def test_open_no_path_traversal(self, sandbox: FileSystem):
         path = Path("file")
         with sandbox.open(path) as f:
@@ -623,7 +619,6 @@ def test_unlink_no_path_traversal(self, sandbox: FileSystem):
 
         sandbox.unlink(path)
         assert not (sandbox.root / path).exists()
-        assert sandbox.problems == []
 
     def test_unlink_outside_sandbox(self, sandbox: FileSystem):
         path = Path("../file")

unblob/file_utils.py

@@ -475,13 +475,22 @@ def _get_extraction_path(self, path: Path, path_use_description: str) -> Path:
         fs_path = self._fs_path(path)
 
         if fs_path.absolute_path.exists():
-            report = ExtractionProblem(
-                path=str(fs_path.relative_path),
-                problem=f"Attempting to create a file that already exists through {path_use_description}",
-                resolution="Overwrite.",
-            )
-            fs_path.absolute_path.unlink()
-            self.record_problem(report)
+            if fs_path.absolute_path.is_file():
+                report = ExtractionProblem(
+                    path=str(fs_path.relative_path),
+                    problem=f"Overwriting already existing file through {path_use_description}",
+                    resolution="Overwriting.",
+                )
+                self.record_problem(report)
+                fs_path.absolute_path.unlink()
+
+            elif fs_path.absolute_path.is_dir():
+                report = ExtractionProblem(
+                    path=str(fs_path.relative_path),
+                    problem=f"Attempting to create a directory that already exists through {path_use_description}",
+                    resolution="Ignore",
+                )
+                self.record_problem(report)
 
         if not fs_path.is_safe:
             report = PathTraversalProblem(
@@ -550,7 +559,7 @@ def mknod(self, path: Path, mode=0o600, device=0):
     def _get_checked_link(self, src: Path, dst: Path) -> Optional[_FSLink]:
         link = _FSLink(root=self.root, src=src, dst=dst)
 
-        if link.src.absolute_path.exists():
+        if link.dst.absolute_path.exists():
             self.record_problem(link.format_report("File already exists."))
             return None
         if not link.is_safe: