Fix touch exists_ok if file exists (#262)

* tests: touch exists_ok behavior

* upath: fix touch exists_ok

* upath.implementations.cloud: remove AzurePath.touch custom method

upath/core.py

@@ -976,7 +976,16 @@ def readlink(self) -> Self:
         raise NotImplementedError
 
     def touch(self, mode=0o666, exist_ok=True) -> None:
-        self.fs.touch(self.path, truncate=not exist_ok)
+        exists = self.fs.exists(self.path)
+        if exists and not exist_ok:
+            raise FileExistsError(str(self))
+        if not exists:
+            self.fs.touch(self.path, truncate=True)
+        else:
+            try:
+                self.fs.touch(self.path, truncate=False)
+            except (NotImplementedError, ValueError):
+                pass  # unsupported by filesystem
 
     def mkdir(self, mode=0o777, parents=False, exist_ok=False) -> None:
         if parents and not exist_ok and self.exists():

upath/implementations/cloud.py

@@ -80,10 +80,3 @@ class S3Path(CloudPath):
 
 class AzurePath(CloudPath):
     __slots__ = ()
-
-    def touch(self, mode=0o666, exist_ok=True):
-        if exist_ok and self.exists():
-            with self.fs.open(self.path, mode="a"):
-                pass
-        else:
-            self.fs.touch(self.path, truncate=True)

upath/tests/cases.py

@@ -320,6 +320,19 @@ def test_rglob(self, pathlib_base):
     def test_symlink_to(self):
         pass
 
+    def test_touch_exists_ok_false(self):
+        f = self.path.joinpath("file1.txt")
+        assert f.exists()
+        with pytest.raises(FileExistsError):
+            f.touch(exist_ok=False)
+
+    def test_touch_exists_ok_true(self):
+        f = self.path.joinpath("file1.txt")
+        assert f.exists()
+        data = f.read_text()
+        f.touch(exist_ok=True)
+        assert f.read_text() == data
+
     def test_touch_unlink(self):
         path = self.path.joinpath("test_touch.txt")
         path.touch()

upath/tests/implementations/test_data.py

@@ -133,9 +133,15 @@ def test_rglob(self, pathlib_base):
         with pytest.raises(NotImplementedError):
             list(self.path.rglob("*"))
 
+    def test_touch_exists_ok_false(self):
+        with pytest.raises(FileExistsError):
+            self.path.touch(exist_ok=False)
+
+    def test_touch_exists_ok_true(self):
+        self.path.touch()
+
     def test_touch_unlink(self):
-        with pytest.raises(NotImplementedError):
-            self.path.touch()
+        self.path.touch()
         with pytest.raises(NotImplementedError):
             self.path.unlink()
 

upath/tests/implementations/test_s3.py

@@ -53,20 +53,6 @@ def test_iterdir_root(self):
             assert x.name != ""
             assert x.exists()
 
-    def test_touch_unlink(self):
-        path = self.path.joinpath("test_touch.txt")
-        path.touch()
-        assert path.exists()
-        path.unlink()
-        assert not path.exists()
-
-        # should raise FileNotFoundError since file is missing
-        with pytest.raises(FileNotFoundError):
-            path.unlink()
-
-        # file doesn't exists, but missing_ok is True
-        path.unlink(missing_ok=True)
-
     @pytest.mark.parametrize(
         "joiner", [["bucket", "path", "file"], ["bucket/path/file"]]
     )