Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_restore_latest_backup(self):
t = local.LocalStorage("", None, skip_prepare=True)
t.find_all = mock.Mock()
last = base.Backup(t, "host_backup", 5000)
t.find_all.return_value = [
base.Backup(t, "host_backup", 1000),
base.Backup(t, "host_backup", 2000),
base.Backup(t, "host_backup", 3000),
base.Backup(t, "host_backup", 4000),
base.Backup(t, "host_backup_f", 1000),
last
]
assert t.find_one("host_backup") == last
def test_find_latest_backup_respects_increments_timestamp(self):
test_backup = base.Backup(None, "host_backup", 5500)
increment = base.Backup(None, "host_backup", 6000, 1, test_backup)
test_backup.add_increment(increment)
t = local.LocalStorage(None, None, skip_prepare=True)
t.find_all = mock.Mock()
t.find_all.return_value = [
test_backup,
base.Backup(None, "host_backup", 2000),
base.Backup(None, "host_backup", 3000),
base.Backup(None, "host_backup", 4000),
base.Backup(None, "host_backup_f", 1000),
base.Backup(None, "host_backup", 5000),
]
assert t.find_one("host_backup") == increment
def test_info(self):
backup_dir, files_dir, work_dir = self.create_dirs()
storage = local.LocalStorage(backup_dir, work_dir)
storage.info()
def test_restore_from_date(self):
t = local.LocalStorage(None, None, skip_prepare=True)
t.find_all = mock.Mock()
backup_restore = base.Backup(None, "host_backup", 3000)
t.find_all.return_value = [
base.Backup(None, "host_backup", 1000),
base.Backup(None, "host_backup", 2000),
backup_restore,
base.Backup(None, "host_backup", 4000),
base.Backup(None, "host_backup_f", 1000),
base.Backup(None, "host_backup", 5000),
]
assert t.find_one("host_backup", 3234) == backup_restore
def test_create_backup(self):
t = local.LocalStorage(None, None, skip_prepare=True)
t.find_all = mock.Mock()
t.find_all.return_value = []
t._find_previous_backup = mock.Mock()
t._find_previous_backup.return_value = \
base.Backup(None, "host_backup", 3000, tar_meta=True)
t.create_backup("", True, 12, False, False)
def test_restart_always_level(self):
t = local.LocalStorage(None, None, skip_prepare=True)
t.find_all = mock.Mock()
t.find_all.return_value = []
backup = base.Backup(None, "host_backup", 3000, tar_meta=True)
t._find_previous_backup([backup], False, None, None, 10)
if storage_name == "swift":
client_manager = backup_args['client_manager']
storage = swift.SwiftStorage(
client_manager, container, max_segment_size)
elif storage_name == "s3":
storage = s3.S3Storage(
backup_args['access_key'],
backup_args['secret_key'],
backup_args['endpoint'],
container,
max_segment_size
)
elif storage_name == "local":
storage = local.LocalStorage(
storage_path=container,
max_segment_size=max_segment_size)
elif storage_name == "ssh":
if backup_args['ssh_password']:
storage = ssh.SshStorage(
container,
backup_args['ssh_username'],
backup_args['ssh_host'],
int(backup_args['ssh_port']),
max_segment_size=max_segment_size,
remote_pwd=backup_args['ssh_password'])
else:
storage = ssh.SshStorage(
container,
backup_args['ssh_username'],
backup_args['ssh_host'],