Merge pull request #131 from ThomasWaldmann/url-unquote

posixfs, s3, sftp: URL-unquote

Author: ThomasWaldmann

src/borgstore/backends/posixfs.py

@@ -5,6 +5,7 @@
 import os
 import re
 import sys
+from urllib.parse import unquote
 from pathlib import Path
 import shutil
 import stat
@@ -36,13 +37,14 @@ def get_file_backend(url):
         file://  # only empty host part is supported.
         (?P<path>(/.*))  # path must be an absolute path. 3rd slash is separator AND part of the path.
     """
+    # the path or drive_and_path could be URL-quoted and thus must be URL-unquoted
     if sys.platform in ("win32", "msys", "cygwin"):
         m = re.match(windows_file_regex, url, re.VERBOSE)
         if m:
-            return PosixFS(path=m["drive_and_path"])
+            return PosixFS(path=unquote(m["drive_and_path"]))
     m = re.match(file_regex, url, re.VERBOSE)
     if m:
-        return PosixFS(path=m["path"])
+        return PosixFS(path=unquote(m["path"]))
 
 
 class PosixFS(BackendBase):

src/borgstore/backends/rclone.py

@@ -62,6 +62,7 @@ def get_rclone_backend(url):
     """
     m = re.match(rclone_regex, url, re.VERBOSE)
     if m:
+        # no URL-unquote here, we just pass through the rclone remote spec "as is"
         return Rclone(path=m["path"])
 
 

src/borgstore/backends/s3.py

@@ -68,8 +68,8 @@ def get_s3_backend(url: str):
         schema = m["schema"]
         hostname = m["hostname"]
         port = m["port"]
-        bucket = m["bucket"]
-        path = m["path"]
+        bucket = m["bucket"]  # no unquote: all valid bucket characters are URL-safe
+        path = urllib.parse.unquote(m["path"])
 
         endpoint_url = None
         if schema and hostname:

src/borgstore/backends/sftp.py

@@ -3,6 +3,7 @@
 """
 
 from pathlib import Path
+from urllib.parse import unquote
 import random
 import re
 import stat
@@ -45,7 +46,12 @@ def get_sftp_backend(url):
     """
     m = re.match(sftp_regex, url, re.VERBOSE)
     if m:
-        return Sftp(username=m["username"], hostname=m["hostname"], port=int(m["port"] or "0"), path=m["path"])
+        return Sftp(
+            username=unquote(m["username"]) if m["username"] else None,
+            hostname=m["hostname"],
+            port=int(m["port"] or "0"),
+            path=unquote(m["path"]),
+        )
 
 
 class Sftp(BackendBase):

tests/test_backends.py

@@ -167,8 +167,11 @@ def get_backend_from_fixture(tested_backends, request):
     return request.getfixturevalue(tested_backends)
 
 
-POSIX_ABS_TESTCASES = [("file:///absolute/path", "/absolute/path")]
-WINDOWS_ABS_TESTCASES = [("file:///C:/absolute/path", "C:/absolute/path")]
+POSIX_ABS_TESTCASES = [("file:///absolute/path", "/absolute/path"), ("file:///absolute/my%20path", "/absolute/my path")]
+WINDOWS_ABS_TESTCASES = [
+    ("file:///C:/absolute/path", "C:/absolute/path"),
+    ("file:///C:/absolute/my%20path", "C:/absolute/my path"),
+]
 
 
 @pytest.mark.parametrize("url,path", WINDOWS_ABS_TESTCASES if os.name == "nt" else POSIX_ABS_TESTCASES)
@@ -210,6 +213,8 @@ def test_invalid_or_remote_file_url(url):
         ("sftp://username@hostname:2222//abs/path", "username", "hostname", 2222, "/abs/path"),
         ("sftp://username@hostname//abs/path", "username", "hostname", 0, "/abs/path"),
         ("sftp://hostname//abs/path", None, "hostname", 0, "/abs/path"),
+        ("sftp://username@hostname/rel/my%20path", "username", "hostname", 0, "rel/my path"),
+        ("sftp://username@hostname//abs/my%20path", "username", "hostname", 0, "/abs/my path"),
     ],
 )
 def test_sftp_url(url, username, hostname, port, path):
@@ -221,6 +226,19 @@ def test_sftp_url(url, username, hostname, port, path):
     assert backend.base_path == path
 
 
+@pytest.mark.parametrize(
+    "url,bucket,path", [("s3:/bucket/path", "bucket", "path"), ("s3:/bucket/my%20path", "bucket", "my path")]
+)
+def test_s3_url(url, bucket, path):
+    from unittest.mock import patch, MagicMock
+
+    with patch("borgstore.backends.s3.boto3", MagicMock()):
+        backend = get_s3_backend(url)
+    assert isinstance(backend, S3)
+    assert backend.bucket == bucket
+    assert backend.base_path == path + "/"
+
+
 def test_flat(tested_backends, request):
     with get_backend_from_fixture(tested_backends, request) as backend:
         k0, v0 = key(0), b"value0"