feat: gotenberg support for PDF block

Author: Kelketek

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)}

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.pdf.convert_to_pdf")
+@patch("xblock_pdf.pdf.fetch_source_asset")
+@pytest.mark.django_db
+def test_failed_conversion(mock_fetch, mock_convert):
+    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_convert.return_value = None
+    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