Backups, remote repositories: change network_storage to dict

- change network_storage storage to dict format for easier handling
- added unittests

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

plinth/modules/backups/forms.py

@@ -47,7 +47,7 @@ def _get_repository_choices():
     """Return the list of available repositories."""
     choices = [('root', ROOT_REPOSITORY_NAME)]
     storages = network_storage.get_storages()
-    for storage in storages:
+    for storage in storages.values():
         choices += [(storage['uuid'], storage['path'])]
     return choices
 

plinth/modules/backups/network_storage.py

@@ -29,40 +29,40 @@ REQUIRED_FIELDS = ['path', 'storage_type', 'added_by_module']
 
 
 def get_storages(storage_type=None):
-    """Get list of network storage storages"""
-    storages = kvstore.get_default(NETWORK_STORAGE_KEY, [])
+    """Get network storage"""
+    storages = kvstore.get_default(NETWORK_STORAGE_KEY, {})
     if storages:
         storages = json.loads(storages)
     if storage_type:
-        storages = [storage for storage in storages if 'type' in storage
-                    and storage['type'] == storage_type]
+        storages = {uuid: storage for uuid, storage in storages.items() if
+                    storage['storage_type'] == storage_type}
     return storages
 
 
 def get(uuid):
     storages = get_storages()
-    return list(filter(lambda storage: storage['uuid'] == uuid,
-                       storages))[0]
+    return storages[uuid]
 
 
 def update_or_add(storage):
     """Update an existing or create a new network location"""
     for field in REQUIRED_FIELDS:
-        assert field in storage
-    storages = get_storages()
+        if field not in storage:
+            raise ValueError('missing storage parameter: %s' % field)
+    existing_storages = get_storages()
     if 'uuid' in storage:
         # Replace the existing storage
-        storages = [_storage if _storage['uuid'] != storage['uuid'] else
-                    storage for _storage in storages]
+        existing_storages[storage['uuid']] = storage
     else:
-        storage['uuid'] = str(uuid1())
-        storages.append(storage)
-    kvstore.set(NETWORK_STORAGE_KEY, json.dumps(storages))
+        uuid = str(uuid1())
+        storage['uuid'] = uuid
+        existing_storages[uuid] = storage
+    kvstore.set(NETWORK_STORAGE_KEY, json.dumps(existing_storages))
+    return storage['uuid']
 
 
 def delete(uuid):
     """Remove a network storage from kvstore"""
     storages = get_storages()
-    storages = list(filter(lambda storage: storage['uuid'] != uuid,
-                           storages))
+    del storages[uuid]
     kvstore.set(NETWORK_STORAGE_KEY, json.dumps(storages))

plinth/modules/backups/repository.py

@@ -304,7 +304,7 @@ class SshBorgRepository(BorgRepository):
 def get_ssh_repositories():
     """Get all SSH Repositories including the archive content"""
     repositories = {}
-    for storage in network_storage.get_storages():
+    for storage in network_storage.get_storages().values():
         repository = SshBorgRepository(automount=False, **storage)
         repositories[storage['uuid']] = repository.get_view_content()
     return repositories

plinth/tests/test_backups.py

@@ -76,7 +76,7 @@ class TestBackups(unittest.TestCase):
         repository = BorgRepository(repo_path)
         repository.create_repository()
         info = repository.get_info()
-        assert 'encryption' in info
+        self.assertTrue('encryption' in info)
 
     @unittest.skipUnless(euid == 0, 'Needs to be root')
     def test_create_and_delete_archive(self):
@@ -98,11 +98,11 @@ class TestBackups(unittest.TestCase):
                         self.data_directory])
 
         archive = repository.list_archives()[0]
-        assert archive['name'] == archive_name
+        self.assertEquals(archive['name'], archive_name)
 
         repository.delete_archive(archive_name)
         content = repository.list_archives()
-        assert len(content) == 0
+        self.assertEquals(len(content), 0)
 
     @unittest.skipUnless(euid == 0 and config.backups_ssh_path,
                          'Needs to be root and ssh credentials provided')
@@ -117,9 +117,9 @@ class TestBackups(unittest.TestCase):
                                      path=ssh_path,
                                      credentials=credentials)
         ssh_repo.mount()
-        assert ssh_repo.is_mounted
+        self.assertTrue(ssh_repo.is_mounted)
         ssh_repo.umount()
-        assert not ssh_repo.is_mounted
+        self.assertFalse(ssh_repo.is_mounted)
 
     @unittest.skipUnless(euid == 0, 'Needs to be root')
     def test_ssh_create_encrypted_repository(self):
@@ -129,10 +129,11 @@ class TestBackups(unittest.TestCase):
         credentials['encryption_passphrase'] = '12345'
         # using SshBorgRepository to provide credentials because
         # BorgRepository does not allow creating encrypted repositories
+        # TODO: find better way to test encryption
         repository = SshBorgRepository(path=encrypted_repo,
                                        credentials=credentials)
         repository.create_repository('repokey')
-        assert repository.get_info()
+        self.assertTrue(bool(repository.get_info()))
 
     def get_credentials(self):
         """

plinth/tests/test_network_storage.py

@@ -0,0 +1,76 @@
+#
+# 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/>.
+#
+"""
+Test network storage.
+"""
+
+from django.test import TestCase
+
+from plinth.modules.backups import network_storage
+
+
+class TestNetworkStorage(TestCase):
+    """Test handling network storage in kvstore"""
+    storages = [{
+        'path': 'test@nonexistent.org:~/',
+        'storage_type': 'ssh',
+        'added_by_module': 'test'
+    }, {
+        'path': 'test@nonexistent.org:~/tmp/repo/',
+        'storage_type': 'ssh',
+        'added_by_module': 'test'
+    }]
+
+    def test_add(self):
+        """Add a storage item"""
+        storage = self.storages[0]
+        uuid = network_storage.update_or_add(storage)
+        _storage = network_storage.get(uuid)
+        self.assertEqual(_storage['path'], storage['path'])
+
+    def test_add_invalid(self):
+        """Add a storage item"""
+        storage_with_missing_type = {
+            'path': 'test@nonexistent.org:~/tmp/repo/',
+            'added_by_module': 'test'
+        }
+        with self.assertRaises(ValueError):
+            network_storage.update_or_add(storage_with_missing_type)
+
+    def test_remove(self):
+        """Add and remove storage items"""
+        storage = self.storages[0]
+        uuid = None
+        for storage in self.storages:
+            uuid = network_storage.update_or_add(storage)
+        storages = network_storage.get_storages()
+        self.assertEqual(len(storages), 2)
+        network_storage.delete(uuid)
+        storages = network_storage.get_storages()
+        self.assertEqual(len(storages), 1)
+
+    def test_update(self):
+        """Update existing storage items"""
+        uuid = None
+        for storage in self.storages:
+            uuid = network_storage.update_or_add(storage)
+        storage = network_storage.get(uuid)
+        new_path = 'test@nonexistent.org:~/tmp/repo_new/'
+        storage['path'] = new_path
+        network_storage.update_or_add(storage)
+        _storage = network_storage.get(uuid)
+        self.assertEquals(_storage['path'], new_path)