98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
import argparse
|
|
import pytest
|
|
import subprocess
|
|
|
|
from unittest import mock
|
|
|
|
from locutus.umount import run_umount
|
|
|
|
|
|
def make_args(mountpoint):
|
|
args = argparse.Namespace()
|
|
args.mountpoint = mountpoint
|
|
return args
|
|
|
|
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run')
|
|
def test_umount_borg_success(mock_run, mock_isdir, capsys):
|
|
mock_run.return_value.returncode = 0
|
|
args = make_args('/mnt/test')
|
|
rc = run_umount(args)
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert 'Unmounted /mnt/test' in out
|
|
cmd_args = mock_run.call_args[0][0]
|
|
assert cmd_args == ['borg', 'umount', '/mnt/test']
|
|
|
|
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run')
|
|
def test_umount_fusermount_success(mock_run, mock_isdir, capsys):
|
|
# Simulate borg umount fails, fusermount -u succeeds
|
|
def side_effect(cmd, **kwargs):
|
|
if cmd[0] == 'borg':
|
|
raise subprocess.CalledProcessError(1, cmd)
|
|
return mock.Mock(returncode=0)
|
|
|
|
mock_run.side_effect = side_effect
|
|
|
|
args = make_args('/mnt/backup')
|
|
rc = run_umount(args)
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert 'Unmounted /mnt/backup' in out
|
|
# Second call is fusermount -u
|
|
assert mock_run.call_args_list[1][0][0] == [
|
|
'fusermount',
|
|
'-u',
|
|
'/mnt/backup',
|
|
]
|
|
|
|
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
|
def test_umount_all_fail(mock_run, mock_isdir, capsys):
|
|
args = make_args('/mnt/doesnotunmount')
|
|
rc = run_umount(args)
|
|
assert rc == 2
|
|
err = capsys.readouterr().err
|
|
assert 'Failed to unmount' in err
|
|
|
|
|
|
@mock.patch('os.path.isdir', return_value=False)
|
|
def test_umount_bad_mountpoint(mock_isdir, capsys):
|
|
args = make_args('/not/a/dir')
|
|
rc = run_umount(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'does not exist or is not a directory' in err
|
|
|
|
|
|
def test_umount_usage_message(capsys):
|
|
args = argparse.Namespace()
|
|
rc = run_umount(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'Usage:' in err
|
|
|
|
|
|
@mock.patch('os.path.isdir', return_value=True)
|
|
@mock.patch('subprocess.run')
|
|
def test_umount_fusermount_inner_exception(mock_run, mock_isdir, capsys):
|
|
# borg umount fails (CalledProcessError), fusermount -u raises generic Exception
|
|
def side_effect(cmd, **kwargs):
|
|
if cmd[0] == 'borg':
|
|
raise subprocess.CalledProcessError(1, cmd)
|
|
if cmd[0] == 'fusermount':
|
|
raise Exception('fusermount explosion!')
|
|
return mock.Mock(returncode=0)
|
|
|
|
mock_run.side_effect = side_effect
|
|
|
|
args = make_args('/mnt/boom')
|
|
rc = run_umount(args)
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert 'Failed to unmount /mnt/boom: fusermount explosion!' in err
|