From 3c24c9d63b0d985a607361f51d577a7ad9786257 Mon Sep 17 00:00:00 2001 From: Michael Pimmer Date: Fri, 30 Nov 2018 15:51:13 +0000 Subject: [PATCH] Backups, remote repositories: change network_storage to dict - change network_storage storage to dict format for easier handling - added unittests Reviewed-by: James Valleroy --- plinth/modules/backups/forms.py | 2 +- plinth/modules/backups/network_storage.py | 30 ++++----- plinth/modules/backups/repository.py | 2 +- plinth/tests/test_backups.py | 13 ++-- plinth/tests/test_network_storage.py | 76 +++++++++++++++++++++++ 5 files changed, 100 insertions(+), 23 deletions(-) create mode 100644 plinth/tests/test_network_storage.py diff --git a/plinth/modules/backups/forms.py b/plinth/modules/backups/forms.py index 6e7b5fa1c..f58a224f4 100644 --- a/plinth/modules/backups/forms.py +++ b/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 diff --git a/plinth/modules/backups/network_storage.py b/plinth/modules/backups/network_storage.py index 174433775..455857481 100644 --- a/plinth/modules/backups/network_storage.py +++ b/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)) diff --git a/plinth/modules/backups/repository.py b/plinth/modules/backups/repository.py index 11e392bd3..c5c20fa9d 100644 --- a/plinth/modules/backups/repository.py +++ b/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 diff --git a/plinth/tests/test_backups.py b/plinth/tests/test_backups.py index 9c48427ae..becc5aa55 100644 --- a/plinth/tests/test_backups.py +++ b/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): """ diff --git a/plinth/tests/test_network_storage.py b/plinth/tests/test_network_storage.py new file mode 100644 index 000000000..be604414b --- /dev/null +++ b/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 . +# +""" +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)