Embed raw Perseus questions as QTI custom interactions

Teach QTIExerciseGenerator to intercept PERSEUS_QUESTION items in
process_assessment_item and emit each as a qti-custom-interaction: write the
Perseus JSON (content-storage refs rewritten to the packaged asset paths) and
its image/graphie assets into the package, and declare them as dependencies of
the wrapper item's manifest resource. Removes the old ValueError rejection of
Perseus questions in the QTI format.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rtibblesbot

contentcuration/contentcuration/tests/utils/test_exercise_creation.py

@@ -7,6 +7,7 @@
 import re
 import zipfile
 from io import BytesIO
+from tempfile import TemporaryDirectory
 from uuid import uuid4
 
 from django.core.files.storage import default_storage as storage
@@ -27,6 +28,7 @@
 from contentcuration.utils.assessment.qti.archive import hex_to_qti_id
 from contentcuration.utils.assessment.qti.archive import QTIExerciseGenerator
 from contentcuration.utils.assessment.qti.validation import parse_qti_xml
+from contentcuration.utils.assessment.qti.validation import validate_qti_item
 
 
 class TestPerseusExerciseCreation(StudioTestCase):
@@ -480,6 +482,45 @@ def _create_perseus_item(self):
 
         return item, graphie_files
 
+    def test_write_raw_perseus_assets_returns_paths_and_writes_files(self):
+        """`_write_raw_perseus_assets` writes an item's images/graphie assets into
+        the given directory and returns their package-relative paths."""
+        image_file = fileobj_exercise_image()
+        graphie_file = fileobj_exercise_graphie(original_filename="mygraphie")
+
+        item = AssessmentItem.objects.create(
+            contentnode=self.exercise_node,
+            assessment_id="fedcba0987654321fedcba0987654321",
+            type=exercises.PERSEUS_QUESTION,
+            raw_data="{}",
+            order=1,
+            randomize=True,
+        )
+        image_file.assessment_item = item
+        image_file.save()
+        graphie_file.assessment_item = item
+        graphie_file.save()
+
+        generator = PerseusExerciseGenerator(
+            self.exercise_node, {}, self.channel.id, "en-US", user_id=self.user.id
+        )
+        with TemporaryDirectory() as tempdir:
+            generator.tempdir = tempdir
+            written = generator._write_raw_perseus_assets(item, "perseus/images")
+
+            image_path = (
+                f"perseus/images/{image_file.checksum}.{image_file.file_format_id}"
+            )
+            svg_path = f"perseus/images/{graphie_file.original_filename}.svg"
+            json_path = f"perseus/images/{graphie_file.original_filename}-data.json"
+
+            self.assertIn(image_path, written)
+            self.assertIn(svg_path, written)
+            self.assertIn(json_path, written)
+
+            for path in (image_path, svg_path, json_path):
+                self.assertTrue(os.path.exists(os.path.join(tempdir, path)))
+
     def test_exercise_with_graphie(self):
         """Test creating an exercise with graphie files (SVG+JSON pairs)"""
 
@@ -1454,32 +1495,83 @@ def test_qti_exercise_without_hints_produces_no_catalog_info(self):
         item_xml = self._render_single_item_xml("abcdef1234567890abcdef1234567890", [])
         self.assertNotIn("<qti-catalog-info", item_xml)
 
