disk_full converted

conftest.py

@@ -2,7 +2,7 @@ import os
 
 import pytest
 
-from borg.testsuite.archiver import exec_cmd
+from borg.testsuite.archiver import BORG_EXES
 
 if hasattr(pytest, "register_assert_rewrite"):
     pytest.register_assert_rewrite("borg.testsuite")
@@ -102,7 +102,7 @@ class ArchiverSetup:
         self.cache_path = str
         self.exclude_file_path = str
         self.patterns_file_path = str
-        self._old_wd = str
+        self.old_wd = str
 
 
 @pytest.fixture()
@@ -129,31 +129,22 @@ 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()
+    archiver.old_wd = os.getcwd()
     os.chdir(archiver.tmpdir)
     yield archiver
-    os.chdir(archiver._old_wd)
+    os.chdir(archiver.old_wd)
 
 
 @pytest.fixture()
 def remote_archiver(archiver):
     archiver.prefix = "ssh://__testsuite__"
-    archiver.repository_location = archiver.prefix + archiver.repository_path
-    return archiver
-
-
-@pytest.fixture()
-def check_binary_availability(archiver):
-    try:
-        exec_cmd("help", exe="borg.exe", fork=True)
-        archiver.BORG_EXES = ["python", "binary"]
-    except FileNotFoundError:
-        archiver.BORG_EXES = ["python"]
+    archiver.repository_location = archiver.prefix + str(archiver.repository_path)
+    yield archiver
 
 
 @pytest.fixture()
-def binary_archiver(archiver, check_binary_availability):
-    if "binary" not in archiver.BORG_EXES:
+def binary_archiver(archiver):
+    if "binary" not in BORG_EXES:
         pytest.skip("No borg.exe binary available")
     archiver.EXE = "borg.exe"
     archiver.FORK_DEFAULT = True

src/borg/testsuite/archiver/__init__.py

@@ -6,6 +6,7 @@ import re
 import stat
 import subprocess
 import sys
+import tempfile
 import time
 from configparser import ConfigParser
 from contextlib import contextmanager
@@ -95,6 +96,29 @@ def exec_cmd(*args, archiver=None, fork=False, exe=None, input=b"", binary_outpu
             sys.stdin, sys.stdout, sys.stderr = stdin, stdout, stderr
 
 
+# check if the binary "borg.exe" is available (for local testing a symlink to virtualenv/bin/borg should do)
+try:
+    exec_cmd("help", exe="borg.exe", fork=True)
+    BORG_EXES = ["python", "binary"]
+except FileNotFoundError:
+    BORG_EXES = ["python"]
+
+
+@pytest.fixture(params=BORG_EXES)
+def cmd_fixture(request):
+    if request.param == "python":
+        exe = None
+    elif request.param == "binary":
+        exe = "borg.exe"
+    else:
+        raise ValueError("param must be 'python' or 'binary'")
+
+    def exec_fn(*args, **kw):
+        return exec_cmd(*args, exe=exe, fork=True, **kw)
+
+    return exec_fn
+
+
 def checkts(ts):
     # check if the timestamp is in the expected format
     assert datetime.strptime(ts, ISO_FORMAT + "%z")  # must not raise

src/borg/testsuite/archiver/disk_full.py

@@ -20,13 +20,14 @@ import shutil
 import pytest
 
 from ...constants import *  # NOQA
-from . import cmd, environment_variable
+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):
+def test_disk_full(cmd_fixture):
     def make_files(dir, count, size, rnd=True):
         shutil.rmtree(dir, ignore_errors=True)
         os.mkdir(dir)
@@ -51,7 +52,7 @@ def test_disk_full(cmd):
             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(f"--repo={repo}", "rcreate")
+            rc, out = cmd_fixture(f"--repo={repo}", "rcreate")
             if rc != EXIT_SUCCESS:
                 print("rcreate", rc, out)
             assert rc == EXIT_SUCCESS
@@ -67,7 +68,7 @@ def test_disk_full(cmd):
                             break
                         raise
                     try:
-                        rc, out = cmd("--repo=%s" % repo, "create", "test%03d" % i, input)
+                        rc, out = cmd_fixture("--repo=%s" % repo, "create", "test%03d" % i, input)
                         success = rc == EXIT_SUCCESS
                         if not success:
                             print("create", rc, out)
@@ -77,12 +78,12 @@ def test_disk_full(cmd):
                         os.remove(os.path.join(repo, "lock.roster"))
             finally:
                 # now some error happened, likely we are out of disk space.
-                # free some space so we can expect borg to be able to work normally:
+                # free some space such that we can expect borg to be able to work normally:
                 shutil.rmtree(reserve, ignore_errors=True)
-            rc, out = cmd(f"--repo={repo}", "rlist")
+            rc, out = cmd_fixture(f"--repo={repo}", "rlist")
             if rc != EXIT_SUCCESS:
                 print("rlist", rc, out)
-            rc, out = cmd(f"--repo={repo}", "check", "--repair")
+            rc, out = cmd_fixture(f"--repo={repo}", "check", "--repair")
             if rc != EXIT_SUCCESS:
                 print("check", rc, out)
             assert rc == EXIT_SUCCESS