From b25204b73bc6a76ae60d52b1a90d1b5919612252 Mon Sep 17 00:00:00 2001 From: Fox Danger Piacenti Date: Sat, 15 Aug 2026 16:52:17 -0500 Subject: [PATCH] feat: gotenberg support for PDF block --- xblock_pdf/pdf.py | 69 +++++++++--- xblock_pdf/tests/test_pdf.py | 203 +++++++++++++++++++++++++---------- xblock_pdf/utils.py | 121 ++++++++++++++++++++- 3 files changed, 317 insertions(+), 76 deletions(-) diff --git a/xblock_pdf/pdf.py b/xblock_pdf/pdf.py index ee20032f..034d8974 100644 --- a/xblock_pdf/pdf.py +++ b/xblock_pdf/pdf.py @@ -1,20 +1,34 @@ """pdfXBlock main Python class.""" import json +from logging import getLogger +from urllib.parse import urlparse +from django.contrib.auth import get_user_model from django.utils.translation import gettext_noop as _ +from requests import HTTPError, Timeout from web_fragments.fragment import Fragment from webob import Response from xblock.core import XBlock from xblock.fields import Boolean, Scope, String from xblock.utils.resources import ResourceLoader -from .utils import bool_from_str, is_all_download_disabled +from .utils import ( + add_asset, + convert_to_pdf, + error_response, + fetch_source_asset, + is_all_download_disabled, + is_gotenberg_enabled, +) resource_loader = ResourceLoader(__name__) +logger = getLogger(__name__) -@XBlock.needs("i18n") + +@XBlock.needs("i18n", "user") +@XBlock.wants("studio_user_permissions") class PDFBlock(XBlock): """PDF XBlock. Allows authors to embed PDFs in their courses.""" @@ -69,6 +83,7 @@ def raw_settings(self): "url": self.url, "allow_download": self.allow_download, "disable_all_download": is_all_download_disabled(), + "conversion_available": is_gotenberg_enabled(), "source_text": self.source_text, "source_url": self.source_url, } @@ -113,17 +128,41 @@ def load_pdf(self, *_args, **_kwargs): """Get the PDF block's settings in JSON format.""" return Response(json.dumps(self.raw_settings), content_type="application/json", charset="utf8") - @XBlock.json_handler - def save_pdf(self, data, suffix=""): # pylint: disable=unused-argument - """Save handler.""" - self.display_name = data["display_name"] - self.url = data["url"] - - if not is_all_download_disabled(): - self.allow_download = bool_from_str(data["allow_download"]) - self.source_text = data["source_text"] - self.source_url = data["source_url"] + def has_authoring_permissions(self) -> bool: + """ + Checks if the current user has authoring permissions. + """ + user_service = self.runtime.service(self, "user") + permissions_service = self.runtime.service(self, "studio_user_permissions") + if permissions_service and permissions_service.can_write(self.context_key): + return True + return user_service.get_current_user().opt_attrs.get("edx-platform.user_is_staff", False) - return { - "result": "success", - } + @XBlock.json_handler + def convert_pdf(self, data, suffix=""): # pylint: disable=unused-argument + """ + PDF Conversion handling. Basically just a frontend to the Gotenberg service which converts the given URL + and then saves it to course assets, returning the URL. + """ + if not self.has_authoring_permissions(): + return error_response( + {"error": _("You do not have permission to manage files for this block.")}, + status=403, + ) + if not is_gotenberg_enabled(): + return error_response({"error": _("Gotenberg not enabled. PDF Conversion unavailable.")}) + user_service = self.runtime.service(self, "user") + user_attrs = user_service.get_current_user().opt_attrs + user = get_user_model().objects.get(id=user_attrs.get("edx-platform.user_id")) + output_name = f"{self.scope_ids.usage_id}.pdf" + try: + file_bytes = fetch_source_asset(self.scope_ids.usage_id, data["url"]) + except (HTTPError, Timeout): + logger.exception(_("Failed to fetch document at %(url)r.") % {"url": data["url"]}) + return error_response({"error": _("Could not fetch source document.")}, status=502) + source_url = urlparse(data["url"]) + source_filename = source_url.path.split("/")[-1] + result = convert_to_pdf(source_filename, file_bytes, output_name) + if result is None: + return error_response({"error": _("PDF Conversion failed.")}, status=500) + return {"url": add_asset(self.scope_ids.usage_id, result, user)} diff --git a/xblock_pdf/tests/test_pdf.py b/xblock_pdf/tests/test_pdf.py index 114f08a3..6b8f57b0 100644 --- a/xblock_pdf/tests/test_pdf.py +++ b/xblock_pdf/tests/test_pdf.py @@ -1,21 +1,73 @@ """Tests for the PDF Block""" import json -from typing import Any +from dataclasses import dataclass +from typing import Any, TypedDict from unittest.mock import MagicMock, patch +import pytest +from django.contrib.auth.models import User from django.test import override_settings +from requests import Response +from requests.exceptions import HTTPError from xblock.field_data import DictFieldData from xblock.fields import ScopeIds from xblock.test.toy_runtime import ToyRuntime from xblock_pdf import PDFBlock +from xblock_pdf.utils import error_response +MockOptValues = TypedDict("MockOptValues", {"edx-platform.user_is_staff": bool, "edx-platform.user_id": int}) -def make_block(**fields: str) -> PDFBlock: + +@dataclass +class MockUser: + opt_attrs: MockOptValues + + +class ToyUserService: + """ + Toy version of the user service that implements just enough for us to work with. + """ + + def __init__(self, *, user_id: int, is_staff=False): + self._user = MockUser(opt_attrs={"edx-platform.user_is_staff": is_staff, "edx-platform.user_id": user_id}) + + def get_current_user(self): + return self._user + + +class ToyPermissionsService: + """ + Toy version of the studio_user_permissions service. + """ + + def __init__(self, can_read=True, can_write=False): + self._can_read = can_read + self._can_write = can_write + + def can_read(self, _context_key): + return self._can_read + + def can_write(self, _context_key): + return self._can_write + + +class ToyServiceRuntime(ToyRuntime): + """ + Modified toy runtime that includes custom services for mocking/testing. + """ + + def __init__(self, *, services: dict[str, Any] | None = None): + super().__init__() + if services is not None: + self._services.update(services) + + +def make_block(*, services: dict[str, Any] | None = None, **fields: str) -> PDFBlock: """Build a block with specific fields set.""" scope_ids = ScopeIds("1", "2", "3", "4") - return PDFBlock(ToyRuntime(), scope_ids=scope_ids, field_data=DictFieldData(data=fields)) + return PDFBlock(ToyServiceRuntime(services=services), scope_ids=scope_ids, field_data=DictFieldData(data=fields)) def get_student_content(block: PDFBlock) -> str: @@ -52,7 +104,7 @@ def test_download_button(): def test_source_url(): - """Test rendering based on whether or not there's a source URL""" + """Test rendering based on whether there's a source URL""" block = make_block() get_student_content(block) content = get_student_content(block) @@ -62,60 +114,6 @@ def test_source_url(): assert "Download the source document" in content -@override_settings(PDFXBLOCK_DISABLE_ALL_DOWNLOAD=False) -def test_saves_settings(): - """Test that PDF settings are saved.""" - block = make_block() - request = mock_handle_request( - { - "display_name": "Novel application of theory", - "url": "https://example.com/nature_article.pdf", - "allow_download": "false", - "source_text": "Get educated", - "source_url": "https://example.com/nature_article.tex", - } - ) - block.save_pdf(request) - assert block.display_name == "Novel application of theory" - assert block.url == "https://example.com/nature_article.pdf" - assert not block.allow_download - assert block.source_text == "Get educated" - assert block.source_url == "https://example.com/nature_article.tex" - - -@override_settings(PDFXBLOCK_DISABLE_ALL_DOWNLOAD=True) -def test_saves_settings_omits_on_download_disabled_flag(): - """ - Test that fields relating to download are ignored when the universal - downloads disabled flag is set. - """ - block = make_block() - request = mock_handle_request( - { - "display_name": "Novel application of theory", - "url": "https://example.com/nature_article.pdf", - # These fields shouldn't be visible on the front end, - # but should be dropped if they somehow are. - # - # Potential future improvement would be saving these - # but ignoring them when rendering. This is not currently - # the case since the fields are entirely absent from the studio - # render, and so would send blank data which would error out. - "allow_download": "false", - "source_text": "Get educated", - "source_url": "https://example.com/nature_article.tex", - } - ) - block.save_pdf(request) - assert block.display_name == "Novel application of theory" - assert block.url == "https://example.com/nature_article.pdf" - # Flag will be the default, which is True, even though download will be - # disabled in practice. - assert block.allow_download - assert block.source_text == "" - assert block.source_url == "" - - @patch.object(ToyRuntime, "publish") def test_download_event_fires(mock_publish): """Test that we fire a download event.""" @@ -138,3 +136,92 @@ def test_get_settings(): request = mock_handle_request({}, method="GET") result = json.loads(block.load_pdf(request).body) assert result["display_name"] == "PDF" + + +@override_settings(GOTENBERG_HOST=None) +def test_convert_pdf_fails_no_gotenberg(): + """ + Test that PDF conversion fails if Gotenberg is not available. + """ + block = make_block(services={"user": ToyUserService(is_staff=True, user_id=1)}) + request = mock_handle_request({"url": "https://example.com/thing.doc"}) + result = block.convert_pdf(request) + assert result.status_code == 400 + assert b"Gotenberg not enabled. PDF Conversion unavailable." in result.body + + +@override_settings(GOTENBERG_HOST="https://gotenberg/") +def test_convert_fails_not_staff(): + """ + Test that PDF conversion fails if user is not staff. + """ + block = make_block(services={"user": ToyUserService(is_staff=False, user_id=1)}) + request = mock_handle_request({"url": "https://example.com/thing.doc"}) + result = block.convert_pdf(request) + assert result.status_code == 403 + assert b"You do not have permission to manage files for this block." in result.body + + +@override_settings(GOTENBERG_HOST="https://gotenberg/") +@patch("xblock_pdf.pdf.logger") +@patch("xblock_pdf.pdf.fetch_source_asset") +@pytest.mark.django_db +def test_failed_fetch_logs(mock_fetch, mock_log): + block = make_block( + services={ + "user": ToyUserService( + is_staff=True, user_id=User.objects.create(username="beep", email="beep@example.com").id + ) + } + ) + mock_fetch.side_effect = HTTPError(response=error_response({"error": "Failed."}, status=400)) + request = mock_handle_request({"url": "https://example.com/thing.doc"}) + result = block.convert_pdf(request) + assert mock_log.exception.has_been_called() + assert result.status_code == 502 + assert b"Could not fetch source document." in result.body + + +@override_settings(GOTENBERG_HOST="https://gotenberg/") +@patch("xblock_pdf.utils.requests") +@patch("xblock_pdf.pdf.fetch_source_asset") +@pytest.mark.django_db +def test_failed_conversion(mock_fetch, mock_requests): + block = make_block( + services={ + "user": ToyUserService( + is_staff=True, user_id=User.objects.create(username="beep", email="beep@example.com").id + ) + } + ) + mock_fetch.return_value = b"beep" + mock_requests.return_value = error_response({"error": "Nope."}) + request = mock_handle_request({"url": "https://example.com/thing.doc"}) + result = block.convert_pdf(request) + assert result.status_code == 500 + assert b"PDF Conversion failed." in result.body + + +@override_settings(GOTENBERG_HOST="https://gotenberg/") +@patch("xblock_pdf.pdf.add_asset") +@patch("xblock_pdf.utils.requests") +@patch("xblock_pdf.pdf.fetch_source_asset") +@pytest.mark.django_db +def test_successful_conversion_with_perms_service(mock_fetch, mock_requests, mock_add_asset): + block = make_block( + services={ + "user": ToyUserService( + is_staff=False, user_id=User.objects.create(username="beep", email="beep@example.com").id + ), + "studio_user_permissions": ToyPermissionsService(can_write=True), + } + ) + mock_fetch.return_value = b"beep" + mock_response = Response() + mock_response.__setstate__({"status_code": 200, "_content": b"boop"}) + mock_requests.post.return_value = mock_response + mock_add_asset.return_value = "https://example.com/exported.pdf" + request = mock_handle_request({"url": "https://example.com/thing.doc"}) + result = block.convert_pdf(request) + assert result.status_code == 200 + assert b"https://example.com/exported.pdf" in result.body diff --git a/xblock_pdf/utils.py b/xblock_pdf/utils.py index 079c58b0..ac4737bf 100644 --- a/xblock_pdf/utils.py +++ b/xblock_pdf/utils.py @@ -1,13 +1,128 @@ """Utility functions for PDF XBlock.""" +import json +from io import BytesIO +from logging import getLogger +from typing import Any + +import requests from django.conf import settings +from django.contrib.auth.models import AbstractBaseUser +from django.core.files.uploadedfile import InMemoryUploadedFile +from opaque_keys.edx.locator import BlockUsageLocator, LibraryUsageLocatorV2 +from webob import Response + +logger = getLogger(__name__) + + +def is_gotenberg_enabled() -> bool: + """ + Returns if gotenberg is enabled. + """ + return bool(get_gotenberg_host()) + + +def get_gotenberg_host() -> str | None: + """ + Returns the hostname of the Gotenberg instance, if configured. + Returns None if Gotenberg is not configured. + """ + return getattr(settings, "GOTENBERG_HOST", None) + + +def get_conversion_url() -> str | None: + """ + Get the URL for sending a document for conversion by Gotenberg + """ + return (base_url := get_gotenberg_host()) and f"{base_url}/forms/libreoffice/convert" + +def add_asset( + location: BlockUsageLocator | LibraryUsageLocatorV2, + # Must have the 'name' attribute set. + asset: InMemoryUploadedFile, + user: AbstractBaseUser, +) -> str: # pragma: no cover + """ + Adds an asset for this block. If we aren't in the studio environment, will create ImportErrors. + Easily mocked for tests. + """ + from cms.djangoapps.contentstore.asset_storage_handlers import update_course_run_asset + from openedx.core.djangoapps.content_libraries.api import add_library_block_static_asset_file -def bool_from_str(str_value): - """Convert string from submitted form to boolean.""" - return str_value.strip().lower() == "true" + match location: + case BlockUsageLocator(): + asset = update_course_run_asset(location.course_key, asset) + return asset.get_static_path_from_location(asset.location) + case LibraryUsageLocatorV2(): + # Static assets with a /static/ prefix rendered by the backend have their paths + # translated to the paths for loading library assets. + path = f"static/{asset.name}" + # The path must be stored without the leading slash. + add_library_block_static_asset_file(location, path, asset.read(), user) + return "/" + path + + +def fetch_source_asset( + location: BlockUsageLocator | LibraryUsageLocatorV2, source_url: str +) -> bytes: # pragma: no cover + """ + Fetch a source asset and return its bytes. When using a full URL, pull it via requests. When using + an absolute URL in a library, find the relevant asset and return its bytes. + """ + from openedx.core.djangoapps.content_libraries import api as libraries_api + from openedx_content import api as content_api + + if not source_url.startswith("/"): + response = requests.get(source_url, timeout=(10, 120)) + response.raise_for_status() + return response.content + match location: + case BlockUsageLocator(): + response = requests.get(source_url, timeout=(10, 120)) + response.raise_for_status() + return response.content + case LibraryUsageLocatorV2(): + version_uuid = libraries_api.get_component_from_usage_key(location).versioning.draft.uuid + component_version = content_api.get_component_version_by_uuid(version_uuid) + media = component_version.componentversionmedia_set.get(path=source_url[1:]).media + with media.read_file() as f: + return f.read() + + +def convert_to_pdf(source_filename: str, data: bytes, filename: str) -> InMemoryUploadedFile | None: + """ + Uses the Gotenberg service to convert the document at `doc_url` to a PDF file. + """ + if not (conversion_url := get_conversion_url()): # pragma: no cover + return None + + pdf_response = requests.post(conversion_url, files={"file": (source_filename, data)}, timeout=(2, 120)) + if pdf_response.status_code != 200: + logger.error(f"Gotenberg error: {pdf_response.content}") + return None + return InMemoryUploadedFile( + file=BytesIO(pdf_response.content), + field_name="file", + content_type="application/pdf", + size=len(pdf_response.content), + charset=None, + name=filename, + ) def is_all_download_disabled(): """Check if all downloads are disabled or not.""" return getattr(settings, "PDFXBLOCK_DISABLE_ALL_DOWNLOAD", False) + + +def error_response(data: dict[Any, Any], status: int = 400): + """ + Returns a JSON response object with the appropriate status. + """ + return Response( + json.dumps(data), + status=status, + content_type="application/json", + charset="utf8", + )