mirror of
https://github.com/freedombox/FreedomBox.git
synced 2026-02-04 08:13:38 +00:00
c33b192f71
Disk labels are unreliable. They may not exist. There are not unique and two disks in the system may have the same label. Instead use the device path of the disk/partition. Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
290 lines
10 KiB
Python
290 lines
10 KiB
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/>.
|
|
#
|
|
"""
|
|
Tests for backups module API.
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
from plinth import cfg, module_loader
|
|
from plinth.errors import PlinthError
|
|
|
|
from .. import api, forms, get_export_locations, get_location_path
|
|
|
|
# pylint: disable=protected-access
|
|
|
|
|
|
def _get_test_manifest(name):
|
|
return api.validate({
|
|
'config': {
|
|
'directories': ['/etc/' + name + '/config.d/'],
|
|
'files': ['/etc/' + name + '/config'],
|
|
},
|
|
'data': {
|
|
'directories': ['/var/lib/' + name + '/data.d/'],
|
|
'files': ['/var/lib/' + name + '/data'],
|
|
},
|
|
'secrets': {
|
|
'directories': ['/etc/' + name + '/secrets.d/'],
|
|
'files': ['/etc/' + name + '/secrets'],
|
|
},
|
|
'services': [name, {
|
|
'type': 'apache',
|
|
'name': name,
|
|
'kind': 'site'
|
|
}]
|
|
})
|
|
|
|
|
|
def _get_backup_app(name):
|
|
"""Return a dummy BackupApp object."""
|
|
return api.BackupApp(name, MagicMock(backup=_get_test_manifest(name)))
|
|
|
|
|
|
class TestBackupApp(unittest.TestCase):
|
|
"""Test the BackupApp class."""
|
|
|
|
def test_run_hook(self):
|
|
"""Test running a hook on an application."""
|
|
packet = api.Packet('backup', 'apps', '/', [])
|
|
hook = 'testhook_pre'
|
|
app = MagicMock()
|
|
backup_app = api.BackupApp('app_name', app)
|
|
backup_app.run_hook(hook, packet)
|
|
app.testhook_pre.assert_has_calls([call(packet)])
|
|
assert not packet.errors
|
|
|
|
app.testhook_pre.reset_mock()
|
|
app.testhook_pre.side_effect = Exception()
|
|
backup_app.run_hook(hook, packet)
|
|
self.assertEqual(packet.errors,
|
|
[api.BackupError('hook', app, hook=hook)])
|
|
|
|
del app.testhook_pre
|
|
backup_app.run_hook(hook, packet)
|
|
|
|
|
|
class TestBackupProcesses(unittest.TestCase):
|
|
"""Test cases for backup processes"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
"""Setup all the test cases."""
|
|
super().setUpClass()
|
|
cfg.read()
|
|
|
|
@staticmethod
|
|
def test_packet_process_manifests():
|
|
"""Test that directories/files are collected from manifests."""
|
|
apps = [_get_backup_app('a'), _get_backup_app('b')]
|
|
packet = api.Packet('backup', 'apps', '/', apps)
|
|
for app in apps:
|
|
for section in ['config', 'data', 'secrets']:
|
|
for directory in app.manifest[section]['directories']:
|
|
assert directory in packet.directories
|
|
for file_path in app.manifest[section]['files']:
|
|
assert file_path in packet.files
|
|
|
|
@staticmethod
|
|
def test_backup_apps():
|
|
"""Test that backup_handler is called."""
|
|
backup_handler = MagicMock()
|
|
api.backup_apps(backup_handler)
|
|
backup_handler.assert_called_once()
|
|
|
|
@staticmethod
|
|
def test_restore_apps():
|
|
"""Test that restore_handler is called."""
|
|
restore_handler = MagicMock()
|
|
api.restore_apps(restore_handler)
|
|
restore_handler.assert_called_once()
|
|
|
|
@patch('plinth.module_loader.loaded_modules.items')
|
|
def test_get_all_apps_for_backup(self, modules):
|
|
"""Test listing apps supporting backup and needing backup."""
|
|
apps = [
|
|
('a', MagicMock(backup=_get_test_manifest('a'))),
|
|
('b', MagicMock(backup=_get_test_manifest('b'))),
|
|
('c', MagicMock(backup=None)),
|
|
('d', MagicMock()),
|
|
]
|
|
del apps[3][1].backup
|
|
modules.return_value = apps
|
|
|
|
module_loader.load_modules()
|
|
returned_apps = api.get_all_apps_for_backup()
|
|
expected_apps = [
|
|
api.BackupApp('a', apps[0][1]),
|
|
api.BackupApp('b', apps[1][1]),
|
|
api.BackupApp('c', apps[2][1])
|
|
]
|
|
self.assertEqual(returned_apps, expected_apps)
|
|
|
|
@staticmethod
|
|
def test_export_locations():
|
|
"""Check get_export_locations returns a list of locations."""
|
|
locations = get_export_locations()
|
|
assert locations
|
|
for location in locations:
|
|
assert isinstance(location, dict)
|
|
assert isinstance(location['path'], str)
|
|
assert isinstance(location['device'], str)
|
|
assert str(location['label'])
|
|
|
|
@staticmethod
|
|
@patch('plinth.module_loader.loaded_modules.items')
|
|
def test__get_apps_in_order(modules):
|
|
"""Test that apps are listed in correct dependency order."""
|
|
apps = [
|
|
('names', MagicMock(backup=_get_test_manifest('names'))),
|
|
('config', MagicMock(backup=_get_test_manifest('config'))),
|
|
]
|
|
modules.return_value = apps
|
|
|
|
module_loader.load_modules()
|
|
app_names = ['config', 'names']
|
|
apps = api._get_apps_in_order(app_names)
|
|
assert apps[0].name == 'names'
|
|
assert apps[1].name == 'config'
|
|
|
|
@staticmethod
|
|
def test__lockdown_apps():
|
|
"""Test that locked flag is set for each app."""
|
|
app_a = MagicMock(locked=False)
|
|
app_b = MagicMock(locked=None)
|
|
apps = [MagicMock(app=app_a), MagicMock(app=app_b)]
|
|
api._lockdown_apps(apps, True)
|
|
assert app_a.locked is True
|
|
assert app_b.locked is True
|
|
|
|
@patch('plinth.action_utils.webserver_is_enabled')
|
|
@patch('plinth.action_utils.service_is_running')
|
|
@patch('plinth.actions.superuser_run')
|
|
def test__shutdown_services(self, run, service_is_running,
|
|
webserver_is_enabled):
|
|
"""Test that services are stopped in correct order."""
|
|
apps = [_get_backup_app('a'), _get_backup_app('b')]
|
|
service_is_running.return_value = True
|
|
webserver_is_enabled.return_value = True
|
|
state = api._shutdown_services(apps)
|
|
|
|
expected_state = [
|
|
api.ServiceHandler.create(apps[0],
|
|
apps[0].manifest['services'][0]),
|
|
api.ServiceHandler.create(apps[0],
|
|
apps[0].manifest['services'][1]),
|
|
api.ServiceHandler.create(apps[1],
|
|
apps[1].manifest['services'][0]),
|
|
api.ServiceHandler.create(apps[1], apps[1].manifest['services'][1])
|
|
]
|
|
self.assertEqual(state, expected_state)
|
|
|
|
service_is_running.assert_has_calls([call('b'), call('a')])
|
|
webserver_is_enabled.assert_has_calls(
|
|
[call('b', kind='site'),
|
|
call('a', kind='site')])
|
|
|
|
calls = [
|
|
call('apache', ['disable', '--name', 'b', '--kind', 'site']),
|
|
call('service', ['stop', 'b']),
|
|
call('apache', ['disable', '--name', 'a', '--kind', 'site']),
|
|
call('service', ['stop', 'a'])
|
|
]
|
|
run.assert_has_calls(calls)
|
|
|
|
@staticmethod
|
|
@patch('plinth.actions.superuser_run')
|
|
def test__restore_services(run):
|
|
"""Test that services are restored in correct order."""
|
|
original_state = [
|
|
api.SystemServiceHandler(None, 'a-service'),
|
|
api.SystemServiceHandler(None, 'b-service'),
|
|
api.ApacheServiceHandler(None, {
|
|
'name': 'c-service',
|
|
'kind': 'site'
|
|
}),
|
|
api.ApacheServiceHandler(None, {
|
|
'name': 'd-service',
|
|
'kind': 'site'
|
|
})
|
|
]
|
|
original_state[0].was_running = True
|
|
original_state[1].was_running = False
|
|
original_state[2].was_enabled = True
|
|
original_state[3].was_enabled = False
|
|
api._restore_services(original_state)
|
|
calls = [
|
|
call('service', ['start', 'a-service']),
|
|
call('apache', ['enable', '--name', 'c-service', '--kind', 'site'])
|
|
]
|
|
run.assert_has_calls(calls)
|
|
|
|
@staticmethod
|
|
def test__run_operation():
|
|
"""Test that operation runs handler and app hooks."""
|
|
apps = [_get_backup_app('a'), _get_backup_app('b')]
|
|
packet = api.Packet('backup', 'apps', '/', apps)
|
|
packet.apps[0].run_hook = MagicMock()
|
|
packet.apps[1].run_hook = MagicMock()
|
|
handler = MagicMock()
|
|
api._run_operation(handler, packet)
|
|
handler.assert_has_calls([call(packet)])
|
|
|
|
calls = [call('backup_pre', packet), call('backup_post', packet)]
|
|
packet.apps[0].run_hook.assert_has_calls(calls)
|
|
packet.apps[1].run_hook.assert_has_calls(calls)
|
|
|
|
|
|
class TestBackupModule(unittest.TestCase):
|
|
"""Tests of the backups django module, like views or forms."""
|
|
|
|
def test_get_location_path(self):
|
|
"""Test the 'get_location_path' method"""
|
|
locations = [{
|
|
'path': '/var/www',
|
|
'device': '/dummy/device'
|
|
}, {
|
|
'path': '/etc',
|
|
'device': '/dangerous'
|
|
}]
|
|
location_path = get_location_path('/dummy/device', locations)
|
|
self.assertEqual(location_path, locations[0]['path'])
|
|
# verify that an unknown location raises an error
|
|
with self.assertRaises(PlinthError):
|
|
get_location_path('/unknown/device', locations)
|
|
|
|
def test_file_upload(self):
|
|
locations = get_export_locations()
|
|
location_name = locations[0]['device']
|
|
post_data = {'location': location_name}
|
|
|
|
# posting a video should fail
|
|
video_file = SimpleUploadedFile("video.mp4", b"file_content",
|
|
content_type="video/mp4")
|
|
form = forms.UploadForm(post_data, {'file': video_file})
|
|
self.assertFalse(form.is_valid())
|
|
|
|
# posting an archive file should work
|
|
archive_file = SimpleUploadedFile("backup.tar.gz", b"file_content",
|
|
content_type="application/gzip")
|
|
form = forms.UploadForm(post_data, {'file': archive_file})
|
|
form.is_valid()
|
|
self.assertTrue(form.is_valid())
|