Fix creation empty file if exception on backup
Change-Id: I237ed62003c811a9844c4078dfce89ba715810a7
This commit is contained in:
parent
058a3478bd
commit
4ee82368b2
@ -13,7 +13,9 @@
|
||||
import contextlib
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import tarfile
|
||||
import tempfile
|
||||
|
||||
from cliff import command
|
||||
|
||||
@ -28,10 +30,18 @@ def backup(path_to_backup, archivators):
|
||||
ext = ext[1:]
|
||||
else:
|
||||
ext = ""
|
||||
tar_obj = tarfile.open(path_to_backup, "w|{0}".format(ext))
|
||||
with contextlib.closing(tar_obj) as archive:
|
||||
for manager in archivators:
|
||||
manager(archive).backup()
|
||||
abs_path_to_backup = os.path.abspath(path_to_backup)
|
||||
prefix = ".{0}.".format(os.path.basename(abs_path_to_backup))
|
||||
dirname = os.path.dirname(abs_path_to_backup)
|
||||
with tempfile.NamedTemporaryFile(dir=dirname, prefix=prefix) as temp:
|
||||
tar_obj = tarfile.open(fileobj=temp, mode="w|{0}".format(ext))
|
||||
with contextlib.closing(tar_obj) as archive:
|
||||
for manager in archivators:
|
||||
manager(archive).backup()
|
||||
if not archive.getmembers():
|
||||
raise Exception("Nothing to backup")
|
||||
shutil.move(temp.name, abs_path_to_backup)
|
||||
temp.delete = False
|
||||
|
||||
|
||||
class BaseBackupCommand(command.Command):
|
||||
|
@ -9,6 +9,7 @@
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
|
||||
@ -43,13 +44,34 @@ def test_parser_empty(mocker, octane_app, cmd, archivators, path):
|
||||
("path.bz2", "w|bz2"),
|
||||
("path.hz2", "w|"),
|
||||
])
|
||||
def test_backup_admin_node_backup_file(mocker, path, mode):
|
||||
@pytest.mark.parametrize("empty", [True, False])
|
||||
def test_backup_admin_node_backup_file(mocker, path, mode, empty):
|
||||
manager = mocker.Mock()
|
||||
tar_obj = mocker.patch("tarfile.open")
|
||||
backup.backup(path, [manager])
|
||||
if empty:
|
||||
tar_obj.return_value.getmembers.return_value = []
|
||||
tmp_file = mocker.patch("tempfile.NamedTemporaryFile")
|
||||
tmp_file.return_value.__enter__.return_value = tmp_file
|
||||
move_mock = mocker.patch("shutil.move")
|
||||
dir_path = "/abs"
|
||||
abs_path = "{0}/{1}".format(dir_path, path)
|
||||
os_abs_path_mock = mocker.patch("os.path.abspath", return_value=abs_path)
|
||||
if empty:
|
||||
with pytest.raises(Exception) as exc:
|
||||
backup.backup(path, [manager])
|
||||
assert "Nothing to backup" == exc.value.message
|
||||
else:
|
||||
backup.backup(path, [manager])
|
||||
os_abs_path_mock.assert_called_once_with(path)
|
||||
manager.assert_called_once_with(tar_obj.return_value)
|
||||
manager.return_value.backup.assert_called_once_with()
|
||||
tmp_file.assert_called_once_with(dir=dir_path, prefix=".{0}.".format(path))
|
||||
if path is not None:
|
||||
tar_obj.assert_called_once_with(path, mode)
|
||||
tar_obj.assert_called_once_with(fileobj=tmp_file, mode=mode)
|
||||
if empty:
|
||||
assert not move_mock.called
|
||||
else:
|
||||
move_mock.assert_called_once_with(tmp_file.name, abs_path)
|
||||
else:
|
||||
tar_obj.assert_called_once_with(fileobj=sys.stdout, mode=mode)
|
||||
assert not move_mock.called
|
||||
|
Loading…
Reference in New Issue
Block a user