Verified Commit e53aae65 authored by Michael P's avatar Michael P

Backups: remove unittests to backups test directory

And add one bugfix to reset cfg.actions_dir after changing it
parent f662fe5a
Pipeline #27606 failed with stages
in 4 minutes and 32 seconds
......@@ -68,22 +68,10 @@ def parse_arguments():
'get-archive-apps',
help='Get list of apps included in archive')
# SSH mount actions
mount = subparsers.add_parser('mount', help='mount an ssh filesystem')
mount.add_argument('--mountpoint', help='Local mountpoint', required=True)
umount = subparsers.add_parser('umount',
help='unmount an ssh filesystem')
umount.add_argument('--mountpoint', help='Mountpoint to unmount',
required=True)
is_mounted = subparsers.add_parser('is-mounted',
help='Check whether an sshfs is mouned')
is_mounted.add_argument('--mountpoint', help='Mountpoint to check',
required=True)
for cmd in [info, init, list_repo, create_archive, delete_archive,
export_tar, get_archive_apps, mount, setup]:
export_tar, get_archive_apps, setup]:
cmd.add_argument('--path', help='Repository or Archive path',
required=False) # TODO: set required to True!
required=False)
cmd.add_argument('--ssh-keyfile', help='Path of private ssh key',
default=None)
cmd.add_argument('--encryption-passphrase',
......@@ -159,66 +147,6 @@ def subcommand_info(arguments):
run(['borg', 'info', '--json', arguments.path], env=env)
def subcommand_mount(arguments):
"""Show repository information."""
try:
validate_mountpoint(arguments.mountpoint)
except AlreadyMountedError:
return
env = get_env(arguments)
remote_path = arguments.path
kwargs = {}
# the shell would expand ~/ to the local home directory
remote_path = remote_path.replace('~/', '').replace('~', '')
cmd = ['sshfs', remote_path, arguments.mountpoint, '-o',
'UserKnownHostsFile=/dev/null', '-o',
'StrictHostKeyChecking=no']
timeout = None
if 'SSHPASS' in env:
cmd += ['-o', 'password_stdin']
kwargs['input'] = env['SSHPASS'].encode()
elif 'SSHKEY' in env:
cmd += ['-o', 'IdentityFile=$SSHKEY']
timeout = TIMEOUT
else:
raise ValueError('mount requires either SSHPASS or SSHKEY in env')
subprocess.run(cmd, check=True, env=env, timeout=timeout, **kwargs)
def subcommand_umount(arguments):
"""Show repository information."""
run(['umount', arguments.mountpoint])
def validate_mountpoint(mountpoint):
"""Check that the folder is empty, and create it if it doesn't exist"""
if os.path.exists(mountpoint):
if _is_mounted(mountpoint):
raise AlreadyMountedError('Mountpoint %s already mounted' %
mountpoint)
if os.listdir(mountpoint) or not os.path.isdir(mountpoint):
raise ValueError('Mountpoint %s is not an empty directory' %
mountpoint)
else:
os.makedirs(mountpoint)
def _is_mounted(mountpoint):
"""Return boolean whether a local directory is a mountpoint."""
cmd = ['mountpoint', '-q', mountpoint]
# mountpoint exits with status non-zero if it didn't find a mountpoint
try:
subprocess.run(cmd, check=True)
return True
except subprocess.CalledProcessError:
return False
def subcommand_is_mounted(arguments):
print(json.dumps(_is_mounted(arguments.mountpoint)))
def subcommand_list_repo(arguments):
"""List repository contents."""
env = get_env(arguments)
......
#!/usr/bin/python3
# -*- mode: python -*-
#
# This file is part of FreedomBox.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
"""
Actions for sshfs.
"""
import argparse
import json
import os
import subprocess
import sys
TIMEOUT = 5
class AlreadyMountedError(Exception):
pass
def parse_arguments():
"""Return parsed command line arguments as dictionary."""
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')
mount = subparsers.add_parser('mount', help='mount an ssh filesystem')
mount.add_argument('--mountpoint', help='Local mountpoint', required=True)
mount.add_argument('--path', help='Remote ssh path to mount',
required=True)
umount = subparsers.add_parser('umount',
help='unmount an ssh filesystem')
umount.add_argument('--mountpoint', help='Mountpoint to unmount',
required=True)
is_mounted = subparsers.add_parser('is-mounted',
help='Check whether an sshfs is mouned')
is_mounted.add_argument('--mountpoint', help='Mountpoint to check',
required=True)
subparsers.required = True
return parser.parse_args()
def get_env(arguments, read_input=True):
"""Create encryption and ssh kwargs out of given arguments"""
env = dict(os.environ)
password = read_password() if read_input else None
if password:
env['SSHPASS'] = password
env['BORG_RSH'] = 'sshpass -e ssh -o StrictHostKeyChecking=no'
return env
def subcommand_mount(arguments):
"""Show repository information."""
try:
validate_mountpoint(arguments.mountpoint)
except AlreadyMountedError:
return
env = get_env(arguments)
remote_path = arguments.path
kwargs = {}
# the shell would expand ~/ to the local home directory
remote_path = remote_path.replace('~/', '').replace('~', '')
cmd = ['sshfs', remote_path, arguments.mountpoint, '-o',
'UserKnownHostsFile=/dev/null', '-o',
'StrictHostKeyChecking=no']
timeout = None
if 'SSHPASS' in env:
cmd += ['-o', 'password_stdin']
kwargs['input'] = env['SSHPASS'].encode()
elif 'SSHKEY' in env:
cmd += ['-o', 'IdentityFile=$SSHKEY']
timeout = TIMEOUT
else:
raise ValueError('mount requires either SSHPASS or SSHKEY in env')
subprocess.run(cmd, check=True, env=env, timeout=timeout, **kwargs)
def subcommand_umount(arguments):
"""Show repository information."""
run(['umount', arguments.mountpoint])
def validate_mountpoint(mountpoint):
"""Check that the folder is empty, and create it if it doesn't exist"""
if os.path.exists(mountpoint):
if _is_mounted(mountpoint):
raise AlreadyMountedError('Mountpoint %s already mounted' %
mountpoint)
if os.listdir(mountpoint) or not os.path.isdir(mountpoint):
raise ValueError('Mountpoint %s is not an empty directory' %
mountpoint)
else:
os.makedirs(mountpoint)
def _is_mounted(mountpoint):
"""Return boolean whether a local directory is a mountpoint."""
cmd = ['mountpoint', '-q', mountpoint]
# mountpoint exits with status non-zero if it didn't find a mountpoint
try:
subprocess.run(cmd, check=True)
return True
except subprocess.CalledProcessError:
return False
def subcommand_is_mounted(arguments):
print(json.dumps(_is_mounted(arguments.mountpoint)))
def read_password():
"""Read the password from stdin."""
if sys.stdin.isatty():
return ''
else:
return ''.join(sys.stdin)
def run(cmd, env=None, check=True):
"""Wrap the command with ssh password or keyfile authentication"""
# Set a timeout to not get stuck if the remote server asks for a password.
subprocess.run(cmd, check=check, env=env, timeout=TIMEOUT)
def main():
"""Parse arguments and perform all duties."""
arguments = parse_arguments()
subcommand = arguments.subcommand.replace('-', '_')
subcommand_method = globals()['subcommand_' + subcommand]
subcommand_method(arguments)
if __name__ == '__main__':
main()
......@@ -58,7 +58,6 @@ KNOWN_ERRORS = [{
class BorgRepository(object):
"""Borg repository on the root filesystem"""
command = 'backups'
storage_type = 'root'
name = ROOT_REPOSITORY_NAME
is_mounted = True
......@@ -68,11 +67,11 @@ class BorgRepository(object):
self.credentials = credentials
def get_info(self):
output = self._run(['info', '--path', self.path])
output = self._run('backups', ['info', '--path', self.path])
return json.loads(output)
def list_archives(self):
output = self._run(['list-repo', '--path', self.path])
output = self._run('backups', ['list-repo', '--path', self.path])
return json.loads(output)['archives']
def get_view_content(self):
......@@ -93,7 +92,7 @@ class BorgRepository(object):
def delete_archive(self, archive_name):
archive_path = self.get_archive_path(archive_name)
self._run(['delete-archive', '--path', archive_path])
self._run('backups', ['delete-archive', '--path', archive_path])
def remove_repository(self):
"""Remove a borg repository"""
......@@ -105,18 +104,18 @@ class BorgRepository(object):
app_names=app_names)
def create_repository(self):
self._run(['init', '--path', self.path, '--encryption', 'none'])
self._run('backups', ['init', '--path', self.path, '--encryption',
'none'])
def get_zipstream(self, archive_name):
archive_path = self.get_archive_path(archive_name)
args = ['export-tar', '--path', archive_path]
kwargs = {'run_in_background': True,
'bufsize': 1}
proc = self._run(args, kwargs=kwargs, use_credentials=False)
proc = self._run('backups', args, kwargs=kwargs, use_credentials=False)
return zipstream.ZipStream(proc.stdout, 'readline')
def get_archive(self, name):
# TODO: can't we get this archive directly?
for archive in self.list_archives():
if archive['name'] == name:
return archive
......@@ -126,7 +125,8 @@ class BorgRepository(object):
def get_archive_apps(self, archive_name):
"""Get list of apps included in an archive."""
archive_path = self.get_archive_path(archive_name)
output = self._run(['get-archive-apps', '--path', archive_path])
output = self._run('backups', ['get-archive-apps', '--path',
archive_path])
return output.splitlines()
def restore_archive(self, archive_name, apps=None):
......@@ -140,16 +140,16 @@ class BorgRepository(object):
def _get_env(self):
return dict(os.environ, BORG_RELOCATED_REPO_ACCESS_IS_OK='yes')
def _run(self, arguments, superuser=True, kwargs=None,
def _run(self, cmd, arguments, superuser=True, kwargs=None,
use_credentials=False):
"""Run a backups action script command."""
if kwargs is None:
kwargs = {}
try:
if superuser:
return actions.superuser_run(self.command, arguments, **kwargs)
return actions.superuser_run(cmd, arguments, **kwargs)
else:
return actions.run(self.command, arguments, **kwargs)
return actions.run(cmd, arguments, **kwargs)
except ActionError as err:
self.reraise_known_error(err)
......@@ -205,7 +205,8 @@ class SshBorgRepository(BorgRepository):
@property
def is_mounted(self):
output = self._run(['is-mounted', '--mountpoint', self.mountpoint])
output = self._run('sshfs', ['is-mounted', '--mountpoint',
self.mountpoint])
return json.loads(output)
def get_archive_path(self, archive_name):
......@@ -234,18 +235,20 @@ class SshBorgRepository(BorgRepository):
def create_repository(self, encryption):
if encryption not in SUPPORTED_BORG_ENCRYPTION:
raise ValueError('Unsupported encryption: %s' % encryption)
self._run(['init', '--path', self.path, '--encryption', encryption])
self._run('backups', ['init', '--path', self.path, '--encryption',
encryption])
def save(self, store_credentials=True):
storage = self._get_network_storage_format(store_credentials)
self.uuid = network_storage.update_or_add(storage)
def mount(self):
cmd = ['mount', '--mountpoint', self.mountpoint, '--path', self.path]
self._run(cmd)
arguments = ['mount', '--mountpoint', self.mountpoint, '--path',
self.path]
self._run('sshfs', arguments)
def umount(self):
self._run(['umount', '--mountpoint', self.mountpoint])
self._run('sshfs', ['umount', '--mountpoint', self.mountpoint])
def remove_repository(self):
"""Remove a repository from the kvstore and delete its mountpoint"""
......@@ -274,7 +277,7 @@ class SshBorgRepository(BorgRepository):
arguments += ['--ssh-keyfile', credentials['ssh_keyfile']]
return (arguments, kwargs)
def _run(self, arguments, superuser=True, use_credentials=True,
def _run(self, cmd, arguments, superuser=True, use_credentials=True,
kwargs=None):
"""Run a backups action script command.
......@@ -294,9 +297,9 @@ class SshBorgRepository(BorgRepository):
kwargs=kwargs)
try:
if superuser:
return actions.superuser_run(self.command, arguments, **kwargs)
return actions.superuser_run(cmd, arguments, **kwargs)
else:
return actions.run(self.command, arguments, **kwargs)
return actions.run(cmd, arguments, **kwargs)
except ActionError as err:
self.reraise_known_error(err)
......
......@@ -28,7 +28,7 @@ from plinth.modules import backups
from plinth.modules.backups.repository import BorgRepository, SshBorgRepository
from plinth import actions
from . import config
from plinth.tests import config as test_config
euid = os.geteuid()
......@@ -42,17 +42,20 @@ class TestBackups(unittest.TestCase):
cls.action_directory = tempfile.TemporaryDirectory()
cls.backup_directory = tempfile.TemporaryDirectory()
cls.data_directory = os.path.join(os.path.dirname(
os.path.realpath(__file__)), 'data')
os.path.realpath(__file__)), 'backup_data')
cls.actions_dir_factory = cfg.actions_dir
cfg.actions_dir = cls.action_directory.name
actions_dir = os.path.join(os.path.dirname(__file__), '..', '..',
'actions')
actions_dir = os.path.join(os.path.dirname(__file__), '..', '..', '..',
'..', 'actions')
shutil.copy(os.path.join(actions_dir, 'backups'), cfg.actions_dir)
shutil.copy(os.path.join(actions_dir, 'sshfs'), cfg.actions_dir)
@classmethod
def tearDownClass(cls):
"""Cleanup after all the tests are completed."""
cls.action_directory.cleanup()
cls.backup_directory.cleanup()
cfg.actions_dir = cls.actions_dir_factory
@unittest.skipUnless(euid == 0, 'Needs to be root')
def test_nonexisting_repository(self):
......@@ -79,7 +82,7 @@ class TestBackups(unittest.TestCase):
self.assertTrue('encryption' in info)
@unittest.skipUnless(euid == 0, 'Needs to be root')
def test_create_and_delete_archive(self):
def test_create_export_delete_archive(self):
"""
- Create a repo
- Create an archive
......@@ -104,18 +107,19 @@ class TestBackups(unittest.TestCase):
content = repository.list_archives()
self.assertEquals(len(content), 0)
@unittest.skipUnless(euid == 0 and config.backups_ssh_path,
@unittest.skipUnless(euid == 0 and test_config.backups_ssh_path,
'Needs to be root and ssh credentials provided')
def test_ssh_mount(self):
"""Test (un)mounting if credentials for a remote location are given"""
credentials = self.get_credentials()
if not credentials:
return
ssh_path = config.backups_ssh_path
ssh_path = test_config.backups_ssh_path
ssh_repo = SshBorgRepository(uuid='plinth_test_sshfs',
path=ssh_path,
credentials=credentials)
credentials=credentials,
automount=False)
ssh_repo.mount()
self.assertTrue(ssh_repo.is_mounted)
ssh_repo.umount()
......@@ -141,8 +145,8 @@ class TestBackups(unittest.TestCase):
Return an empty dict if no valid access params are found.
"""
credentials = {}
if config.backups_ssh_password:
credentials['ssh_password'] = config.backups_ssh_password
elif config.backups_ssh_keyfile:
credentials['ssh_keyfile'] = config.backups_ssh_keyfile
if test_config.backups_ssh_password:
credentials['ssh_password'] = test_config.backups_ssh_password
elif test_config.backups_ssh_keyfile:
credentials['ssh_keyfile'] = test_config.backups_ssh_keyfile
return credentials
......@@ -63,9 +63,6 @@ class IndexView(TemplateView):
"""View to show list of archives."""
template_name = 'backups.html'
def get_remote_archives(self):
return {} # uuid --> archive list
def get_context_data(self, **kwargs):
"""Return additional context for rendering the template."""
context = super().get_context_data(**kwargs)
......
......@@ -43,6 +43,7 @@ class TestActions(unittest.TestCase):
def setUpClass(cls):
"""Initial setup for all the classes."""
cls.action_directory = tempfile.TemporaryDirectory()
cls.actions_dir_factory = cfg.actions_dir
cfg.actions_dir = cls.action_directory.name
actions_dir = os.path.join(os.path.dirname(__file__), '..', '..',
'actions')
......@@ -56,6 +57,7 @@ class TestActions(unittest.TestCase):
def tearDownClass(cls):
"""Cleanup after all the tests are completed."""
cls.action_directory.cleanup()
cfg.actions_dir = cls.actions_dir_factory
def notest_run_as_root(self):
"""1. Privileged actions run as root. """
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment