file url fixes

src/borg/legacyrepository.py

@@ -5,6 +5,7 @@
 import stat
 import struct
 import time
+from pathlib import Path
 from collections import defaultdict
 from configparser import ConfigParser
 from functools import partial
@@ -190,8 +191,9 @@ class PathPermissionDenied(Error):
         exit_mcode = 21
 
     def __init__(self, path, create=False, exclusive=False, lock_wait=None, lock=True, send_log_cb=None):
-        self.path = os.path.abspath(path)
-        self._location = Location("file://%s" % self.path)
+        p = Path(path).absolute()
+        self.path = str(p)
+        self._location = Location(p.as_uri())
         self.version = None
         # long-running repository methods which emit log or progress output are responsible for calling
         # the ._send_log method periodically to get log and progress output transferred to the borg client

src/borg/repository.py

@@ -1,6 +1,6 @@
 import os
-import sys
 import time
+from pathlib import Path
 
 from borgstore.store import Store
 from borgstore.store import ObjectNotFound as StoreObjectNotFound
@@ -106,11 +106,11 @@ def __init__(
         if isinstance(path_or_location, Location):
             location = path_or_location
             if location.proto == "file":
-                url = _local_abspath_to_file_url(location.path)  # frequently users give without file:// prefix
+                url = Path(location.path).as_uri()  # frequently users give without file:// prefix
             else:
                 url = location.processed  # location as given by user, processed placeholders
         else:
-            url = _local_abspath_to_file_url(os.path.abspath(path_or_location))
+            url = Path(path_or_location).absolute().as_uri()
             location = Location(url)
         self._location = location
         self.url = url
@@ -566,16 +566,3 @@ def store_delete(self, name, *, deleted=False):
     def store_move(self, name, new_name=None, *, delete=False, undelete=False, deleted=False):
         self._lock_refresh()
         return self.store.move(name, new_name, delete=delete, undelete=undelete, deleted=deleted)
-
-
-def _local_abspath_to_file_url(path: str) -> str:
-    """Create a file URL from a local, absolute path.
-
-    Expects `path` to be an absolute path on the local filesystem, e.g.:
-    - POSIX: `/foo/bar`
-    - Windows: `c:/foo/bar` (or `c:\foo\bar`)
-    The easiest way to ensure this is for the caller to pass `path` through `os.path.abspath` first.
-    """
-    if sys.platform in ("win32", "msys", "cygwin"):
-        path = "/" + path.replace("\\", "/")
-    return "file://%s" % path

src/borg/testsuite/archiver/lock_cmds_test.py

@@ -1,6 +1,7 @@
 import os
 import subprocess
 import time
+from pathlib import Path
 
 from ...constants import *  # NOQA
 from . import cmd, generate_archiver_tests, RK_ENCRYPTION
@@ -18,7 +19,7 @@ def test_break_lock(archivers, request):
 def test_with_lock(tmp_path):
     repo_path = tmp_path / "repo"
     env = os.environ.copy()
-    env["BORG_REPO"] = "file://" + str(repo_path)
+    env["BORG_REPO"] = Path(repo_path).as_uri()
     command0 = "python3", "-m", "borg", "repo-create", "--encryption=none"
     # Timings must be adjusted so that command1 keeps running while command2 tries to get the lock,
     # so that lock acquisition for command2 fails as the test expects it.

src/borg/testsuite/storelocking_test.py

@@ -1,4 +1,5 @@
 import time
+from pathlib import Path
 
 import pytest
 
@@ -12,7 +13,7 @@
 
 @pytest.fixture()
 def lockstore(tmpdir):
-    store = Store("file://" + str(tmpdir / "lockstore"), levels={"locks/": [0]})
+    store = Store(Path(tmpdir / "lockstore").as_uri(), levels={"locks/": [0]})
     store.create()
     with store:
         yield store