diff --git a/src/locutus/mount.py b/src/locutus/mount.py new file mode 100644 index 0000000..dd80fec --- /dev/null +++ b/src/locutus/mount.py @@ -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] ', 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 diff --git a/test/test_mount.py b/test/test_mount.py new file mode 100644 index 0000000..f32c1e5 --- /dev/null +++ b/test/test_mount.py @@ -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