locking tests: reuse _create_lock and the fslocking free_pid fixture, see #9870

write_raw_lock duplicated _create_lock's wire format (field layout, timestamp
format, sha256 key, store path), so a future format change (e.g. AEAD lock
objects) would have made the skew tests silently keep writing the old format.
_create_lock gained an explicit content timestamp parameter instead; the
helper keeps only the store-side mtime override. The free_pid fixture was a
verbatim copy of the one in fslocking_test.py - import it like
platform_test.py does (incl. the same per-file F811 ignore).

Co-Authored-By: Claude Fable 5 <[email protected]>

pyproject.toml

@@ -155,6 +155,7 @@ dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
 "src/borg/testsuite/archiver/return_codes_test.py" = ["F811"]
 "src/borg/testsuite/benchmark_test.py" = ["F811"]
 "src/borg/testsuite/platform/platform_test.py" = ["F811"]
+"src/borg/testsuite/storelocking_test.py" = ["F811"]
 
 [tool.pytest.ini_options]
 markers = []

src/borg/storelocking.py

@@ -109,9 +109,10 @@ class Lock:
     def __repr__(self):
         return f"<{self.__class__.__name__}: {self.id!r}>"
 
-    def _create_lock(self, *, exclusive=None, update_last_refresh=False):
+    def _create_lock(self, *, exclusive=None, dt=None, update_last_refresh=False):
         assert exclusive is not None
-        now = datetime.datetime.now(datetime.UTC)
+        # dt: explicit content timestamp (default: now) - tests use it to simulate skewed clocks.
+        now = dt if dt is not None else datetime.datetime.now(datetime.UTC)
         timestamp = now.isoformat(timespec="milliseconds")
         lock = dict(exclusive=exclusive, hostid=self.id[0], processid=self.id[1], threadid=self.id[2], time=timestamp)
         value = json.dumps(lock).encode("utf-8")

src/borg/testsuite/storelocking_test.py

@@ -1,8 +1,5 @@
 import datetime
-import hashlib
-import json
 import os
-import random
 import time
 from pathlib import Path
 
@@ -10,24 +7,14 @@ import pytest
 
 from borgstore.store import ObjectNotFound, Store
 
-from ..platform import get_process_id, process_alive
+from .fslocking_test import free_pid  # NOQA
+from ..platform import get_process_id
 from ..storelocking import Lock, NotLocked, LockTimeout
 
 ID1 = "foo", 1, 1
 ID2 = "bar", 2, 2
 
 
-@pytest.fixture()
-def free_pid():
-    """Return a free PID not used by any process (naturally this is racy)."""
-    host, pid, tid = get_process_id()
-    while True:
-        # PIDs are often restricted to a small range. On Linux the range >32k is by default not used.
-        pid = random.randint(33000, 65000)
-        if not process_alive(host, pid, tid):
-            return pid
-
-
 @pytest.fixture()
 def lockstore(tmp_path):
     store = Store(Path(tmp_path / "lockstore").as_uri(), config={"locks/": {"levels": [0]}})
@@ -39,15 +26,11 @@ def lockstore(tmp_path):
 
 def write_raw_lock(store, id, *, exclusive, dt, mtime=None):
     """
-    Write a lock object like Lock._create_lock does, but with an arbitrary content timestamp <dt>
-    (simulating a writer whose clock is skewed against ours). The object's store-side mtime is
-    "now" (a real store write), unless <mtime> is given (then the file's mtime is set to it).
+    Write a lock object with an arbitrary content timestamp <dt> (simulating a writer whose clock
+    is skewed against ours). The object's store-side mtime is "now" (a real store write), unless
+    <mtime> is given (then the file's mtime is set to it).
     """
-    timestamp = dt.isoformat(timespec="milliseconds")
-    lock = dict(exclusive=exclusive, hostid=id[0], processid=id[1], threadid=id[2], time=timestamp)
-    value = json.dumps(lock).encode("utf-8")
-    key = hashlib.sha256(value).hexdigest()
-    store.store(f"locks/{key}", value)
+    key = Lock(store, id=id)._create_lock(exclusive=exclusive, dt=dt)
     if mtime is not None:
         path = store.backend.base_path / "locks" / key  # posixfs, locks/ nesting levels [0]
         os.utime(path, (mtime, mtime))