added quote to urls, adjusted tests to cover this (#194)

Ref #193

The same as in #192 but for all other DAV methods.
Tests were expanded to catch this case.

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

Author: bigcat88

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file.
 ### Fixed
 
 - files: proper url encoding of special chars in `mkdir` and `delete` methods. #191 Thanks to @tobenary
+- files: proper url encoding of special chars in all other `DAV` methods. #194
 
 ## [0.7.1 - 2022-12-21]
 

nc_py_api/files/files.py

@@ -86,7 +86,7 @@ def find(self, req: list, path: str | FsNode = "") -> list[FsNode]:
     def download(self, path: str | FsNode) -> bytes:
         """Downloads and returns the content of a file."""
         path = path.user_path if isinstance(path, FsNode) else path
-        response = self._session.adapter_dav.get(dav_get_obj_path(self._session.user, path))
+        response = self._session.adapter_dav.get(quote(dav_get_obj_path(self._session.user, path)))
         check_error(response, f"download: user={self._session.user}, path={path}")
         return response.content
 
@@ -138,7 +138,7 @@ def upload(self, path: str | FsNode, content: bytes | str) -> FsNode:
         """
         path = path.user_path if isinstance(path, FsNode) else path
         full_path = dav_get_obj_path(self._session.user, path)
-        response = self._session.adapter_dav.put(full_path, content=content)
+        response = self._session.adapter_dav.put(quote(full_path), content=content)
         check_error(response, f"upload: user={self._session.user}, path={path}, size={len(content)}")
         return FsNode(full_path.strip("/"), **etag_fileid_from_response(response))
 
@@ -219,11 +219,11 @@ def move(self, path_src: str | FsNode, path_dest: str | FsNode, overwrite=False)
         full_dest_path = dav_get_obj_path(
             self._session.user, path_dest.user_path if isinstance(path_dest, FsNode) else path_dest
         )
-        dest = self._session.cfg.dav_endpoint + full_dest_path
+        dest = self._session.cfg.dav_endpoint + quote(full_dest_path)
         headers = Headers({"Destination": dest, "Overwrite": "T" if overwrite else "F"}, encoding="utf-8")
         response = self._session.adapter_dav.request(
             "MOVE",
-            dav_get_obj_path(self._session.user, path_src),
+            quote(dav_get_obj_path(self._session.user, path_src)),
             headers=headers,
         )
         check_error(response, f"move: user={self._session.user}, src={path_src}, dest={dest}, {overwrite}")
@@ -241,11 +241,11 @@ def copy(self, path_src: str | FsNode, path_dest: str | FsNode, overwrite=False)
         full_dest_path = dav_get_obj_path(
             self._session.user, path_dest.user_path if isinstance(path_dest, FsNode) else path_dest
         )
-        dest = self._session.cfg.dav_endpoint + full_dest_path
+        dest = self._session.cfg.dav_endpoint + quote(full_dest_path)
         headers = Headers({"Destination": dest, "Overwrite": "T" if overwrite else "F"}, encoding="utf-8")
         response = self._session.adapter_dav.request(
             "COPY",
-            dav_get_obj_path(self._session.user, path_src),
+            quote(dav_get_obj_path(self._session.user, path_src)),
             headers=headers,
         )
         check_error(response, f"copy: user={self._session.user}, src={path_src}, dest={dest}, {overwrite}")
@@ -277,7 +277,7 @@ def setfav(self, path: str | FsNode, value: int | bool) -> None:
         path = path.user_path if isinstance(path, FsNode) else path
         root = build_setfav_req(value)
         webdav_response = self._session.adapter_dav.request(
-            "PROPPATCH", dav_get_obj_path(self._session.user, path), content=element_tree_as_str(root)
+            "PROPPATCH", quote(dav_get_obj_path(self._session.user, path)), content=element_tree_as_str(root)
         )
         check_error(webdav_response, f"setfav: path={path}, value={value}")
 
@@ -301,7 +301,7 @@ def trashbin_restore(self, path: str | FsNode) -> None:
         headers = Headers({"Destination": dest}, encoding="utf-8")
         response = self._session.adapter_dav.request(
             "MOVE",
-            f"/trashbin/{self._session.user}/{path}",
+            quote(f"/trashbin/{self._session.user}/{path}"),
             headers=headers,
         )
         check_error(response, f"trashbin_restore: user={self._session.user}, src={path}, dest={dest}")
@@ -313,7 +313,7 @@ def trashbin_delete(self, path: str | FsNode, not_fail=False) -> None:
         :param not_fail: if set to ``True`` and the object is not found, it does not raise an exception.
         """
         path = path.user_path if isinstance(path, FsNode) else path
-        response = self._session.adapter_dav.delete(f"/trashbin/{self._session.user}/{path}")
+        response = self._session.adapter_dav.delete(quote(f"/trashbin/{self._session.user}/{path}"))
         if response.status_code == 404 and not_fail:
             return
         check_error(response)
@@ -431,7 +431,7 @@ def _listdir(
         root, dav_path = build_listdir_req(user, path, properties, prop_type)
         webdav_response = self._session.adapter_dav.request(
             "PROPFIND",
-            dav_path,
+            quote(dav_path),
             content=element_tree_as_str(root),
             headers={"Depth": "infinity" if depth == -1 else str(depth)},
         )
@@ -440,16 +440,17 @@ def _listdir(
         )
 
     def __download2stream(self, path: str, fp, **kwargs) -> None:
-        with self._session.adapter_dav.stream("GET", dav_get_obj_path(self._session.user, path)) as response:
+        with self._session.adapter_dav.stream("GET", quote(dav_get_obj_path(self._session.user, path))) as response:
             check_error(response, f"download_stream: user={self._session.user}, path={path}")
             for data_chunk in response.iter_raw(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
                 fp.write(data_chunk)
 
     def __upload_stream(self, path: str, fp, chunk_size: int) -> FsNode:
-        _dav_path = dav_get_obj_path(self._session.user, "nc-py-api-" + random_string(56), root_path="/uploads")
+        _tmp_path = "nc-py-api-" + random_string(56)
+        _dav_path = quote(dav_get_obj_path(self._session.user, _tmp_path, root_path="/uploads"))
         _v2 = bool(self._session.cfg.options.upload_chunk_v2 and chunk_size >= 5 * 1024 * 1024)
         full_path = dav_get_obj_path(self._session.user, path)
-        headers = Headers({"Destination": self._session.cfg.dav_endpoint + full_path}, encoding="utf-8")
+        headers = Headers({"Destination": self._session.cfg.dav_endpoint + quote(full_path)}, encoding="utf-8")
         if _v2:
             response = self._session.adapter_dav.request("MKCOL", _dav_path, headers=headers)
         else:
@@ -548,7 +549,7 @@ async def find(self, req: list, path: str | FsNode = "") -> list[FsNode]:
     async def download(self, path: str | FsNode) -> bytes:
         """Downloads and returns the content of a file."""
         path = path.user_path if isinstance(path, FsNode) else path
-        response = await self._session.adapter_dav.get(dav_get_obj_path(await self._session.user, path))
+        response = await self._session.adapter_dav.get(quote(dav_get_obj_path(await self._session.user, path)))
         check_error(response, f"download: user={await self._session.user}, path={path}")
         return response.content
 
@@ -602,7 +603,7 @@ async def upload(self, path: str | FsNode, content: bytes | str) -> FsNode:
         """
         path = path.user_path if isinstance(path, FsNode) else path
         full_path = dav_get_obj_path(await self._session.user, path)
-        response = await self._session.adapter_dav.put(full_path, content=content)
+        response = await self._session.adapter_dav.put(quote(full_path), content=content)
         check_error(response, f"upload: user={await self._session.user}, path={path}, size={len(content)}")
         return FsNode(full_path.strip("/"), **etag_fileid_from_response(response))
 
@@ -683,11 +684,11 @@ async def move(self, path_src: str | FsNode, path_dest: str | FsNode, overwrite=
         full_dest_path = dav_get_obj_path(
             await self._session.user, path_dest.user_path if isinstance(path_dest, FsNode) else path_dest
         )
-        dest = self._session.cfg.dav_endpoint + full_dest_path
+        dest = self._session.cfg.dav_endpoint + quote(full_dest_path)
         headers = Headers({"Destination": dest, "Overwrite": "T" if overwrite else "F"}, encoding="utf-8")
         response = await self._session.adapter_dav.request(
             "MOVE",
-            dav_get_obj_path(await self._session.user, path_src),
+            quote(dav_get_obj_path(await self._session.user, path_src)),
             headers=headers,
         )
         check_error(response, f"move: user={await self._session.user}, src={path_src}, dest={dest}, {overwrite}")
@@ -705,11 +706,11 @@ async def copy(self, path_src: str | FsNode, path_dest: str | FsNode, overwrite=
         full_dest_path = dav_get_obj_path(
             await self._session.user, path_dest.user_path if isinstance(path_dest, FsNode) else path_dest
         )
-        dest = self._session.cfg.dav_endpoint + full_dest_path
+        dest = self._session.cfg.dav_endpoint + quote(full_dest_path)
         headers = Headers({"Destination": dest, "Overwrite": "T" if overwrite else "F"}, encoding="utf-8")
         response = await self._session.adapter_dav.request(
             "COPY",
-            dav_get_obj_path(await self._session.user, path_src),
+            quote(dav_get_obj_path(await self._session.user, path_src)),
             headers=headers,
         )
         check_error(response, f"copy: user={await self._session.user}, src={path_src}, dest={dest}, {overwrite}")
@@ -741,7 +742,7 @@ async def setfav(self, path: str | FsNode, value: int | bool) -> None:
         path = path.user_path if isinstance(path, FsNode) else path
         root = build_setfav_req(value)
         webdav_response = await self._session.adapter_dav.request(
-            "PROPPATCH", dav_get_obj_path(await self._session.user, path), content=element_tree_as_str(root)
+            "PROPPATCH", quote(dav_get_obj_path(await self._session.user, path)), content=element_tree_as_str(root)
         )
         check_error(webdav_response, f"setfav: path={path}, value={value}")
 
@@ -770,7 +771,7 @@ async def trashbin_restore(self, path: str | FsNode) -> None:
         headers = Headers({"Destination": dest}, encoding="utf-8")
         response = await self._session.adapter_dav.request(
             "MOVE",
-            f"/trashbin/{await self._session.user}/{path}",
+            quote(f"/trashbin/{await self._session.user}/{path}"),
             headers=headers,
         )
         check_error(response, f"trashbin_restore: user={await self._session.user}, src={path}, dest={dest}")
@@ -782,7 +783,7 @@ async def trashbin_delete(self, path: str | FsNode, not_fail=False) -> None:
         :param not_fail: if set to ``True`` and the object is not found, it does not raise an exception.
         """
         path = path.user_path if isinstance(path, FsNode) else path
-        response = await self._session.adapter_dav.delete(f"/trashbin/{await self._session.user}/{path}")
+        response = await self._session.adapter_dav.delete(quote(f"/trashbin/{await self._session.user}/{path}"))
         if response.status_code == 404 and not_fail:
             return
         check_error(response)
@@ -900,7 +901,7 @@ async def _listdir(
         root, dav_path = build_listdir_req(user, path, properties, prop_type)
         webdav_response = await self._session.adapter_dav.request(
             "PROPFIND",
-            dav_path,
+            quote(dav_path),
             content=element_tree_as_str(root),
             headers={"Depth": "infinity" if depth == -1 else str(depth)},
         )
@@ -910,17 +911,18 @@ async def _listdir(
 
     async def __download2stream(self, path: str, fp, **kwargs) -> None:
         async with self._session.adapter_dav.stream(
-            "GET", dav_get_obj_path(await self._session.user, path)
+            "GET", quote(dav_get_obj_path(await self._session.user, path))
         ) as response:
             check_error(response, f"download_stream: user={await self._session.user}, path={path}")
             async for data_chunk in response.aiter_raw(chunk_size=kwargs.get("chunk_size", 5 * 1024 * 1024)):
                 fp.write(data_chunk)
 
     async def __upload_stream(self, path: str, fp, chunk_size: int) -> FsNode:
-        _dav_path = dav_get_obj_path(await self._session.user, "nc-py-api-" + random_string(56), root_path="/uploads")
+        _tmp_path = "nc-py-api-" + random_string(56)
+        _dav_path = quote(dav_get_obj_path(await self._session.user, _tmp_path, root_path="/uploads"))
         _v2 = bool(self._session.cfg.options.upload_chunk_v2 and chunk_size >= 5 * 1024 * 1024)
         full_path = dav_get_obj_path(await self._session.user, path)
-        headers = Headers({"Destination": self._session.cfg.dav_endpoint + full_path}, encoding="utf-8")
+        headers = Headers({"Destination": self._session.cfg.dav_endpoint + quote(full_path)}, encoding="utf-8")
         if _v2:
             response = await self._session.adapter_dav.request("MKCOL", _dav_path, headers=headers)
         else:

tests/actual_tests/conftest.py

@@ -39,6 +39,7 @@ def init_filesystem_for_user(nc_any, rand_bytes):
     /test_12345_text.txt
     /test_generated_image.png                       **Favorite**
     /test_dir_tmp
+    /test_###_dir
     """
     clean_filesystem_for_user(nc_any)
     im = BytesIO()
@@ -48,6 +49,7 @@ def init_filesystem_for_user(nc_any, rand_bytes):
     nc_any.files.makedirs("/test_dir/subdir")
     nc_any.files.mkdir("/test_dir/test_empty_child_dir/")
     nc_any.files.mkdir("/test_dir_tmp")
+    nc_any.files.mkdir("/test_###_dir")
 
     def init_folder(folder: str = ""):
         nc_any.files.upload(path.join(folder, "test_empty_text.txt"), content=b"")
@@ -72,6 +74,7 @@ def clean_filesystem_for_user(nc_any):
         "test_64_bytes.bin",
         "test_12345_text.txt",
         "test_generated_image.png",
+        "test_###_dir",
     ]
     for i in clean_up_list:
         nc_any.files.delete(i, not_fail=True)