calibre: Use privileged decorator for actions

Tests:

- Unit and functional tests work.
- Creating a library works.
  - An error while creating library shows as proper message.
- Deleting a library works.
  - An error while deleting library shows as proper messages.
- Creating/deleting library reflects properly in the list of libraries.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
This commit is contained in:
Sunil Mohan Adapa 2022-08-26 16:08:29 -07:00 committed by James Valleroy
parent cdb04bb46a
commit d7a60b1aca
No known key found for this signature in database
GPG Key ID: 77C0C75E7B650808
8 changed files with 136 additions and 197 deletions

View File

@ -1,76 +0,0 @@
#!/usr/bin/python3
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Configuration helper for calibre.
"""
import argparse
import json
import pathlib
import shutil
import subprocess
from plinth.modules import calibre
LIBRARIES_PATH = pathlib.Path('/var/lib/calibre-server-freedombox/libraries')
def parse_arguments():
"""Return parsed command line arguments as dictionary."""
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest='subcommand', help='Sub command')
subparsers.add_parser('list-libraries',
help='Return the list of libraries setup')
subparser = subparsers.add_parser('create-library',
help='Create an empty library')
subparser.add_argument('name', help='Name of the new library')
subparser = subparsers.add_parser('delete-library',
help='Delete a library and its contents')
subparser.add_argument('name', help='Name of the library to delete')
subparsers.required = True
return parser.parse_args()
def subcommand_list_libraries(_):
"""Return the list of libraries setup."""
libraries = []
for library in LIBRARIES_PATH.glob('*/metadata.db'):
libraries.append(str(library.parent.name))
print(json.dumps({'libraries': libraries}))
def subcommand_create_library(arguments):
"""Create an empty library."""
calibre.validate_library_name(arguments.name)
library = LIBRARIES_PATH / arguments.name
library.mkdir(mode=0o755) # Raise exception if already exists
subprocess.call(
['calibredb', '--with-library', library, 'list_categories'],
stdout=subprocess.DEVNULL)
# Force systemd StateDirectory= logic to assign proper ownership to the
# DynamicUser=
shutil.chown(LIBRARIES_PATH.parent, 'root', 'root')
def subcommand_delete_library(arguments):
"""Delete a library and its contents."""
calibre.validate_library_name(arguments.name)
library = LIBRARIES_PATH / arguments.name
shutil.rmtree(library)
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()

View File

@ -1,14 +1,10 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
FreedomBox app for calibre e-book library.
"""
"""FreedomBox app for calibre e-book library."""
import json
import re
from django.utils.translation import gettext_lazy as _
from plinth import actions
from plinth import app as app_module
from plinth import cfg, frontpage, menu
from plinth.daemon import Daemon
@ -109,21 +105,3 @@ def validate_library_name(library_name):
"""Raise exception if library name does not fit the accepted pattern."""
if not re.fullmatch(r'[A-Za-z0-9_.-]+', library_name):
raise Exception('Invalid library name')
def list_libraries():
"""Return a list of libraries."""
output = actions.superuser_run('calibre', ['list-libraries'])
return json.loads(output)['libraries']
def create_library(name):
"""Create an empty library."""
actions.superuser_run('calibre', ['create-library', name])
actions.superuser_run('service', ['try-restart', CalibreApp.DAEMON])
def delete_library(name):
"""Delete a library and its contents."""
actions.superuser_run('calibre', ['delete-library', name])
actions.superuser_run('service', ['try-restart', CalibreApp.DAEMON])

View File

@ -1,14 +1,12 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Django form for configuring calibre.
"""
"""Django form for configuring calibre."""
from django import forms
from django.core import validators
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
from plinth.modules import calibre
from . import privileged
class CreateLibraryForm(forms.Form):
@ -25,7 +23,7 @@ class CreateLibraryForm(forms.Form):
"""Check if the library name is valid."""
name = self.cleaned_data['name']
if name in calibre.list_libraries():
if name in privileged.list_libraries():
raise ValidationError(
_('A library with this name already exists.'))

View File

@ -0,0 +1,47 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Configuration helper for calibre."""
import pathlib
import shutil
import subprocess
from plinth import action_utils
from plinth.actions import privileged
from plinth.modules import calibre
LIBRARIES_PATH = pathlib.Path('/var/lib/calibre-server-freedombox/libraries')
@privileged
def list_libraries() -> list[str]:
"""Return the list of libraries setup."""
libraries = []
for library in LIBRARIES_PATH.glob('*/metadata.db'):
libraries.append(str(library.parent.name))
return libraries
@privileged
def create_library(name: str):
"""Create an empty library."""
calibre.validate_library_name(name)
library = LIBRARIES_PATH / name
library.mkdir(mode=0o755) # Raise exception if already exists
subprocess.call(
['calibredb', '--with-library', library, 'list_categories'],
stdout=subprocess.DEVNULL)
# Force systemd StateDirectory= logic to assign proper ownership to the
# DynamicUser=
shutil.chown(LIBRARIES_PATH.parent, 'root', 'root')
action_utils.service_try_restart(calibre.CalibreApp.DAEMON)
@privileged
def delete_library(name: str):
"""Delete a library and its contents."""
calibre.validate_library_name(name)
library = LIBRARIES_PATH / name
shutil.rmtree(library)
action_utils.service_try_restart(calibre.CalibreApp.DAEMON)

