Archiver folder - all tests converted from unittest to pytest

conftest.py

@@ -88,21 +88,20 @@ def set_env_variables():
 class ArchiverSetup:
     EXE: str = None  # python source based
     FORK_DEFAULT = False
-    prefix = ""
     BORG_EXES = []
 
     def __init__(self):
         self.archiver = None
-        self.tmpdir = str
-        self.repository_path = str
-        self.repository_location = str
-        self.input_path = str
-        self.output_path = str
-        self.keys_path = str
-        self.cache_path = str
-        self.exclude_file_path = str
-        self.patterns_file_path = str
-        self.old_wd = str
+        self.tmpdir: str
+        self.prefix = ""
+        self.repository_path: str
+        self.repository_location: str
+        self.input_path: str
+        self.output_path: str
+        self.keys_path: str
+        self.cache_path: str
+        self.exclude_file_path: str
+        self.patterns_file_path: str
 
 
 @pytest.fixture()
@@ -129,10 +128,10 @@ def archiver(tmp_path, set_env_variables):
         fd.write(b"input/file2\n# A comment line, then a blank line\n\n")
     with open(archiver.patterns_file_path, "wb") as fd:
         fd.write(b"+input/file_important\n- input/file*\n# A comment line, then a blank line\n\n")
-    archiver.old_wd = os.getcwd()
+    old_wd = os.getcwd()
     os.chdir(archiver.tmpdir)
     yield archiver
-    os.chdir(archiver.old_wd)
+    os.chdir(old_wd)
 
 
 @pytest.fixture()

src/borg/testsuite/archiver/__init__.py

