mirror
/
borg
mirror of https://github.com/borgbackup/borg.git
1
0
Fork 0
Code Issues Releases Wiki Activity

tests: remove usage of environment_variable context manager 

rather use monkeypatch.setenv/delenv.
This commit is contained in:
Thomas Waldmann 2023-07-25 13:15:56 +02:00
parent 2e59a702f6
commit b8a52a4769
No known key found for this signature in database
GPG Key ID: 243ACFA951F78E01
6 changed files with 94 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/borg/testsuite/__init__.py
View File

@ -383,27 +383,6 @@ class changedir:
os.chdir(self.old)
class environment_variable:
def __init__(self, **values):
self.values = values
self.old_values = {}
def __enter__(self):
for k, v in self.values.items():
self.old_values[k] = os.environ.get(k)
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
def __exit__(self, *args, **kw):
for k, v in self.old_values.items():
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
class FakeInputs:
"""Simulate multiple user inputs, can be used as input() replacement"""

7
src/borg/testsuite/archiver/benchmark_cmd.py
View File

@ -1,9 +1,8 @@
from ...constants import * # NOQA
from .. import environment_variable
from . import cmd, RK_ENCRYPTION
def test_benchmark_crud(archiver):
def test_benchmark_crud(archiver, monkeypatch):
cmd(archiver, "rcreate", RK_ENCRYPTION)
with environment_variable(_BORG_BENCHMARK_CRUD_TEST="YES"):
cmd(archiver, "benchmark", "crud", archiver.input_path)
monkeypatch.setenv("_BORG_BENCHMARK_CRUD_TEST", "YES")
cmd(archiver, "benchmark", "crud", archiver.input_path)

47
src/borg/testsuite/archiver/checks.py
View File

@ -15,7 +15,7 @@ from ...manifest import Manifest, MandatoryFeatureUnsupported
from ...remote import RemoteRepository, PathNotAllowed
from ...repository import Repository
from .. import llfuse
from .. import changedir, environment_variable
from .. import changedir
from . import cmd, _extract_repository_id, open_repository, check_cache, create_test_files, create_src_archive
from . import _set_repository_id, create_regular_file, assert_creates_file, generate_archiver_tests, RK_ENCRYPTION
@ -123,7 +123,7 @@ def test_repository_swap_detection2_no_cache(archivers, request):
cmd(archiver, "create", "test.2", "input")
def test_repository_swap_detection_repokey_blank_passphrase(archivers, request):
def test_repository_swap_detection_repokey_blank_passphrase(archivers, request, monkeypatch):
archiver = request.getfixturevalue(archivers)
# Check that a repokey repo with a blank passphrase is considered like a plaintext repo.
create_test_files(archiver.input_path)
@ -132,30 +132,33 @@ def test_repository_swap_detection_repokey_blank_passphrase(archivers, request):
cmd(archiver, "create", "test", "input")
# Attacker replaces it with her own repository, which is encrypted but has no passphrase set
shutil.rmtree(archiver.repository_path)
with environment_variable(BORG_PASSPHRASE=""):
cmd(archiver, "rcreate", RK_ENCRYPTION)
# Delete cache & security database, AKA switch to user perspective
cmd(archiver, "rdelete", "--cache-only")
shutil.rmtree(get_security_directory(archiver.repository_path))
with environment_variable(BORG_PASSPHRASE=None):
# This is the part were the user would be tricked, e.g. she assumes that BORG_PASSPHRASE
# is set, while it isn't. Previously this raised no warning,
# since the repository is, technically, encrypted.
if archiver.FORK_DEFAULT:
cmd(archiver, "create", "test.2", "input", exit_code=EXIT_ERROR)
else:
with pytest.raises(Cache.CacheInitAbortedError):
cmd(archiver, "create", "test.2", "input")
monkeypatch.setenv("BORG_PASSPHRASE", "")
cmd(archiver, "rcreate", RK_ENCRYPTION)
# Delete cache & security database, AKA switch to user perspective
cmd(archiver, "rdelete", "--cache-only")
shutil.rmtree(get_security_directory(archiver.repository_path))
monkeypatch.delenv("BORG_PASSPHRASE")
# This is the part were the user would be tricked, e.g. she assumes that BORG_PASSPHRASE
# is set, while it isn't. Previously this raised no warning,
# since the repository is, technically, encrypted.
if archiver.FORK_DEFAULT:
cmd(archiver, "create", "test.2", "input", exit_code=EXIT_ERROR)
else:
with pytest.raises(Cache.CacheInitAbortedError):
cmd(archiver, "create", "test.2", "input")
def test_repository_move(archivers, request):
def test_repository_move(archivers, request, monkeypatch):
archiver = request.getfixturevalue(archivers)
cmd(archiver, "rcreate", RK_ENCRYPTION)
security_dir = get_security_directory(archiver.repository_path)
os.replace(archiver.repository_path, archiver.repository_path + "_new")
archiver.repository_location += "_new"
with environment_variable(BORG_RELOCATED_REPO_ACCESS_IS_OK="yes"):
cmd(archiver, "rinfo")
monkeypatch.setenv("BORG_RELOCATED_REPO_ACCESS_IS_OK", "yes")
cmd(archiver, "rinfo")
monkeypatch.delenv("BORG_RELOCATED_REPO_ACCESS_IS_OK")
with open(os.path.join(security_dir, "location")) as fd:
location = fd.read()
assert location == Location(archiver.repository_location).canonical_path()
@ -179,7 +182,7 @@ def test_security_dir_compat(archivers, request):
cmd(archiver, "rinfo")
def test_unknown_unencrypted(archivers, request):
def test_unknown_unencrypted(archivers, request, monkeypatch):
archiver = request.getfixturevalue(archivers)
cmd(archiver, "rcreate", "--encryption=none")
# Ok: repository is known
@ -196,8 +199,8 @@ def test_unknown_unencrypted(archivers, request):
else:
with pytest.raises(Cache.CacheInitAbortedError):
cmd(archiver, "rinfo")
with environment_variable(BORG_UNKNOWN_UNENCRYPTED_REPO_ACCESS_IS_OK="yes"):
cmd(archiver, "rinfo")
monkeypatch.setenv("BORG_UNKNOWN_UNENCRYPTED_REPO_ACCESS_IS_OK", "yes")
cmd(archiver, "rinfo")
def test_unknown_feature_on_create(archivers, request):

