Add mount command
This commit is contained in:
Alexander Wainwright
95
src/locutus/mount.py
Normal file
95
src/locutus/mount.py
Normal file
@@ -0,0 +1,95 @@
|
||||
import argparse
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
from locutus.config import LocutusConfig
|
||||
|
||||
|
||||
def run_mount(args: argparse.Namespace) -> int:
|
||||
"""Mount a backup archive (by index or name) to the specified directory
|
||||
|
||||
Args:
|
||||
args: argparse.Namespace with --config, --profile, 'archive'
|
||||
(index or name), 'mountpoint' (directory).
|
||||
|
||||
Returns:
|
||||
0 on success, 1 on error
|
||||
"""
|
||||
if not hasattr(args, 'mountpoint'):
|
||||
print('Usage: locutus mount [index|name] <mountpoint>', file=sys.stderr)
|
||||
return 1
|
||||
|
||||
cfg = LocutusConfig(args.config, args.profile)
|
||||
repo = cfg.get_repo()
|
||||
if not repo:
|
||||
print('No BORG_REPO configured.', file=sys.stderr)
|
||||
return 1
|
||||
env = {**os.environ, **cfg.env}
|
||||
|
||||
mountpoint = args.mountpoint
|
||||
if not os.path.isdir(mountpoint):
|
||||
print(
|
||||
f"Mountpoint '{mountpoint}' does not exist or is not a directory.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
|
||||
# Determine archive name (by index or direct)
|
||||
target = getattr(args, 'archive', None)
|
||||
archive_name = None
|
||||
if target is None:
|
||||
# Mount all archives
|
||||
archive_spec = repo
|
||||
else:
|
||||
try:
|
||||
index = int(target)
|
||||
# list all archives, reverse order
|
||||
cmd = ['borg', 'list', '--short', repo]
|
||||
result = subprocess.run(
|
||||
cmd, env=env, capture_output=True, text=True, check=True
|
||||
)
|
||||
archives = result.stdout.strip().splitlines()[::-1]
|
||||
if not archives:
|
||||
print('No archives found.', file=sys.stderr)
|
||||
return 1
|
||||
if not (0 <= index < len(archives)):
|
||||
print(
|
||||
f'Archive index out of range (0..{len(archives) - 1}).',
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
archive_name = archives[index]
|
||||
except ValueError:
|
||||
archive_name = target
|
||||
except Exception as e:
|
||||
print(
|
||||
f'borg mount failed during archive lookup: {e}', file=sys.stderr
|
||||
)
|
||||
return 1
|
||||
archive_spec = f'{repo}::{archive_name}'
|
||||
|
||||
# Check mountpoint
|
||||
mountpoint = args.mountpoint
|
||||
if not os.path.isdir(mountpoint):
|
||||
print(
|
||||
f"Mountpoint '{mountpoint}' does not exist or is not a directory.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
|
||||
# Run borg mount
|
||||
cmd = ['borg', 'mount', archive_spec, mountpoint]
|
||||
try:
|
||||
subprocess.run(cmd, env=env, check=True)
|
||||
if archive_name:
|
||||
print(f"Archive '{archive_name}' mounted on {mountpoint}")
|
||||
else:
|
||||
print(f'Archives mounted on {mountpoint}')
|
||||
return 0
|
||||
except subprocess.CalledProcessError as e:
|
||||
print(f'borg mount failed: {e}', file=sys.stderr)
|
||||
return 1
|
||||
except Exception as e:
|
||||
print(f'borg mount failed (unexpected): {e}', file=sys.stderr)
|
||||
return 1
|
||||
163
test/test_mount.py
Normal file
163
test/test_mount.py
Normal file
@@ -0,0 +1,163 @@
|
||||
import argparse
|
||||
from unittest import mock
|
||||
import pytest
|
||||
|
||||
from locutus.mount import run_mount
|
||||
|
||||
|
||||
def make_args(config, profile, mountpoint, archive=None):
|
||||
args = argparse.Namespace()
|
||||
args.config = config
|
||||
args.profile = profile
|
||||
args.mountpoint = mountpoint
|
||||
if archive is not None:
|
||||
args.archive = archive
|
||||
return args
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_all_archives_success(mock_run, mock_isdir, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_isdir.return_value = True
|
||||
mock_run.return_value.returncode = 0
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', '/mnt/test')
|
||||
rc = run_mount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert 'Archives mounted on /mnt/test' in out
|
||||
cmd_args = mock_run.call_args[0][0]
|
||||
assert cmd_args == ['borg', 'mount', '/repo', '/mnt/test']
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_by_name_success(mock_run, mock_isdir, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_isdir.return_value = True
|
||||
mock_run.return_value.returncode = 0
|
||||
|
||||
args = make_args(
|
||||
'dummy.toml', 'dummy.rc', '/mnt/test', archive='archive-xyz'
|
||||
)
|
||||
rc = run_mount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "Archive 'archive-xyz' mounted on /mnt/test" in out
|
||||
cmd_args = mock_run.call_args[0][0]
|
||||
assert cmd_args == ['borg', 'mount', '/repo::archive-xyz', '/mnt/test']
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir')
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_by_index_success(mock_run, mock_isdir, mock_config, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_isdir.return_value = True
|
||||
mock_run.return_value.returncode = 0
|
||||
archives = ['archA', 'archB', 'archC'] # Oldest to newest
|
||||
mock_run.side_effect = [
|
||||
# First run: borg list
|
||||
mock.Mock(stdout='\n'.join(archives) + '\n', returncode=0),
|
||||
# Second run: borg mount
|
||||
mock.Mock(returncode=0),
|
||||
]
|
||||
|
||||
args = make_args('dummy.toml', 'dummy.rc', '/mnt/test', archive='1')
|
||||
rc = run_mount(args)
|
||||
assert rc == 0
|
||||
out = capsys.readouterr().out
|
||||
# Index 1 in reversed list is archB (so index 0=newest=archC, 1=archB, 2=archA)
|
||||
assert "Archive 'archB' mounted on /mnt/test" in out
|
||||
cmd_args = mock_run.call_args_list[1][0][0]
|
||||
assert cmd_args == ['borg', 'mount', '/repo::archB', '/mnt/test']
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
def test_mount_no_repo(mock_config, tmp_path, capsys):
|
||||
mock_config.return_value.get_repo.return_value = None
|
||||
mock_config.return_value.env = {}
|
||||
args = make_args('dummy.toml', 'dummy.rc', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'No BORG_REPO configured' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=False)
|
||||
def test_mount_invalid_mountpoint(mock_isdir, mock_config, tmp_path, capsys):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
args = make_args('dummy.toml', 'dummy.rc', '/not/a/dir', archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'does not exist or is not a directory' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run', side_effect=Exception('fail'))
|
||||
def test_mount_unexpected_exception(
|
||||
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
||||
):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
args = make_args('dummy.toml', 'dummy.rc', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'borg mount failed' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_by_index_out_of_range(
|
||||
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
||||
):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
archives = ['a', 'b']
|
||||
mock_run.side_effect = [
|
||||
mock.Mock(stdout='\n'.join(archives) + '\n', returncode=0)
|
||||
]
|
||||
args = make_args(
|
||||
'dummy.toml', 'dummy.rc', str(tmp_path), archive='5'
|
||||
) # Out of range
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'Archive index out of range' in err
|
||||
|
||||
|
||||
@mock.patch('locutus.mount.LocutusConfig')
|
||||
@mock.patch('os.path.isdir', return_value=True)
|
||||
@mock.patch('subprocess.run')
|
||||
def test_mount_by_index_no_archives(
|
||||
mock_run, mock_isdir, mock_config, tmp_path, capsys
|
||||
):
|
||||
mock_config.return_value.get_repo.return_value = '/repo'
|
||||
mock_config.return_value.env = {}
|
||||
mock_run.return_value.stdout = '' # No archives
|
||||
mock_run.return_value.returncode = 0
|
||||
args = make_args('dummy.toml', 'dummy.rc', str(tmp_path), archive='0')
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'No archives found' in err
|
||||
|
||||
|
||||
def test_mount_usage_message(capsys):
|
||||
args = argparse.Namespace()
|
||||
rc = run_mount(args)
|
||||
assert rc == 1
|
||||
err = capsys.readouterr().err
|
||||
assert 'Usage:' in err
|
||||
Reference in New Issue
Block a user