From 162dd2fd258f34f62c776d46ac25fa5b58ef5032 Mon Sep 17 00:00:00 2001 From: Andreas Poehlmann Date: Fri, 30 Aug 2024 18:52:38 +0200 Subject: [PATCH] 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 | 11 ++++++++++- upath/implementations/cloud.py | 7 ------- upath/tests/cases.py | 13 +++++++++++++ upath/tests/implementations/test_data.py | 10 ++++++++-- upath/tests/implementations/test_s3.py | 14 -------------- 5 files changed, 31 insertions(+), 24 deletions(-) diff --git a/upath/core.py b/upath/core.py index 5650819f..4c86e28d 100644 --- a/upath/core.py +++ b/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(): diff --git a/upath/implementations/cloud.py b/upath/implementations/cloud.py index a7fe60bc..e2f4cb98 100644 --- a/upath/implementations/cloud.py +++ b/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) diff --git a/upath/tests/cases.py b/upath/tests/cases.py index 1d50af5a..4f02539b 100644 --- a/upath/tests/cases.py +++ b/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() diff --git a/upath/tests/implementations/test_data.py b/upath/tests/implementations/test_data.py index 6fb1132a..9ada0687 100644 --- a/upath/tests/implementations/test_data.py +++ b/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() diff --git a/upath/tests/implementations/test_s3.py b/upath/tests/implementations/test_s3.py index 391bd4fb..89e71165 100644 --- a/upath/tests/implementations/test_s3.py +++ b/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"]] )