Backups: remove unittests to backups test directory

And add one bugfix to reset cfg.actions_dir after changing it

Reviewed-by: James Valleroy <jvalleroy@mailbox.org>

actions/backups

@@ -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)

actions/sshfs

@@ -0,0 +1,152 @@
+#!/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()

plinth/modules/backups/repository.py

@@ -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)
 

plinth/modules/backups/tests/backup_data/sample_file.txt

@@ -0,0 +1 @@
+sample file content

plinth/modules/backups/tests/test_backups.py

@@ -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

plinth/modules/backups/views.py

@@ -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)

plinth/tests/test_actions.py

@@ -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. """