@@ -89,7 +89,7 @@ def exec_cmd(*args, archiver=None, fork=False, exe=None, input=b"", binary_outpu
             try:
                 ret = archiver.run(args)  # calls setup_logging internally
             finally:
-                flush_logging()  # usually done at exit, but we do not exit here
+                flush_logging()  # usually done via atexit, but we do not exit here
             output_text.flush()
             return ret, output.getvalue() if binary_output else output.getvalue().decode()
         finally:
@@ -119,6 +119,12 @@ def exec_fn(*args, **kw):
     return exec_fn
 
 
+def pytest_generate_tests(metafunc):
+    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
+    if "archivers" in metafunc.fixturenames:
+        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+
+
 def checkts(ts):
     # check if the timestamp is in the expected format
     assert datetime.strptime(ts, ISO_FORMAT + "%z")  # must not raise
@@ -159,12 +165,10 @@ def open_archive(repo_path, name):
 
 
 def open_repository(archiver):
-    if archiver.prefix == "":
-        return Repository(archiver.repository_path, exclusive=True)
-    elif archiver.prefix == "ssh://__testsuite__":
+    if archiver.repository_location.startswith("ssh://__testsuite__"):
         return RemoteRepository(Location(archiver.repository_location))
     else:
-        raise ValueError(f"Archiver prefix '{archiver.prefix}' is not a valid prefix.")
+        return Repository(archiver.repository_path, exclusive=True)
 
 
 def create_regular_file(input_path, name, size=0, contents=None):

src/borg/testsuite/archiver/check_cmd.py

@@ -9,26 +9,20 @@
 from ...helpers import msgpack
 from ...manifest import Manifest
 from ...repository import Repository
-from . import src_file
-from . import cmd, create_src_archive, open_archive, RK_ENCRYPTION
+from . import pytest_generate_tests  # NOQA
+from . import cmd, src_file, create_src_archive, open_archive, RK_ENCRYPTION
 
 
-def check_cmd_setUp(archiver):
+def check_cmd_setup(archiver):
     with patch.object(ChunkBuffer, "BUFFER_SIZE", 10):
         cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
         create_src_archive(archiver, "archive1")
         create_src_archive(archiver, "archive2")
 
 
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
-
-
 def test_check_usage(archivers, request):
     archiver = request.getfixturevalue(archivers)
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     repo_location = archiver.repository_location
 
     output = cmd(archiver, f"--repo={repo_location}", "check", "-v", "--progress", exit_code=0)
@@ -61,7 +55,7 @@ def test_check_usage(archivers, request):
 
 def test_date_matching(archivers, request):
     archiver = request.getfixturevalue(archivers)
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
 
     shutil.rmtree(repo_path)
@@ -102,7 +96,7 @@ def test_date_matching(archivers, request):
 def test_missing_file_chunk(archivers, request):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
 
     archive, repository = open_archive(repo_path, "archive1")
 
@@ -165,7 +159,7 @@ def test_missing_file_chunk(archivers, request):
 def test_missing_archive_item_chunk(archivers, request):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     archive, repository = open_archive(repo_path, "archive1")
 
     with repository:
@@ -180,7 +174,7 @@ def test_missing_archive_item_chunk(archivers, request):
 def test_missing_archive_metadata(archivers, request):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     archive, repository = open_archive(repo_path, "archive1")
 
     with repository:
@@ -195,7 +189,7 @@ def test_missing_archive_metadata(archivers, request):
 def test_missing_manifest(archivers, request):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     archive, repository = open_archive(repo_path, "archive1")
 
     with repository:
@@ -212,7 +206,7 @@ def test_missing_manifest(archivers, request):
 def test_corrupted_manifest(archivers, request):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     archive, repository = open_archive(repo_path, "archive1")
 
     with repository:
@@ -231,7 +225,7 @@ def test_corrupted_manifest(archivers, request):
 def test_manifest_rebuild_corrupted_chunk(archivers, request):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     archive, repository = open_archive(repo_path, "archive1")
 
     with repository:
@@ -252,7 +246,7 @@ def test_manifest_rebuild_corrupted_chunk(archivers, request):
 def test_manifest_rebuild_duplicate_archive(archivers, request):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     archive, repository = open_archive(repo_path, "archive1")
     repo_objs = archive.repo_objs
 
@@ -288,7 +282,7 @@ def test_extra_chunks(archivers, request):
     if archiver.prefix:
         pytest.skip("only works locally")
     repo_location = archiver.repository_location
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     cmd(archiver, f"--repo={repo_location}", "check", exit_code=0)
 
     with Repository(repo_location, exclusive=True) as repository:
@@ -306,7 +300,7 @@ def test_extra_chunks(archivers, request):
 def test_verify_data(archivers, request, init_args):
     archiver = request.getfixturevalue(archivers)
     repo_location, repo_path = archiver.repository_location, archiver.repository_path
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
     shutil.rmtree(repo_path)
     cmd(archiver, f"--repo={repo_location}", "rcreate", *init_args)
     create_src_archive(archiver, "archive1")
@@ -337,7 +331,7 @@ def test_empty_repository(archivers, request):
     if archiver.prefix:
         pytest.skip("only works locally")
     repo_location = archiver.repository_location
-    check_cmd_setUp(archiver)
+    check_cmd_setup(archiver)
 
     with Repository(repo_location, exclusive=True) as repository:
         for id_ in repository.list():

src/borg/testsuite/archiver/checks.py

@@ -43,7 +43,7 @@ def cmd_raises_unknown_feature(archiver, args):
 
 
 def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
+    # Generate tests for local and remote archivers
     if "archivers" in metafunc.fixturenames:
         metafunc.parametrize("archivers", ["archiver", "remote_archiver"])
 

src/borg/testsuite/archiver/create_cmd.py

@@ -27,7 +27,7 @@
     same_ts_ns,
     is_root,
 )
-
+from . import pytest_generate_tests  # NOQA
 from . import (
     cmd,
     create_test_files,
@@ -44,12 +44,6 @@
 )
 
 
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
-
-
 def test_basic_functionality(archivers, request):
     archiver = request.getfixturevalue(archivers)
     if archiver.EXE:

src/borg/testsuite/archiver/debug_cmds.py

@@ -6,12 +6,7 @@
 from .. import changedir
 from ..compress import Compressor
 from . import cmd, create_test_files, create_regular_file, RK_ENCRYPTION
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import pytest_generate_tests  # NOQA
 
 
 def test_debug_profile(archivers, request):

src/borg/testsuite/archiver/delete_cmd.py

@@ -2,13 +2,8 @@
 from ...constants import *  # NOQA
 from ...manifest import Manifest
 from ...repository import Repository
-from . import cmd, create_regular_file, RK_ENCRYPTION, src_file, create_src_archive
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import cmd, create_regular_file, src_file, create_src_archive, RK_ENCRYPTION
+from . import pytest_generate_tests  # NOQA
 
 
 def test_delete(archivers, request):

src/borg/testsuite/archiver/diff_cmd.py

@@ -7,12 +7,7 @@
 from .. import are_symlinks_supported, are_hardlinks_supported
 from ..platform import is_win32, is_darwin
 from . import cmd, create_regular_file, RK_ENCRYPTION, assert_line_exists
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import pytest_generate_tests  # NOQA
 
 
 def test_basic_functionality(archivers, request):

src/borg/testsuite/archiver/extract_cmd.py

@@ -14,6 +14,7 @@
 from .. import changedir, same_ts_ns
 from .. import are_symlinks_supported, are_hardlinks_supported, is_utime_fully_supported, is_birthtime_fully_supported
 from ..platform import is_darwin, is_win32
+from . import pytest_generate_tests  # NOQA
 from . import (
     RK_ENCRYPTION,
     requires_hardlinks,
@@ -26,12 +27,6 @@
 )
 
 
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
-
-
 @pytest.mark.skipif(not are_symlinks_supported(), reason="symlinks not supported")
 def test_symlink_extract(archivers, request):
     archiver = request.getfixturevalue(archivers)

src/borg/testsuite/archiver/info_cmd.py

@@ -2,13 +2,8 @@
 import os
 
 from ...constants import *  # NOQA
-from . import cmd, RK_ENCRYPTION, checkts, create_regular_file
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import cmd, checkts, create_regular_file, RK_ENCRYPTION
+from . import pytest_generate_tests  # NOQA
 
 
 def test_info(archivers, request):

src/borg/testsuite/archiver/key_cmds.py

@@ -13,12 +13,7 @@
 from .. import environment_variable
 from .. import key
 from . import RK_ENCRYPTION, KF_ENCRYPTION, cmd, _extract_repository_id, _set_repository_id
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import pytest_generate_tests  # NOQA
 
 
 def test_change_passphrase(archivers, request):

src/borg/testsuite/archiver/list_cmd.py

@@ -3,12 +3,7 @@
 
 from ...constants import *  # NOQA
 from . import src_dir, cmd, create_regular_file, RK_ENCRYPTION
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import pytest_generate_tests  # NOQA
 
 
 def test_list_format(archivers, request):

src/borg/testsuite/archiver/lock_cmds.py

@@ -2,12 +2,7 @@
 
 from ...constants import *  # NOQA
 from . import cmd, RK_ENCRYPTION
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import pytest_generate_tests  # NOQA
 
 
 def test_break_lock(archivers, request):

src/borg/testsuite/archiver/mount_cmds.py

@@ -13,14 +13,9 @@
 from .. import changedir, no_selinux, same_ts_ns
 from .. import are_symlinks_supported, are_hardlinks_supported, are_fifos_supported
 from ..platform import fakeroot_detected
-from . import RK_ENCRYPTION, cmd, assert_dirs_equal, create_regular_file, create_src_archive, open_archive
-from . import src_file, requires_hardlinks, _extract_hardlinks_setup, fuse_mount, create_test_files
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import RK_ENCRYPTION, cmd, assert_dirs_equal, create_regular_file, create_src_archive, open_archive, src_file
+from . import requires_hardlinks, _extract_hardlinks_setup, fuse_mount, create_test_files
+from . import pytest_generate_tests  # NOQA
 
 
 @requires_hardlinks

src/borg/testsuite/archiver/prune_cmd.py

@@ -3,12 +3,7 @@
 
 from ...constants import *  # NOQA
 from . import cmd, RK_ENCRYPTION, src_dir
-
-
-def pytest_generate_tests(metafunc):
-    # Generate tests for different scenarios: local repository, remote repository, and using the borg binary.
-    if "archivers" in metafunc.fixturenames:
-        metafunc.parametrize("archivers", ["archiver", "remote_archiver", "binary_archiver"])
+from . import pytest_generate_tests  # NOQA
 
 
 def _create_archive_ts(archiver, name, y, m, d, H=0, M=0, S=0):