95
src/borg/testsuite/archiver/disk_full.py
View File

@ -20,14 +20,13 @@ import shutil
import pytest
from ...constants import * # NOQA
from .. import environment_variable
from . import cmd_fixture
DF_MOUNT = "/tmp/borg-mount"
@pytest.mark.skipif(not os.path.exists(DF_MOUNT), reason="needs a 16MB fs mounted on %s" % DF_MOUNT)
def test_disk_full(cmd_fixture):
def test_disk_full(cmd_fixture, monkeypatch):
def make_files(dir, count, size, rnd=True):
shutil.rmtree(dir, ignore_errors=True)
os.mkdir(dir)
@ -41,49 +40,49 @@ def test_disk_full(cmd_fixture):
data = os.urandom(size)
f.write(data)
with environment_variable(BORG_CHECK_I_KNOW_WHAT_I_AM_DOING="YES"):
mount = DF_MOUNT
assert os.path.exists(mount)
repo = os.path.join(mount, "repo")
input = os.path.join(mount, "input")
reserve = os.path.join(mount, "reserve")
for j in range(100):
shutil.rmtree(repo, ignore_errors=True)
shutil.rmtree(input, ignore_errors=True)
# keep some space and some inodes in reserve that we can free up later:
make_files(reserve, 80, 100000, rnd=False)
rc, out = cmd_fixture(f"--repo={repo}", "rcreate")
if rc != EXIT_SUCCESS:
print("rcreate", rc, out)
assert rc == EXIT_SUCCESS
try:
success, i = True, 0
while success:
i += 1
try:
make_files(input, 20, 200000)
except OSError as err:
if err.errno == errno.ENOSPC:
# already out of space
break
raise
try:
rc, out = cmd_fixture("--repo=%s" % repo, "create", "test%03d" % i, input)
success = rc == EXIT_SUCCESS
if not success:
print("create", rc, out)
finally:
# make sure repo is not locked
shutil.rmtree(os.path.join(repo, "lock.exclusive"), ignore_errors=True)
os.remove(os.path.join(repo, "lock.roster"))
finally:
# now some error happened, likely we are out of disk space.
# free some space such that we can expect borg to be able to work normally:
shutil.rmtree(reserve, ignore_errors=True)
rc, out = cmd_fixture(f"--repo={repo}", "rlist")
if rc != EXIT_SUCCESS:
print("rlist", rc, out)
rc, out = cmd_fixture(f"--repo={repo}", "check", "--repair")
if rc != EXIT_SUCCESS:
print("check", rc, out)
assert rc == EXIT_SUCCESS
monkeypatch.setenv("BORG_CHECK_I_KNOW_WHAT_I_AM_DOING", "YES")
mount = DF_MOUNT
assert os.path.exists(mount)
repo = os.path.join(mount, "repo")
input = os.path.join(mount, "input")
reserve = os.path.join(mount, "reserve")
for j in range(100):
shutil.rmtree(repo, ignore_errors=True)
shutil.rmtree(input, ignore_errors=True)
# keep some space and some inodes in reserve that we can free up later:
make_files(reserve, 80, 100000, rnd=False)
rc, out = cmd_fixture(f"--repo={repo}", "rcreate")
if rc != EXIT_SUCCESS:
print("rcreate", rc, out)
assert rc == EXIT_SUCCESS
try:
success, i = True, 0
while success:
i += 1
try:
make_files(input, 20, 200000)
except OSError as err:
if err.errno == errno.ENOSPC:
# already out of space
break
raise
try:
rc, out = cmd_fixture("--repo=%s" % repo, "create", "test%03d" % i, input)
success = rc == EXIT_SUCCESS
if not success:
print("create", rc, out)
finally:
# make sure repo is not locked
shutil.rmtree(os.path.join(repo, "lock.exclusive"), ignore_errors=True)
os.remove(os.path.join(repo, "lock.roster"))
finally:
# now some error happened, likely we are out of disk space.
# free some space such that we can expect borg to be able to work normally:
shutil.rmtree(reserve, ignore_errors=True)
rc, out = cmd_fixture(f"--repo={repo}", "rlist")
if rc != EXIT_SUCCESS:
print("rlist", rc, out)
rc, out = cmd_fixture(f"--repo={repo}", "check", "--repair")
if rc != EXIT_SUCCESS:
print("check", rc, out)
assert rc == EXIT_SUCCESS

