|
@@ -8,46 +8,49 @@ from .. import llfuse
|
|
from . import cmd, create_src_archive, RK_ENCRYPTION, read_only, fuse_mount
|
|
from . import cmd, create_src_archive, RK_ENCRYPTION, read_only, fuse_mount
|
|
|
|
|
|
|
|
|
|
-# need to convert fuse_mount and read_only from ../__init__
|
|
|
|
def test_readonly_check(archiver):
|
|
def test_readonly_check(archiver):
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
|
|
|
|
|
|
+ repo_location, repo_path = archiver.repository_location, archiver.repository_path
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rcreate", RK_ENCRYPTION)
|
|
create_src_archive(archiver, "test")
|
|
create_src_archive(archiver, "test")
|
|
- with read_only(archiver.repository_path):
|
|
|
|
|
|
+
|
|
|
|
+ with read_only(repo_path):
|
|
# verify that command normally doesn't work with read-only repo
|
|
# verify that command normally doesn't work with read-only repo
|
|
if archiver.FORK_DEFAULT:
|
|
if archiver.FORK_DEFAULT:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "check", "--verify-data", exit_code=EXIT_ERROR)
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "check", "--verify-data", exit_code=EXIT_ERROR)
|
|
else:
|
|
else:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "check", "--verify-data")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "check", "--verify-data")
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "check", "--verify-data", "--bypass-lock")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "check", "--verify-data", "--bypass-lock")
|
|
|
|
|
|
|
|
|
|
def test_readonly_diff(archiver):
|
|
def test_readonly_diff(archiver):
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
|
|
|
|
|
|
+ repo_location, repo_path = archiver.repository_location, archiver.repository_path
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rcreate", RK_ENCRYPTION)
|
|
create_src_archive(archiver, "a")
|
|
create_src_archive(archiver, "a")
|
|
create_src_archive(archiver, "b")
|
|
create_src_archive(archiver, "b")
|
|
- with read_only(archiver.repository_path):
|
|
|
|
|
|
+
|
|
|
|
+ with read_only(repo_path):
|
|
# verify that command normally doesn't work with read-only repo
|
|
# verify that command normally doesn't work with read-only repo
|
|
if archiver.FORK_DEFAULT:
|
|
if archiver.FORK_DEFAULT:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "diff", "a", "b", exit_code=EXIT_ERROR)
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "diff", "a", "b", exit_code=EXIT_ERROR)
|
|
else:
|
|
else:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "diff", "a", "b")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "diff", "a", "b")
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
- # cmd(archiver, f"--repo={archiver.repository_location}", "diff", "a", "b", "--bypass-lock")
|
|
|
|
- # Fails - ItemDiff.__init__ 'str' object has no attribute 'get'
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "diff", "a", "b", "--bypass-lock")
|
|
|
|
|
|
|
|
|
|
def test_readonly_export_tar(archiver):
|
|
def test_readonly_export_tar(archiver):
|
|
- repo_location = archiver.repository_location
|
|
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
|
|
|
|
|
|
+ repo_location, repo_path = archiver.repository_location, archiver.repository_path
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rcreate", RK_ENCRYPTION)
|
|
create_src_archive(archiver, "test")
|
|
create_src_archive(archiver, "test")
|
|
- with read_only(archiver.repository_path):
|
|
|
|
|
|
+
|
|
|
|
+ with read_only(repo_path):
|
|
# verify that command normally doesn't work with read-only repo
|
|
# verify that command normally doesn't work with read-only repo
|
|
if archiver.FORK_DEFAULT:
|
|
if archiver.FORK_DEFAULT:
|
|
cmd(archiver, f"--repo={repo_location}", "export-tar", "test", "test.tar", exit_code=EXIT_ERROR)
|
|
cmd(archiver, f"--repo={repo_location}", "export-tar", "test", "test.tar", exit_code=EXIT_ERROR)
|
|
@@ -61,69 +64,77 @@ def test_readonly_export_tar(archiver):
|
|
|
|
|
|
|
|
|
|
def test_readonly_extract(archiver):
|
|
def test_readonly_extract(archiver):
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
|
|
|
|
|
|
+ repo_location, repo_path = archiver.repository_location, archiver.repository_path
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rcreate", RK_ENCRYPTION)
|
|
create_src_archive(archiver, "test")
|
|
create_src_archive(archiver, "test")
|
|
- with read_only(archiver.repository_path):
|
|
|
|
|
|
+
|
|
|
|
+ with read_only(repo_path):
|
|
# verify that command normally doesn't work with read-only repo
|
|
# verify that command normally doesn't work with read-only repo
|
|
if archiver.FORK_DEFAULT:
|
|
if archiver.FORK_DEFAULT:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "extract", "test", exit_code=EXIT_ERROR)
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "extract", "test", exit_code=EXIT_ERROR)
|
|
else:
|
|
else:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "extract", "test")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "extract", "test")
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "extract", "test", "--bypass-lock")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "extract", "test", "--bypass-lock")
|
|
|
|
|
|
|
|
|
|
def test_readonly_info(archiver):
|
|
def test_readonly_info(archiver):
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
|
|
|
|
|
|
+ repo_location, repo_path = archiver.repository_location, archiver.repository_path
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rcreate", RK_ENCRYPTION)
|
|
create_src_archive(archiver, "test")
|
|
create_src_archive(archiver, "test")
|
|
- with read_only(archiver.repository_path):
|
|
|
|
|
|
+
|
|
|
|
+ with read_only(repo_path):
|
|
# verify that command normally doesn't work with read-only repo
|
|
# verify that command normally doesn't work with read-only repo
|
|
if archiver.FORK_DEFAULT:
|
|
if archiver.FORK_DEFAULT:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rinfo", exit_code=EXIT_ERROR)
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rinfo", exit_code=EXIT_ERROR)
|
|
else:
|
|
else:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rinfo")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rinfo")
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rinfo", "--bypass-lock")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rinfo", "--bypass-lock")
|
|
|
|
|
|
|
|
|
|
def test_readonly_list(archiver):
|
|
def test_readonly_list(archiver):
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
|
|
|
|
|
|
+ repo_location, repo_path = archiver.repository_location, archiver.repository_path
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rcreate", RK_ENCRYPTION)
|
|
create_src_archive(archiver, "test")
|
|
create_src_archive(archiver, "test")
|
|
- with read_only(archiver.repository_path):
|
|
|
|
|
|
+
|
|
|
|
+ with read_only(repo_path):
|
|
# verify that command normally doesn't work with read-only repo
|
|
# verify that command normally doesn't work with read-only repo
|
|
if archiver.FORK_DEFAULT:
|
|
if archiver.FORK_DEFAULT:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rlist", exit_code=EXIT_ERROR)
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rlist", exit_code=EXIT_ERROR)
|
|
else:
|
|
else:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rlist")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rlist")
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rlist", "--bypass-lock")
|
|
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rlist", "--bypass-lock")
|
|
|
|
|
|
|
|
|
|
@pytest.mark.skipif(not llfuse, reason="llfuse not installed")
|
|
@pytest.mark.skipif(not llfuse, reason="llfuse not installed")
|
|
def test_readonly_mount(archiver):
|
|
def test_readonly_mount(archiver):
|
|
- cmd(archiver, f"--repo={archiver.repository_location}", "rcreate", RK_ENCRYPTION)
|
|
|
|
|
|
+ repo_location, repo_path = archiver.repository_location, archiver.repository_path
|
|
|
|
+ cmd(archiver, f"--repo={repo_location}", "rcreate", RK_ENCRYPTION)
|
|
create_src_archive(archiver, "test")
|
|
create_src_archive(archiver, "test")
|
|
- with read_only(archiver.repository_path):
|
|
|
|
|
|
+
|
|
|
|
+ with read_only(repo_path):
|
|
# verify that command normally doesn't work with read-only repo
|
|
# verify that command normally doesn't work with read-only repo
|
|
if archiver.FORK_DEFAULT:
|
|
if archiver.FORK_DEFAULT:
|
|
- with fuse_mount(archiver, archiver.repository_location, exit_code=EXIT_ERROR):
|
|
|
|
|
|
+ with fuse_mount(archiver, repo_location, exit_code=EXIT_ERROR):
|
|
pass
|
|
pass
|
|
else:
|
|
else:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
with pytest.raises((LockFailed, RemoteRepository.RPCError)) as excinfo:
|
|
# self.fuse_mount always assumes fork=True, so for this test we have to set fork=False manually
|
|
# self.fuse_mount always assumes fork=True, so for this test we have to set fork=False manually
|
|
- with fuse_mount(archiver, archiver.repository_location, fork=False):
|
|
|
|
|
|
+ with fuse_mount(archiver, repo_location, fork=False):
|
|
pass
|
|
pass
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
if isinstance(excinfo.value, RemoteRepository.RPCError):
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
assert excinfo.value.exception_class == "LockFailed"
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
# verify that command works with read-only repo when using --bypass-lock
|
|
- with fuse_mount(archiver, archiver.repository_location, None, "--bypass-lock"):
|
|
|
|
|
|
+ with fuse_mount(archiver, repo_location, None, "--bypass-lock"):
|
|
pass
|
|
pass
|