Add test cases increasing coverage in benchmark, lock, debug, create, and check cmds

src/borg/testsuite/archiver/benchmark_cmd_test.py

@@ -6,3 +6,54 @@ def test_benchmark_crud(archiver, monkeypatch):
     cmd(archiver, "repo-create", RK_ENCRYPTION)
     monkeypatch.setenv("_BORG_BENCHMARK_CRUD_TEST", "YES")
     cmd(archiver, "benchmark", "crud", archiver.input_path)
+
+def test_benchmark_crud_info_progress_logjson_lockwait(archiver, monkeypatch):
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    monkeypatch.setenv("_BORG_BENCHMARK_CRUD_TEST", "YES")
+
+
+    cmd(
+        archiver,
+        "benchmark",
+        "--info",
+        "--progress",
+        "--log-json",
+        "--lock-wait", "10",
+        "crud",
+        archiver.input_path,
+    )
+
+def test_benchmark_cpu(archiver):
+    cmd(archiver, "benchmark", "cpu")
+
+def test_benchmark_crud_full_tests(archiver, monkeypatch):
+    """Test that the full benchmark test suite is defined when not in test mode."""
+    # Ensure the environment variable is NOT set, so we hit lines 106-113
+    monkeypatch.delenv("_BORG_BENCHMARK_CRUD_TEST", raising=False)
+
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    # We'll run the benchmark, but since it will execute the full tests (which take forever),
+    # we only do this to ensure the code path is covered.
+    # The actual benchmark will run, so this might take a bit longer.
+    cmd(archiver, "benchmark", "crud", archiver.input_path)
+
+def test_benchmark_crud_remote_options(archiver, monkeypatch):
+    """Test benchmark crud with --rsh and --remote-path options to cover lines 24 and 26."""
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    monkeypatch.setenv("_BORG_BENCHMARK_CRUD_TEST", "YES")
+
+    # Test with --rsh option (covers line 24)
+    cmd(archiver, "--rsh", "ssh -o StrictHostKeyChecking=no", "benchmark", "crud", archiver.input_path)
+
+    # Test with --remote-path option (covers line 26)
+    cmd(archiver, "--remote-path", "borg", "benchmark", "crud", archiver.input_path)
+
+    # Test with both options
+    cmd(
+        archiver,
+        "--rsh", "ssh",
+        "--remote-path", "borg",
+        "benchmark",
+        "crud",
+        archiver.input_path
+    )

src/borg/testsuite/archiver/check_cmd_test.py

@@ -446,3 +446,83 @@ def test_empty_repository(archivers, request):
         for id, _ in repository.list():
             repository.delete(id)
     cmd(archiver, "check", exit_code=1)
+
+
+def test_check_repair_user_cancellation(archivers, request, monkeypatch):
+    """Test that check --repair is cancelled if user doesn't confirm with YES."""
+    from ...helpers import CancelledByUser
+
+    archiver = request.getfixturevalue(archivers)
+    check_cmd_setup(archiver)
+
+    # Simulate user entering "no" instead of "YES"
+    monkeypatch.setenv("BORG_CHECK_I_KNOW_WHAT_I_AM_DOING", "no")
+
+    with pytest.raises(CancelledByUser):
+        cmd(archiver, "check", "--repair")
+
+
+def test_check_repository_only_conflicts(archivers, request):
+    """Test that --repository-only conflicts with archive-related options."""
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    check_cmd_setup(archiver)
+
+    # --repository-only conflicts with --verify-data
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "check", "--repository-only", "--verify-data")
+    assert "contradicts" in str(exc_info.value)
+
+    # --repository-only conflicts with --match-archives
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "check", "--repository-only", "--match-archives=*")
+    assert "contradicts" in str(exc_info.value)
+
+    # --repository-only conflicts with --first
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "check", "--repository-only", "--first=1")
+    assert "contradicts" in str(exc_info.value)
+
+    # --repository-only conflicts with --last
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "check", "--repository-only", "--last=1")
+    assert "contradicts" in str(exc_info.value)
+
+
+def test_check_repository_only_find_lost_archives_conflict(archivers, request):
+    """Test that --repository-only conflicts with --find-lost-archives."""
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    check_cmd_setup(archiver)
+
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "check", "--repository-only", "--find-lost-archives")
+    assert "contradicts" in str(exc_info.value)
+    assert "--find-lost-archives" in str(exc_info.value)
+
+
+def test_check_repair_max_duration_conflict(archivers, request):
+    """Test that --repair conflicts with --max-duration."""
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    check_cmd_setup(archiver)
+
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "check", "--repair", "--max-duration=60")
+    assert "--repair does not allow --max-duration" in str(exc_info.value)
+
+
+def test_check_max_duration_requires_repository_only(archivers, request):
+    """Test that --max-duration requires --repository-only."""
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    check_cmd_setup(archiver)
+
+    # --max-duration without --repository-only should fail
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "check", "--max-duration=60")
+    assert "--repository-only is required for --max-duration" in str(exc_info.value)