7
src/borg/testsuite/archiver/key_cmds.py
View File

@ -10,7 +10,6 @@ from ...helpers import EXIT_ERROR
from ...helpers import bin_to_hex
from ...helpers import msgpack
from ...repository import Repository
from .. import environment_variable
from .. import key
from . import RK_ENCRYPTION, KF_ENCRYPTION, cmd, _extract_repository_id, _set_repository_id, generate_archiver_tests
@ -96,7 +95,7 @@ def test_key_export_keyfile(archivers, request):
assert key_contents2 == key_contents
def test_key_import_keyfile_with_borg_key_file(archivers, request):
def test_key_import_keyfile_with_borg_key_file(archivers, request, monkeypatch):
archiver = request.getfixturevalue(archivers)
cmd(archiver, "rcreate", KF_ENCRYPTION)
@ -109,8 +108,8 @@ def test_key_import_keyfile_with_borg_key_file(archivers, request):
os.unlink(key_file)
imported_key_file = os.path.join(archiver.output_path, "imported")
with environment_variable(BORG_KEY_FILE=imported_key_file):
cmd(archiver, "key", "import", exported_key_file)
monkeypatch.setenv("BORG_KEY_FILE", imported_key_file)
cmd(archiver, "key", "import", exported_key_file)
assert not os.path.isfile(key_file), '"borg key import" should respect BORG_KEY_FILE'
with open(imported_key_file) as fd:

33
src/borg/testsuite/archiver/rcreate_cmd.py
View File

@ -7,7 +7,6 @@ from ...helpers.errors import Error
from ...constants import * # NOQA
from ...crypto.key import FlexiKey
from ...repository import Repository
from .. import environment_variable
from . import cmd, generate_archiver_tests, RK_ENCRYPTION, KF_ENCRYPTION
pytest_generate_tests = lambda metafunc: generate_archiver_tests(metafunc, kinds="local,remote,binary") # NOQA
@ -58,24 +57,24 @@ def test_rcreate_nested_repositories(archivers, request):
cmd(archiver, "rcreate", RK_ENCRYPTION)
def test_rcreate_refuse_to_overwrite_keyfile(archivers, request):
def test_rcreate_refuse_to_overwrite_keyfile(archivers, request, monkeypatch):
# BORG_KEY_FILE=something borg rcreate should quit if "something" already exists.
# See: https://github.com/borgbackup/borg/pull/6046
archiver = request.getfixturevalue(archivers)
keyfile = os.path.join(archiver.tmpdir, "keyfile")
monkeypatch.setenv("BORG_KEY_FILE", keyfile)
original_location = archiver.repository_location
with environment_variable(BORG_KEY_FILE=keyfile):
archiver.repository_location = original_location + "0"
cmd(archiver, "rcreate", KF_ENCRYPTION)
with open(keyfile) as file:
before = file.read()
archiver.repository_location = original_location + "1"
arg = ("rcreate", KF_ENCRYPTION)
if archiver.FORK_DEFAULT:
cmd(archiver, *arg, exit_code=2)
else:
with pytest.raises(Error):
cmd(archiver, *arg)
with open(keyfile) as file:
after = file.read()
assert before == after
archiver.repository_location = original_location + "0"
cmd(archiver, "rcreate", KF_ENCRYPTION)
with open(keyfile) as file:
before = file.read()
archiver.repository_location = original_location + "1"
arg = ("rcreate", KF_ENCRYPTION)
if archiver.FORK_DEFAULT:
cmd(archiver, *arg, exit_code=2)
else:
with pytest.raises(Error):
cmd(archiver, *arg)
with open(keyfile) as file:
after = file.read()
assert before == after