mirror of https://github.com/borgbackup/borg.git
Add tests for mandatory repository feature flags.
This commit is contained in:
parent
8783460fc3
commit
ca5ea4f53d
|
@ -25,7 +25,7 @@ from ..archiver import Archiver
|
||||||
from ..cache import Cache
|
from ..cache import Cache
|
||||||
from ..crypto import bytes_to_long, num_aes_blocks
|
from ..crypto import bytes_to_long, num_aes_blocks
|
||||||
from ..helpers import Manifest, PatternMatcher, parse_pattern, EXIT_SUCCESS, EXIT_WARNING, EXIT_ERROR, bin_to_hex, \
|
from ..helpers import Manifest, PatternMatcher, parse_pattern, EXIT_SUCCESS, EXIT_WARNING, EXIT_ERROR, bin_to_hex, \
|
||||||
get_security_dir, MAX_S
|
get_security_dir, MAX_S, MandatoryFeatureUnsupported
|
||||||
from ..key import RepoKey, KeyfileKey, Passphrase, TAMRequiredError
|
from ..key import RepoKey, KeyfileKey, Passphrase, TAMRequiredError
|
||||||
from ..keymanager import RepoIdMismatch, NotABorgKeyFile
|
from ..keymanager import RepoIdMismatch, NotABorgKeyFile
|
||||||
from ..remote import RemoteRepository, PathNotAllowed
|
from ..remote import RemoteRepository, PathNotAllowed
|
||||||
|
@ -918,6 +918,67 @@ class ArchiverTestCase(ArchiverTestCaseBase):
|
||||||
manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
|
manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
|
||||||
self.assert_equal(len(manifest.archives), 0)
|
self.assert_equal(len(manifest.archives), 0)
|
||||||
|
|
||||||
|
def add_unknown_feature(self, operation):
|
||||||
|
with Repository(self.repository_path, exclusive=True) as repository:
|
||||||
|
manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
|
||||||
|
manifest.config[b'feature_flags'] = {operation.value.encode(): {b'mandatory': [b'unknown-feature']}}
|
||||||
|
manifest.write()
|
||||||
|
repository.commit()
|
||||||
|
|
||||||
|
def cmd_raises_unknown_feature(self, args):
|
||||||
|
if self.FORK_DEFAULT:
|
||||||
|
self.cmd(*args, exit_code=EXIT_ERROR)
|
||||||
|
else:
|
||||||
|
with pytest.raises(MandatoryFeatureUnsupported) as excinfo:
|
||||||
|
self.cmd(*args)
|
||||||
|
assert excinfo.value.args == (['unknown-feature'],)
|
||||||
|
|
||||||
|
def test_unknown_feature_on_create(self):
|
||||||
|
print(self.cmd('init', self.repository_location))
|
||||||
|
self.add_unknown_feature(Manifest.Operation.WRITE)
|
||||||
|
self.cmd_raises_unknown_feature(['create', self.repository_location + '::test', 'input'])
|
||||||
|
|
||||||
|
def test_unknown_feature_on_change_passphrase(self):
|
||||||
|
print(self.cmd('init', self.repository_location))
|
||||||
|
self.add_unknown_feature(Manifest.Operation.CHECK)
|
||||||
|
self.cmd_raises_unknown_feature(['change-passphrase', self.repository_location])
|
||||||
|
|
||||||
|
def test_unknown_feature_on_read(self):
|
||||||
|
print(self.cmd('init', self.repository_location))
|
||||||
|
self.cmd('create', self.repository_location + '::test', 'input')
|
||||||
|
self.add_unknown_feature(Manifest.Operation.READ)
|
||||||
|
with changedir('output'):
|
||||||
|
self.cmd_raises_unknown_feature(['extract', self.repository_location + '::test'])
|
||||||
|
|
||||||
|
self.cmd_raises_unknown_feature(['list', self.repository_location])
|
||||||
|
self.cmd_raises_unknown_feature(['info', self.repository_location + '::test'])
|
||||||
|
|
||||||
|
def test_unknown_feature_on_rename(self):
|
||||||
|
print(self.cmd('init', self.repository_location))
|
||||||
|
self.cmd('create', self.repository_location + '::test', 'input')
|
||||||
|
self.add_unknown_feature(Manifest.Operation.CHECK)
|
||||||
|
self.cmd_raises_unknown_feature(['rename', self.repository_location + '::test', 'other'])
|
||||||
|
|
||||||
|
def test_unknown_feature_on_delete(self):
|
||||||
|
print(self.cmd('init', self.repository_location))
|
||||||
|
self.cmd('create', self.repository_location + '::test', 'input')
|
||||||
|
self.add_unknown_feature(Manifest.Operation.DELETE)
|
||||||
|
# delete of an archive raises
|
||||||
|
self.cmd_raises_unknown_feature(['delete', self.repository_location + '::test'])
|
||||||
|
self.cmd_raises_unknown_feature(['prune', '--keep-daily=3', self.repository_location])
|
||||||
|
# delete of the whole repository ignores features
|
||||||
|
self.cmd('delete', self.repository_location)
|
||||||
|
|
||||||
|
@unittest.skipUnless(has_llfuse, 'llfuse not installed')
|
||||||
|
def test_unknown_feature_on_mount(self):
|
||||||
|
self.cmd('init', self.repository_location)
|
||||||
|
self.cmd('create', self.repository_location + '::test', 'input')
|
||||||
|
self.add_unknown_feature(Manifest.Operation.READ)
|
||||||
|
mountpoint = os.path.join(self.tmpdir, 'mountpoint')
|
||||||
|
os.mkdir(mountpoint)
|
||||||
|
# XXX this might hang if it doesn't raise an error
|
||||||
|
self.cmd_raises_unknown_feature(['mount', self.repository_location + '::test', mountpoint])
|
||||||
|
|
||||||
def test_progress(self):
|
def test_progress(self):
|
||||||
self.create_regular_file('file1', size=1024 * 80)
|
self.create_regular_file('file1', size=1024 * 80)
|
||||||
self.cmd('init', self.repository_location)
|
self.cmd('init', self.repository_location)
|
||||||
|
|
Loading…
Reference in New Issue