src/borg/testsuite/archiver/create_cmd_test.py

@@ -16,6 +16,9 @@
 from ...platform import is_win32
 from ...repository import Repository
 from ...helpers import CommandError, BackupPermissionError
+import pytest
+from . import cmd
+from .. import has_lchflags
 from .. import has_lchflags, has_mknod
 from .. import changedir
 from .. import (
@@ -1085,3 +1088,117 @@ def test_exclude_nodump_dir_with_file(archivers, request):
     list_output = cmd(archiver, "list", "test", "--short")
     assert "input/nd\n" not in list_output
     assert "input/nd/file_in_ndir\n" not in list_output
+
+
+
+# def test_invalid_option_errors(archiver):
+#     out = cmd(archiver, "create", "--definitely-not-a-flag", exit_code=2)
+#     assert "unrecognized" in out.lower() or "error" in out.lower()
+
+
+@pytest.mark.skipif(not are_fifos_supported(), reason="FIFOs not supported")
+def test_create_read_special_fifo_direct(archivers, request):
+    """Test --read-special with a FIFO (not via symlink) to cover lines 327-336."""
+    from threading import Thread
+
+    def fifo_feeder(fn, data):
+        with open(fn, "wb") as f:
+            f.write(data)
+
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    data = b"test data from fifo" * 100
+
+    fifo_fn = os.path.join(archiver.input_path, "test_fifo")
+    os.mkfifo(fifo_fn)
+
+    t = Thread(target=fifo_feeder, args=(fifo_fn, data))
+    t.start()
+    try:
+        cmd(archiver, "create", "--read-special", "test", "input/test_fifo")
+    finally:
+        # Cleanup
+        fd = os.open(fifo_fn, os.O_RDONLY | os.O_NONBLOCK)
+        try:
+            os.read(fd, len(data))
+        except OSError:
+            pass
+        finally:
+            os.close(fd)
+        t.join()
+
+    with changedir("output"):
+        cmd(archiver, "extract", "test")
+        with open("input/test_fifo", "rb") as f:
+            extracted_data = f.read()
+    assert extracted_data == data
+
+
+@pytest.mark.skipif(not are_symlinks_supported(), reason="symlinks not supported")
+@pytest.mark.skipif(not are_fifos_supported(), reason="FIFOs not supported")
+def test_create_read_special_symlink_to_fifo_content(archivers, request):
+    """Test --read-special with symlink pointing to FIFO reads the content (lines 305-316)."""
+    from threading import Thread
+
+    def fifo_feeder(fn, data):
+        with open(fn, "wb") as f:
+            f.write(data)
+
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    data = b"special content via symlink" * 50
+
+    fifo_fn = os.path.join(archiver.input_path, "real_fifo")
+    symlink_fn = os.path.join(archiver.input_path, "link_to_fifo")
+    os.mkfifo(fifo_fn)
+    os.symlink(fifo_fn, symlink_fn)
+
+    t = Thread(target=fifo_feeder, args=(fifo_fn, data))
+    t.start()
+    try:
+        # When using --read-special on a symlink to a FIFO, it should read the FIFO content
+        cmd(archiver, "create", "--read-special", "test", "input/link_to_fifo")
+    finally:
+        # Cleanup
+        fd = os.open(fifo_fn, os.O_RDONLY | os.O_NONBLOCK)
+        try:
+            os.read(fd, len(data))
+        except OSError:
+            pass
+        finally:
+            os.close(fd)
+        t.join()
+
+    with changedir("output"):
+        cmd(archiver, "extract", "test")
+        # The extracted file should contain the FIFO content
+        with open("input/link_to_fifo", "rb") as f:
+            extracted_data = f.read()
+    assert extracted_data == data
+
+
+@pytest.mark.skipif(not is_root(), reason="need (fake)root to create device files")
+def test_create_read_special_char_device(archivers, request):
+    """Test --read-special with character device (lines 337-352)."""
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+
+    # Create a character device (like /dev/null)
+    # We'll create a device that we can actually read from
+    dev_path = os.path.join(archiver.input_path, "char_dev")
+    # Create a character device with major/minor numbers (1, 3) = /dev/null equivalent
+    os.mknod(dev_path, stat.S_IFCHR | 0o666, os.makedev(1, 3))
+
+    # Also create regular file for comparison
+    create_regular_file(archiver.input_path, "regular", contents=b"test")
+
+    # Without --read-special, should store as device
+    cmd(archiver, "create", "test1", "input")
+    output1 = cmd(archiver, "list", "test1", "--format", "{type} {path}{NL}")
+    assert "c input/char_dev" in output1  # 'c' for character device
+
+    # With --read-special, should read it as a file (will be empty since it's like /dev/null)
+    cmd(archiver, "create", "--read-special", "test2", "input/char_dev")
+    output2 = cmd(archiver, "list", "test2", "--format", "{type} {path}{NL}")
+    # When read-special is used, device content is read and stored as a file
+    assert "input/char_dev" in output2

src/borg/testsuite/archiver/debug_cmds_test.py

@@ -162,3 +162,112 @@ def test_debug_info(archivers, request):
     archiver = request.getfixturevalue(archivers)
     output = cmd(archiver, "debug", "info")
     assert "Python" in output
+
+
+def test_debug_search_repo_objs(archivers, request):
+    """Test the debug search-repo-objs command with hex and string patterns."""
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+
+    # Create test data with known content
+    test_string = "unique_test_string_12345"
+    test_data = f"some data before {test_string} some data after".encode()
+    create_regular_file(archiver.input_path, "searchable_file", contents=test_data)
+
+    # Create an archive with this data
+    cmd(archiver, "create", "test", "input")
+
+    # Search for the string pattern
+    output = cmd(archiver, "debug", "search-repo-objs", f"str:{test_string}")
+    # The search should find the string in repository objects
+    # We just verify the command completes and produces output
+    assert "Done." in output
+
+    # Search for a hex pattern (using a simple hex sequence)
+    hex_pattern = "hex:313233"  # hex for "123"
+    output = cmd(archiver, "debug", "search-repo-objs", hex_pattern)
+    assert "Done." in output
+
+
+def test_debug_search_repo_objs_invalid_pattern(archivers, request):
+    """Test search-repo-objs with an invalid search pattern."""
+    import pytest
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    cmd(archiver, "create", "test", "input")
+
+    # Try to search with invalid pattern (no hex: or str: prefix)
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "debug", "search-repo-objs", "invalid_pattern")
+
+    assert "search term needs to be hex:" in str(exc_info.value) or "str:" in str(exc_info.value)
+
+
+def test_debug_get_obj_invalid_id(archivers, request):
+    """Test get-obj with an invalid (malformed) hex ID."""
+    import pytest
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+
+    # Try to get object with invalid hex ID (not proper hex format)
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "debug", "get-obj", "not_a_valid_hex_id", "output/file")
+
+    assert "is invalid" in str(exc_info.value)
+
+
+def test_debug_get_obj_not_found(archivers, request):
+    """Test get-obj with a non-existent but valid hex ID."""
+    import pytest
+    from ...helpers import RTError
+
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+
+    # Valid hex ID format but object doesn't exist in repository
+    fake_id = "0" * 64  # 64 hex characters (32 bytes)
+
+    with pytest.raises(RTError) as exc_info:
+        cmd(archiver, "debug", "get-obj", fake_id, "output/file")
+
+    assert "not found" in str(exc_info.value)
+
+
+def test_debug_parse_obj_invalid_id(archivers, request):
+    """Test parse-obj with an invalid (malformed) hex ID."""
+    import pytest
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+
+    # Create dummy files for the command
+    create_regular_file(archiver.input_path, "object.bin", contents=b"dummy")
+
+    # Try to parse with invalid hex ID
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "debug", "parse-obj", "invalid_id", "input/object.bin", "output/data.bin", "output/meta.json")
+
+    assert "is invalid" in str(exc_info.value)
+
+
+def test_debug_put_obj_invalid_id(archivers, request):
+    """Test put-obj with an invalid (malformed) hex ID."""
+    import pytest
+    from ...helpers import CommandError
+
+    archiver = request.getfixturevalue(archivers)
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+
+    # Create test file to put
+    create_regular_file(archiver.input_path, "file", contents=b"test data")
+
+    # Try to put with invalid hex ID
+    with pytest.raises(CommandError) as exc_info:
+        cmd(archiver, "debug", "put-obj", "bad_hex_id", "input/file")
+
+    assert "is invalid" in str(exc_info.value)