From b8a52a47696cdc363d8447356549084e98e63d51 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Tue, 25 Jul 2023 13:15:56 +0200 Subject: [PATCH] tests: remove usage of environment_variable context manager rather use monkeypatch.setenv/delenv. --- src/borg/testsuite/__init__.py | 21 ----- src/borg/testsuite/archiver/benchmark_cmd.py | 7 +- src/borg/testsuite/archiver/checks.py | 47 +++++----- src/borg/testsuite/archiver/disk_full.py | 95 ++++++++++---------- src/borg/testsuite/archiver/key_cmds.py | 7 +- src/borg/testsuite/archiver/rcreate_cmd.py | 33 ++++--- 6 files changed, 94 insertions(+), 116 deletions(-) diff --git a/src/borg/testsuite/__init__.py b/src/borg/testsuite/__init__.py index a20949c79..a88267222 100644 --- a/src/borg/testsuite/__init__.py +++ b/src/borg/testsuite/__init__.py @@ -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""" diff --git a/src/borg/testsuite/archiver/benchmark_cmd.py b/src/borg/testsuite/archiver/benchmark_cmd.py index 1ac424874..ee7451997 100644 --- a/src/borg/testsuite/archiver/benchmark_cmd.py +++ b/src/borg/testsuite/archiver/benchmark_cmd.py @@ -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) diff --git a/src/borg/testsuite/archiver/checks.py b/src/borg/testsuite/archiver/checks.py index a1444065c..7f3c977ad 100644 --- a/src/borg/testsuite/archiver/checks.py +++ b/src/borg/testsuite/archiver/checks.py @@ -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): diff --git a/src/borg/testsuite/archiver/disk_full.py b/src/borg/testsuite/archiver/disk_full.py index f6e84eac0..8bca8c87d 100644 --- a/src/borg/testsuite/archiver/disk_full.py +++ b/src/borg/testsuite/archiver/disk_full.py @@ -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 diff --git a/src/borg/testsuite/archiver/key_cmds.py b/src/borg/testsuite/archiver/key_cmds.py index 45d0cd9b6..dfddd41d9 100644 --- a/src/borg/testsuite/archiver/key_cmds.py +++ b/src/borg/testsuite/archiver/key_cmds.py @@ -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: diff --git a/src/borg/testsuite/archiver/rcreate_cmd.py b/src/borg/testsuite/archiver/rcreate_cmd.py index 96bd97cab..93f121f8f 100644 --- a/src/borg/testsuite/archiver/rcreate_cmd.py +++ b/src/borg/testsuite/archiver/rcreate_cmd.py @@ -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