View File

@ -1,67 +0,0 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Test module for calibre actions.
"""
import json
import pathlib
from unittest.mock import call, patch
import pytest
actions_name = 'calibre'
@pytest.fixture(autouse=True)
def fixture_libraries_path(actions_module, tmpdir):
"""Set the libraries path in the actions module."""
actions_module.LIBRARIES_PATH = pathlib.Path(str(tmpdir))
@pytest.fixture(autouse=True)
def fixture_patch():
"""Patch some underlying methods."""
def side_effect(*args, **_kwargs):
if args[0][0] != 'calibredb':
return
path = pathlib.Path(args[0][2]) / 'metadata.db'
path.touch()
with patch('subprocess.call') as subprocess_call, \
patch('shutil.chown'):
subprocess_call.side_effect = side_effect
yield
def test_list_libraries(call_action):
"""Test listing libraries."""
assert json.loads(call_action(['list-libraries'])) == {'libraries': []}
call_action(['create-library', 'TestLibrary'])
expected_output = {'libraries': ['TestLibrary']}
assert json.loads(call_action(['list-libraries'])) == expected_output
@patch('shutil.chown')
def test_create_library(chown, call_action, actions_module):
"""Test creating a library."""
call_action(['create-library', 'TestLibrary'])
library = actions_module.LIBRARIES_PATH / 'TestLibrary'
assert (library / 'metadata.db').exists()
assert library.stat().st_mode == 0o40755
expected_output = {'libraries': ['TestLibrary']}
assert json.loads(call_action(['list-libraries'])) == expected_output
chown.assert_has_calls([call(library.parent.parent, 'root', 'root')])
def test_delete_library(call_action):
"""Test deleting a library."""
call_action(['create-library', 'TestLibrary'])
call_action(['delete-library', 'TestLibrary'])
assert json.loads(call_action(['list-libraries'])) == {'libraries': []}
with pytest.raises(FileNotFoundError):
call_action(['delete-library', 'TestLibrary'])

View File

@ -0,0 +1,64 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Test module for calibre actions."""
import pathlib
from unittest.mock import call, patch
import pytest
from plinth.modules.calibre import privileged
pytestmark = pytest.mark.usefixtures('mock_privileged')
privileged_modules_to_mock = ['plinth.modules.calibre.privileged']
@pytest.fixture(autouse=True)
def fixture_libraries_path(tmpdir):
"""Set the libraries path in the actions module."""
privileged.LIBRARIES_PATH = pathlib.Path(str(tmpdir))
@pytest.fixture(autouse=True)
def fixture_patch():
"""Patch some underlying methods."""
def side_effect(*args, **_kwargs):
if args[0][0] != 'calibredb':
return
path = pathlib.Path(args[0][2]) / 'metadata.db'
path.touch()
with patch('subprocess.call') as subprocess_call, \
patch('shutil.chown'):
subprocess_call.side_effect = side_effect
yield
def test_list_libraries():
"""Test listing libraries."""
assert privileged.list_libraries() == []
privileged.create_library('TestLibrary')
assert privileged.list_libraries() == ['TestLibrary']
@patch('shutil.chown')
def test_create_library(chown):
"""Test creating a library."""
privileged.create_library('TestLibrary')
library = privileged.LIBRARIES_PATH / 'TestLibrary'
assert (library / 'metadata.db').exists()
assert library.stat().st_mode == 0o40755
assert privileged.list_libraries() == ['TestLibrary']
chown.assert_has_calls([call(library.parent.parent, 'root', 'root')])
def test_delete_library():
"""Test deleting a library."""
privileged.create_library('TestLibrary')
privileged.delete_library('TestLibrary')
assert privileged.list_libraries() == []
with pytest.raises(FileNotFoundError):
privileged.delete_library('TestLibrary')

View File

