upstream/borgbackup
1
0
Fork 0
mirror of https://github.com/borgbackup/borg.git synced 2026-06-11 01:41:57 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

Merge pull request #9461 from ThomasWaldmann/fix-with-lock-test-master

Browse source 
fix race condition in test_with_lock, fixes #8810
This commit is contained in:
TW 2026-03-11 13:23:36 +01:00 committed by GitHub
commit 1c75f505d6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 8 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
src/borg/testsuite/archiver/lock_cmds_test.py
View file

@ -1,7 +1,6 @@
import os
import subprocess
import sys
import time
from pathlib import Path
import pytest
@ -32,24 +31,25 @@ def test_with_lock(tmp_path):
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.
lock_wait, execution_time, startup_wait = 2, 4, 1
assert lock_wait < execution_time - startup_wait
command1 = "python3", "-c", f'import time; print("first command - acquires the lock"); time.sleep({execution_time})'
lock_wait = 2
command1 = ("python3", "-c", 'import sys; print("first command - acquires the lock", flush=True); sys.stdin.read()')
command2 = "python3", "-c", 'print("second command - should never get executed")'
borgwl = "python3", "-m", "borg", "with-lock", f"--lock-wait={lock_wait}"
popen_options = dict(stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env)
subprocess.run(command0, env=env, check=True, text=True, capture_output=True)
assert repo_path.exists()
with subprocess.Popen([*borgwl, *command1], **popen_options) as p1:
time.sleep(startup_wait) # Wait until p1 is running
p1_options = popen_options.copy()
p1_options["stdin"] = subprocess.PIPE
with subprocess.Popen([*borgwl, *command1], **p1_options) as p1:
assert "first command" in p1.stdout.readline() # Wait until p1 is running and has acquired the lock
# Now try to acquire another lock on the same repository:
with subprocess.Popen([*borgwl, *command2], **popen_options) as p2:
out, err_out = p2.communicate()
assert "second command" not in out # command2 is "locked out"
assert "Failed to create/acquire the lock" in err_out
assert p2.returncode == 73 # LockTimeout: could not acquire the lock, p1 already has it
out, err_out = p1.communicate()
assert "first command" in out # command1 was executed and had the lock
out, err_out = p1.communicate(input="") # Unblock command1 and read output
assert not err_out
assert p1.returncode == 0