-    def test_perseus_question_rejection(self):
-        """Test that Perseus questions are properly rejected"""
-        assessment_id = "aaaa1111bbbb2222cccc3333dddd4444"
-        # Create a mock Perseus question
-        item = AssessmentItem.objects.create(
+    def test_perseus_custom_interaction_embedded_with_native_qti(self):
+        """A node mixing a native QTI item and a raw Perseus question yields one
+        QTI package: the native item, plus the Perseus question wrapped as a
+        ``qti-custom-interaction`` with its JSON and image packaged and declared."""
+        native_id = "1234567890abcdef1234567890abcdef"
+        native_item = self._create_native_qti_item(
+            VALID_CHOICE_ITEM, assessment_id=native_id
+        )
+
+        image_file = fileobj_exercise_image()
+        image_url = exercises.CONTENT_STORAGE_FORMAT.format(image_file.filename())
+        perseus_id = "aaaa1111bbbb2222cccc3333dddd4444"
+        perseus_item = AssessmentItem.objects.create(
             contentnode=self.exercise_node,
-            assessment_id=assessment_id,
+            assessment_id=perseus_id,
             type=exercises.PERSEUS_QUESTION,
-            raw_data='{"question": {"content": "Perseus content"}}',
-            order=1,
+            raw_data=json.dumps(
+                {"question": {"content": f"See ![shape]({image_url})", "images": {}}},
+                ensure_ascii=False,
+            ),
+            order=2,
+            randomize=False,
         )
+        image_file.assessment_item = perseus_item
+        image_file.save()
 
         exercise_data = {
             "mastery_model": exercises.M_OF_N,
             "randomize": True,
             "n": 1,
             "m": 1,
-            "all_assessment_items": [item.assessment_id],
-            "assessment_mapping": {item.assessment_id: exercises.PERSEUS_QUESTION},
+            "all_assessment_items": [
+                native_item.assessment_id,
+                perseus_item.assessment_id,
+            ],
+            "assessment_mapping": {
+                native_item.assessment_id: exercises.QTI,
+                perseus_item.assessment_id: exercises.PERSEUS_QUESTION,
+            },
         }
 
-        # Should raise ValueError for Perseus questions
-        with self.assertRaises(ValueError) as context:
-            self._create_qti_zip(exercise_data)
+        self._create_qti_zip(exercise_data)
+        exercise_file = self.exercise_node.files.get(preset_id=format_presets.QTI_ZIP)
+        zip_file = self._validate_qti_zip_structure(exercise_file)
+
+        namelist = zip_file.namelist()
+        perseus_item_path = f"items/{hex_to_qti_id(perseus_id)}.xml"
+        perseus_json_path = f"perseus/{perseus_id}.json"
+        image_path = f"perseus/images/{image_file.filename()}"
+
+        # Native QTI item and the Perseus wrapper item both present.
+        self.assertIn("items/item_1.xml", namelist)
+        self.assertIn(perseus_item_path, namelist)
+        # Perseus JSON and its image packaged.
+        self.assertIn(perseus_json_path, namelist)
+        self.assertIn(image_path, namelist)
+
+        # The wrapper item validates and references the Perseus JSON.
+        wrapper_xml = zip_file.read(perseus_item_path).decode("utf-8")
+        self.assertTrue(validate_qti_item(wrapper_xml.encode("utf-8")).is_valid)
+        parsed = parse_qti_xml(wrapper_xml.encode("utf-8"))
+        custom = parsed.getroot().iter("{*}qti-custom-interaction")
+        custom_el = next(custom)
+        self.assertEqual(custom_el.get("data-type"), "perseus")
+        self.assertEqual(custom_el.get("data-perseus-path"), perseus_json_path)
+
+        # The Perseus JSON's image reference was rewritten to the packaged path.
+        packaged_json = zip_file.read(perseus_json_path).decode("utf-8")
+        self.assertIn(
+            f"${exercises.IMG_PLACEHOLDER}/perseus/images/{image_file.filename()}",
+            packaged_json,
+        )
 
-        self.assertIn("Perseus questions are not supported", str(context.exception))
+        # The manifest lists the JSON and image as files of the perseus resource.
+        manifest_xml = zip_file.read("imsmanifest.xml").decode("utf-8")
+        self.assertIn(f'<file href="{perseus_json_path}" />', manifest_xml)
+        self.assertIn(f'<file href="{image_path}" />', manifest_xml)
 
     def test_exercise_with_image(self):
         """Test QTI exercise generation with images"""

contentcuration/contentcuration/utils/assessment/qti/archive.py

@@ -11,6 +11,9 @@
 
 from contentcuration.utils.assessment.base import ExerciseArchiveGenerator
 from contentcuration.utils.assessment.qti.constants import ResourceType
+from contentcuration.utils.assessment.qti.convert import (
+    build_perseus_custom_interaction_item,
+)
 from contentcuration.utils.assessment.qti.convert import (
     convert_legacy_assessment_item_to_qti,
 )
@@ -43,6 +46,8 @@ class QTIExerciseGenerator(ExerciseArchiveGenerator):
     file_format = "zip"
     preset = format_presets.QTI_ZIP
 
+    PERSEUS_IMAGE_DIR = "perseus/images"
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.qti_resources: List[QTIResource] = []
@@ -61,6 +66,16 @@ def get_image_ref_prefix(self):
     def _qti_item_filepath(self, assessment_id):
         return f"items/{assessment_id}.xml"
 
+    def _node_language(self):
+        return (
+            self.ccnode.language.lang_code
+            if self.ccnode.language
+            else self.default_language
+        )
+
+    def _next_item_title(self):
+        return f"{self.ccnode.title} {len(self.qti_resources) + 1}"
+
     def _add_resource(self, resource: QTIResource) -> None:
         if any(r.identifier == resource.identifier for r in self.qti_resources):
             raise ValueError(
@@ -136,6 +151,44 @@ def _write_qti_media_files(self, assessment_item) -> Tuple[str, List[str]]:
         item_xml = rewrite_qti_media_paths(raw_data, path_by_filename)
         return item_xml, sorted(path_by_filename.values())
 
+    def process_assessment_item(self, assessment_item):
+        if assessment_item.type == exercises.PERSEUS_QUESTION:
+            return self._create_perseus_custom_interaction(assessment_item)
+        return super().process_assessment_item(assessment_item)
+
+    def _create_perseus_custom_interaction(self, assessment_item) -> None:
+        """Embed a raw Perseus question as a ``qti-custom-interaction``.
+
+        Writes the Perseus JSON (with its content-storage references rewritten to
+        the packaged asset paths) and its image/graphie assets into the package,
+        and declares them as dependencies of the wrapper item's manifest resource.
+        """
+        asset_paths = self._write_raw_perseus_assets(
+            assessment_item, self.PERSEUS_IMAGE_DIR
+        )
+        perseus_json = self._rewrite_content_storage_refs(
+            assessment_item.raw_data, self.PERSEUS_IMAGE_DIR
+        )
+        perseus_path = f"perseus/{assessment_item.assessment_id}.json"
+        self.add_file_to_write(perseus_path, perseus_json.encode("utf-8"))
+
+        result = build_perseus_custom_interaction_item(
+            assessment_item.assessment_id,
+            perseus_path,
+            self._next_item_title(),
+            self._node_language(),
+        )
+
+        item_path = self._qti_item_filepath(result.identifier)
+        self.add_file_to_write(item_path, result.xml.encode("utf-8"))
+        self._add_resource(
+            QTIResource(
+                identifier=result.identifier,
+                filepath=item_path,
+                file_dependencies=[perseus_path, *asset_paths],
+            )
+        )
+
     def create_assessment_item(
         self, assessment_item, processed_data: Dict[str, Any]
     ) -> Optional[Tuple[str, bytes]]:
@@ -144,26 +197,15 @@ def create_assessment_item(
         if assessment_item.type == exercises.QTI:
             return self._create_native_qti_item(assessment_item)
 
-        # Skip Perseus questions as they can't be easily converted
-        if assessment_item.type == exercises.PERSEUS_QUESTION:
-            raise ValueError(
-                f"Perseus questions are not supported in QTI format: {assessment_item.assessment_id}"
-            )
-
-        language = (
-            self.ccnode.language.lang_code
-            if self.ccnode.language
-            else self.default_language
-        )
         legacy_item = LegacyAssessmentItem(
             type=assessment_item.type,
             question=processed_data["question"],
             answers=processed_data.get("answers", []),
             hints=processed_data.get("hints", []),
             randomize=processed_data.get("randomize", False),
             assessment_id=assessment_item.assessment_id,
-            title=f"{self.ccnode.title} {len(self.qti_resources) + 1}",
-            language=language,
+            title=self._next_item_title(),
+            language=self._node_language(),
         )
         result = convert_legacy_assessment_item_to_qti(legacy_item)