@ -10,7 +10,7 @@ from django import urls
from django.contrib.messages.storage.fallback import FallbackStorage
from django.http.response import Http404
from plinth import actions, module_loader
from plinth import module_loader
from plinth.modules.calibre import views
# For all tests, use plinth.urls instead of urls configured for testing
@ -30,7 +30,8 @@ def fixture_calibre_urls():
@pytest.fixture(autouse=True)
def calibre_patch():
"""Patch calibre methods."""
with patch('plinth.modules.calibre.list_libraries') as list_libraries:
with patch('plinth.modules.calibre.privileged.list_libraries'
) as list_libraries:
list_libraries.return_value = ['TestExistingLibrary']
yield
@ -46,7 +47,7 @@ def make_request(request, view, **kwargs):
return response, messages
@patch('plinth.modules.calibre.create_library')
@patch('plinth.modules.calibre.privileged.create_library')
def test_create_library(create_library, rf):
"""Test that create library view works."""
form_data = {'calibre-name': 'TestLibrary'}
@ -60,11 +61,10 @@ def test_create_library(create_library, rf):
create_library.assert_has_calls([call('TestLibrary')])
@patch('plinth.modules.calibre.create_library')
@patch('plinth.modules.calibre.privileged.create_library')
def test_create_library_failed(create_library, rf):
"""Test that create library fails as expected."""
create_library.side_effect = actions.ActionError('calibre', 'TestOutput',
'TestError')
create_library.side_effect = RuntimeError('TestError')
form_data = {'calibre-name': 'TestLibrary'}
request = rf.post(urls.reverse('calibre:create-library'), data=form_data)
view = views.CreateLibraryView.as_view()
@ -109,7 +109,7 @@ def test_delete_library_confirmation_view(_app, rf):
assert response.context_data['name'] == 'TestExistingLibrary'
@patch('plinth.modules.calibre.delete_library')
@patch('plinth.modules.calibre.privileged.delete_library')
@patch('plinth.app.App.get')
def test_delete_library(_app, delete_library, rf):
"""Test that deleting a library works."""
@ -121,18 +121,16 @@ def test_delete_library(_app, delete_library, rf):
delete_library.assert_has_calls([call('TestExistingLibrary')])
@patch('plinth.modules.calibre.delete_library')
@patch('plinth.modules.calibre.privileged.delete_library')
def test_delete_library_error(delete_library, rf):
"""Test that deleting a library shows error when operation fails."""
delete_library.side_effect = actions.ActionError('calibre', 'TestInput',
'TestError')
delete_library.side_effect = ValueError('TestError')
response, messages = make_request(rf.post(''), views.delete_library,
name='TestExistingLibrary')
assert response.status_code == 302
assert response.url == urls.reverse('calibre:index')
assert list(messages)[0].message == \
'Could not delete TestExistingLibrary: '\
"('calibre', 'TestInput', 'TestError')"
'Could not delete TestExistingLibrary: TestError'
def test_delete_library_non_existing(rf):

View File

@ -1,7 +1,5 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Views for the calibre module.
"""
"""Views for the calibre module."""
from django.contrib import messages
from django.contrib.messages.views import SuccessMessageMixin
@ -12,28 +10,28 @@ from django.urls import reverse_lazy
from django.utils.translation import gettext as _
from django.views.generic.edit import FormView
from plinth import actions
from plinth import app as app_module
from plinth import views
from plinth.modules import calibre
from . import forms
from . import forms, privileged
class CalibreAppView(views.AppView):
"""Serve configuration form."""
app_id = 'calibre'
template_name = 'calibre.html'
def get_context_data(self, **kwargs):
"""Return additional context for rendering the template."""
context = super().get_context_data(**kwargs)
context['libraries'] = calibre.list_libraries()
context['libraries'] = privileged.list_libraries()
return context
class CreateLibraryView(SuccessMessageMixin, FormView):
"""View to create an empty library."""
form_class = forms.CreateLibraryForm
prefix = 'calibre'
template_name = 'form.html'
@ -43,28 +41,27 @@ class CreateLibraryView(SuccessMessageMixin, FormView):
def form_valid(self, form):
"""Create the library on valid form submission."""
try:
calibre.create_library(form.cleaned_data['name'])
except actions.ActionError as error:
privileged.create_library(form.cleaned_data['name'])
except Exception as error:
self.success_message = ''
error_text = error.args[2].split('\n')[0]
messages.error(
self.request, "{0} {1}".format(
_('An error occurred while creating the library.'),
error_text))
str(error)))
return super().form_valid(form)
def delete_library(request, name):
"""View to delete a library."""
if name not in calibre.list_libraries():
if name not in privileged.list_libraries():
raise Http404
if request.method == 'POST':
try:
calibre.delete_library(name)
privileged.delete_library(name)
messages.success(request, _('{name} deleted.').format(name=name))
except actions.ActionError as error:
except Exception as error:
messages.error(
request,
_('Could not delete {name}: {error}').format(