From 02c4226d1e56c5fa60bc3bd1506adf4e5c6c3628 Mon Sep 17 00:00:00 2001 From: rtibblesbot Date: Fri, 3 Jul 2026 11:04:38 -0700 Subject: [PATCH 1/4] feat: package native QTI items from raw_data into QTI/IMSCP archives Route type=QTI assessment items through publish verbatim instead of rebuilding item XML from structured fields, since their authored XML is already stored in raw_data. - Validate raw_data against the QTI schema before packaging - Resolve and write referenced media alongside the item - Extract the media-reference scanner into its own module, shared with the AssessmentItem viewset - Refactor QTIExerciseGenerator's manifest bookkeeping to a generator-agnostic resource list so both the pydantic-built and native paths populate it Co-Authored-By: Claude Sonnet 5 --- .../tests/utils/qti/test_media.py | 32 ++++ .../tests/utils/test_exercise_creation.py | 168 ++++++++++++++++++ .../utils/assessment/qti/archive.py | 107 +++++++++-- .../utils/assessment/qti/media.py | 45 +++++ .../viewsets/assessmentitem.py | 32 +--- 5 files changed, 339 insertions(+), 45 deletions(-) create mode 100644 contentcuration/contentcuration/tests/utils/qti/test_media.py create mode 100644 contentcuration/contentcuration/utils/assessment/qti/media.py diff --git a/contentcuration/contentcuration/tests/utils/qti/test_media.py b/contentcuration/contentcuration/tests/utils/qti/test_media.py new file mode 100644 index 0000000000..1e204b43c6 --- /dev/null +++ b/contentcuration/contentcuration/tests/utils/qti/test_media.py @@ -0,0 +1,32 @@ +from contentcuration.utils.assessment.qti.media import get_qti_media_references + +CHECKSUM_A = "a" * 32 +CHECKSUM_B = "b" * 32 + + +def test_extracts_checksum_from_src_href_data(): + xml = ( + f'' + f'x' + f'' + ) + assert get_qti_media_references(xml) == {f"{CHECKSUM_A}.png", f"{CHECKSUM_B}.pdf"} + + +def test_extracts_checksums_from_srcset(): + xml = f'' + assert get_qti_media_references(xml) == {f"{CHECKSUM_A}.png", f"{CHECKSUM_B}.png"} + + +def test_ignores_non_checksum_values(): + xml = 'x' + assert get_qti_media_references(xml) == set() + + +def test_returns_empty_set_for_malformed_xml(): + assert get_qti_media_references("") == set() + + +def test_accepts_bytes(): + xml = f''.encode("utf-8") + assert get_qti_media_references(xml) == {f"{CHECKSUM_A}.png"} diff --git a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py index ae6a295e41..2c92d47616 100644 --- a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py +++ b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py @@ -20,6 +20,8 @@ from contentcuration.tests.base import StudioTestCase from contentcuration.tests.testdata import fileobj_exercise_graphie from contentcuration.tests.testdata import fileobj_exercise_image +from contentcuration.tests.utils.qti.test_validation import _item_xml +from contentcuration.tests.utils.qti.test_validation import VALID_CHOICE_ITEM from contentcuration.utils.assessment.perseus import PerseusExerciseGenerator from contentcuration.utils.assessment.qti.archive import hex_to_qti_id from contentcuration.utils.assessment.qti.archive import QTIExerciseGenerator @@ -1231,6 +1233,20 @@ class TestQTIExerciseCreation(StudioTestCase): maxDiff = None + NATIVE_ITEM_XML = _item_xml( + "native_item_1", + "Native Item", + '' + "choice_0" + "", + 'Pick one. ' + 'diagram' + 'A' + 'B' + "", + ) + def setUp(self): self.setUpBase() @@ -1278,6 +1294,20 @@ def _create_assessment_item( ) return item + def _create_native_qti_item(self, raw_data, assessment_id=None): + """Helper to create a native type=QTI assessment item with the given raw_data.""" + return AssessmentItem.objects.create( + contentnode=self.exercise_node, + assessment_id=assessment_id or uuid4().hex, + type=exercises.QTI, + question="", + answers="[]", + hints="[]", + raw_data=raw_data, + order=len(self.exercise_node.assessment_items.all()) + 1, + randomize=False, + ) + def _create_qti_zip(self, exercise_data): """Create QTI exercise zip using the generator""" generator = QTIExerciseGenerator( @@ -2107,3 +2137,141 @@ def test_input_question(self): self._normalize_xml(expected_item_xml), self._normalize_xml(actual_item_xml), ) + + def test_native_qti_item_written_verbatim(self): + """The item XML in the zip must byte-match the authored raw_data.""" + raw_data = _item_xml( + "native_item_1", + "Native Item", + '' + "choice_0" + "", + 'Pick one.' + 'A' + 'B' + "", + ) + item = self._create_native_qti_item(raw_data) + exercise_data = { + "mastery_model": exercises.M_OF_N, + "randomize": True, + "n": 5, + "m": 3, + "all_assessment_items": [item.assessment_id], + "assessment_mapping": {item.assessment_id: exercises.QTI}, + } + 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) + self.assertIn("items/native_item_1.xml", zip_file.namelist()) + self.assertEqual( + zip_file.read("items/native_item_1.xml").decode("utf-8"), raw_data + ) + + def test_native_qti_item_media_included_and_addressed(self): + # fileobj_exercise_image() writes real bytes to storage keyed by their + # actual md5 checksum + "jpg" ext -- use that real checksum/ext rather + # than a hardcoded one, or _write_qti_media_files' storage.open() call + # finds nothing there (see test_exercise_with_image, same file). + image_file = fileobj_exercise_image() + raw_data = self.NATIVE_ITEM_XML.format( + checksum=image_file.checksum, ext=image_file.file_format_id + ) + item = self._create_native_qti_item(raw_data) + image_file.assessment_item = item + image_file.save() + exercise_data = { + "mastery_model": exercises.M_OF_N, + "randomize": True, + "n": 5, + "m": 3, + "all_assessment_items": [item.assessment_id], + "assessment_mapping": {item.assessment_id: exercises.QTI}, + } + 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) + media_filename = f"{image_file.checksum}.{image_file.file_format_id}" + self.assertIn(f"items/{media_filename}", zip_file.namelist()) + manifest = zip_file.read("imsmanifest.xml").decode("utf-8") + self.assertIn(media_filename, manifest) + + def test_native_qti_item_invalid_raw_data_raises(self): + invalid_raw_data = VALID_CHOICE_ITEM.replace( + 'orientation="vertical"', 'orientation="sideways"' + ) + item = self._create_native_qti_item(invalid_raw_data) + exercise_data = { + "mastery_model": exercises.M_OF_N, + "randomize": True, + "n": 5, + "m": 3, + "all_assessment_items": [item.assessment_id], + "assessment_mapping": {item.assessment_id: exercises.QTI}, + } + with self.assertRaises(ValueError): + self._create_qti_zip(exercise_data) + self.assertFalse( + self.exercise_node.files.filter(preset_id=format_presets.QTI_ZIP).exists() + ) + + def test_native_qti_item_missing_media_file_raises(self): + raw_data = self.NATIVE_ITEM_XML.format(checksum="b" * 32, ext="png") + item = self._create_native_qti_item(raw_data) # no File row linked + exercise_data = { + "mastery_model": exercises.M_OF_N, + "randomize": True, + "n": 5, + "m": 3, + "all_assessment_items": [item.assessment_id], + "assessment_mapping": {item.assessment_id: exercises.QTI}, + } + with self.assertRaises(ValueError): + self._create_qti_zip(exercise_data) + + def test_native_qti_duplicate_identifier_raises(self): + item1 = self._create_native_qti_item(VALID_CHOICE_ITEM) + item2 = self._create_native_qti_item( + VALID_CHOICE_ITEM + ) # same "item_1" identifier + exercise_data = { + "mastery_model": exercises.M_OF_N, + "randomize": True, + "n": 5, + "m": 3, + "all_assessment_items": [item1.assessment_id, item2.assessment_id], + "assessment_mapping": { + item1.assessment_id: exercises.QTI, + item2.assessment_id: exercises.QTI, + }, + } + with self.assertRaises(ValueError): + self._create_qti_zip(exercise_data) + + def test_republish_replaces_stale_native_qti_archive(self): + item = self._create_native_qti_item(VALID_CHOICE_ITEM) + exercise_data = { + "mastery_model": exercises.M_OF_N, + "randomize": True, + "n": 5, + "m": 3, + "all_assessment_items": [item.assessment_id], + "assessment_mapping": {item.assessment_id: exercises.QTI}, + } + self._create_qti_zip(exercise_data) + first_checksum = self.exercise_node.files.get( + preset_id=format_presets.QTI_ZIP + ).checksum + + item.raw_data = VALID_CHOICE_ITEM.replace("Sample Item", "Renamed Item") + item.save() + self._create_qti_zip(exercise_data) + + self.assertEqual( + self.exercise_node.files.filter(preset_id=format_presets.QTI_ZIP).count(), 1 + ) + second_checksum = self.exercise_node.files.get( + preset_id=format_presets.QTI_ZIP + ).checksum + self.assertNotEqual(first_checksum, second_checksum) diff --git a/contentcuration/contentcuration/utils/assessment/qti/archive.py b/contentcuration/contentcuration/utils/assessment/qti/archive.py index 7574192a61..18aa6c151e 100644 --- a/contentcuration/contentcuration/utils/assessment/qti/archive.py +++ b/contentcuration/contentcuration/utils/assessment/qti/archive.py @@ -1,4 +1,5 @@ import base64 +from dataclasses import dataclass from typing import Any from typing import Dict from typing import List @@ -37,7 +38,10 @@ from contentcuration.utils.assessment.qti.interaction_types.text_based import ( TextEntryInteraction, ) +from contentcuration.utils.assessment.qti.media import get_qti_media_references from contentcuration.utils.assessment.qti.prompt import Prompt +from contentcuration.utils.assessment.qti.validation import parse_qti_xml +from contentcuration.utils.assessment.qti.validation import validate_qti_item choice_interactions = { @@ -56,6 +60,13 @@ def hex_to_qti_id(hex_string): return f"K{base64.urlsafe_b64encode(bytes_data).decode('ascii').rstrip('=')}" +@dataclass +class QTIResource: + identifier: str + filepath: str + file_dependencies: List[str] + + class QTIExerciseGenerator(ExerciseArchiveGenerator): """ Exercise zip generator for QTI format exercises. @@ -67,7 +78,7 @@ class QTIExerciseGenerator(ExerciseArchiveGenerator): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) - self.qti_items = [] + self.qti_resources: List[QTIResource] = [] def get_image_file_path(self) -> str: """Get the file path for QTI assessment items.""" @@ -179,11 +190,76 @@ def _create_text_entry_interaction_and_response( def _qti_item_filepath(self, assessment_id): return f"items/{assessment_id}.xml" + def _add_resource(self, resource: QTIResource) -> None: + if any(r.identifier == resource.identifier for r in self.qti_resources): + raise ValueError( + f"Duplicate QTI item identifier '{resource.identifier}' on node {self.ccnode.pk}" + ) + self.qti_resources.append(resource) + + def _create_native_qti_item(self, assessment_item) -> tuple[str, bytes]: + raw_bytes = assessment_item.raw_data.encode("utf-8") + + result = validate_qti_item(raw_bytes) + if not result.is_valid: + error_messages = "; ".join( + f"line {e.line}, column {e.column}: {e.message}" for e in result.errors + ) + raise ValueError( + f"QTI item {assessment_item.assessment_id} on node {self.ccnode.pk} " + f"failed schema validation: {error_messages}" + ) + + identifier = parse_qti_xml(raw_bytes).getroot().get("identifier") + if not identifier: + raise ValueError( + f"QTI item {assessment_item.assessment_id} is missing a root identifier attribute" + ) + + filepath = f"items/{identifier}.xml" + file_dependencies = self._write_qti_media_files(assessment_item) + + self._add_resource( + QTIResource( + identifier=identifier, + filepath=filepath, + file_dependencies=file_dependencies, + ) + ) + + return filepath, raw_bytes + + def _write_qti_media_files(self, assessment_item) -> List[str]: + filenames = get_qti_media_references(assessment_item.raw_data) + if not filenames: + return [] + + # assessment_item.files is prefetched by ExerciseArchiveGenerator.create_exercise_archive + files_by_name = { + f"{f.checksum}.{f.file_format_id}": f for f in assessment_item.files.all() + } + missing = sorted(filenames - files_by_name.keys()) + if missing: + raise ValueError( + f"QTI item {assessment_item.assessment_id} references media files " + f"not linked to it: {', '.join(missing)}" + ) + + sorted_filenames = sorted(filenames) + for filename in sorted_filenames: + file_obj = files_by_name[filename] + self._add_original_image(file_obj.checksum, filename, "items") + + return sorted_filenames + def create_assessment_item( self, assessment_item, processed_data: Dict[str, Any] ) -> tuple[str, bytes]: """Create QTI assessment item XML.""" + 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( @@ -227,7 +303,7 @@ def create_assessment_item( # Create the assessment item qti_item = AssessmentItem( identifier=qti_item_id, - title=f"{self.ccnode.title} {len(self.qti_items) + 1}", + title=f"{self.ccnode.title} {len(self.qti_resources) + 1}", language=language, adaptive=False, time_dependent=False, @@ -237,9 +313,6 @@ def create_assessment_item( response_processing=response_processing, ) - # Store for manifest creation - self.qti_items.append(qti_item) - # Generate XML content xml_content = qti_item.to_xml_string() @@ -247,26 +320,30 @@ def create_assessment_item( full_xml = f'\n{xml_content}' filename = self._qti_item_filepath(qti_item_id) + + self._add_resource( + QTIResource( + identifier=qti_item.identifier, + filepath=filename, + file_dependencies=qti_item.get_file_dependencies(), + ) + ) + return filename, full_xml.encode("utf-8") def _create_manifest_resources(self) -> List[Resource]: """Create manifest resources for all QTI items.""" resources = [] - for qti_item in self.qti_items: - # Get file dependencies (images, etc.) - file_dependencies = qti_item.get_file_dependencies() - - # Create file entries - qti_item_filepath = self._qti_item_filepath(qti_item.identifier) - files = [ManifestFile(href=qti_item_filepath)] - for dep in file_dependencies: + for qti_resource in self.qti_resources: + files = [ManifestFile(href=qti_resource.filepath)] + for dep in qti_resource.file_dependencies: files.append(ManifestFile(href=dep)) resource = Resource( - identifier=qti_item.identifier, + identifier=qti_resource.identifier, type_=ResourceType.ASSESSMENT_ITEM.value, - href=qti_item_filepath, + href=qti_resource.filepath, files=files, ) resources.append(resource) diff --git a/contentcuration/contentcuration/utils/assessment/qti/media.py b/contentcuration/contentcuration/utils/assessment/qti/media.py new file mode 100644 index 0000000000..0522d7fe59 --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/qti/media.py @@ -0,0 +1,45 @@ +import re + +from lxml import etree + +from contentcuration.utils.assessment.qti.fields import ( + entry_pattern as srcset_entry_pattern, +) +from contentcuration.utils.assessment.qti.validation import parse_qti_xml + +QTI_REFERENCE_ATTRIBUTES = ("src", "href", "data") +QTI_CHECKSUM_FILENAME_REGEX = re.compile(r"^[a-f0-9]{32}\.[0-9a-z]+$") +QTI_MEDIA_REFERENCE_XPATH = etree.XPath( + "//*[" + + " or ".join(f"@{attribute}" for attribute in QTI_REFERENCE_ATTRIBUTES) + + " or @srcset]" +) + + +def get_qti_media_references(raw_data): + """ + Scan QTI item XML for . media references in src/href/data/srcset + attributes, matching the TipTap editor's permanentSrc="." convention. + """ + if isinstance(raw_data, str): + raw_data = raw_data.encode("utf-8") + checksums = set() + try: + doc = parse_qti_xml(raw_data) + except etree.XMLSyntaxError: + return checksums + for element in QTI_MEDIA_REFERENCE_XPATH(doc): + candidates = [ + element.attrib.get(attribute) for attribute in QTI_REFERENCE_ATTRIBUTES + ] + srcset = element.attrib.get("srcset") + if srcset: + candidates += [ + entry[0].strip() for entry in re.findall(srcset_entry_pattern, srcset) + ] + checksums.update( + value + for value in candidates + if value and QTI_CHECKSUM_FILENAME_REGEX.match(value) + ) + return checksums diff --git a/contentcuration/contentcuration/viewsets/assessmentitem.py b/contentcuration/contentcuration/viewsets/assessmentitem.py index 961c7eaf88..30678b9570 100644 --- a/contentcuration/contentcuration/viewsets/assessmentitem.py +++ b/contentcuration/contentcuration/viewsets/assessmentitem.py @@ -4,7 +4,6 @@ from django.db import transaction from le_utils.constants import exercises from le_utils.constants import format_presets -from lxml import etree from rest_framework import serializers from rest_framework.permissions import IsAuthenticated from rest_framework.serializers import ValidationError @@ -13,10 +12,7 @@ from contentcuration.models import ContentNode from contentcuration.models import File from contentcuration.models import generate_object_storage_name -from contentcuration.utils.assessment.qti.fields import ( - entry_pattern as srcset_entry_pattern, -) -from contentcuration.utils.assessment.qti.validation import parse_qti_xml +from contentcuration.utils.assessment.qti.media import get_qti_media_references from contentcuration.utils.assessment.qti.validation import validate_qti_item from contentcuration.viewsets.base import BulkCreateMixin from contentcuration.viewsets.base import BulkListSerializer @@ -64,32 +60,8 @@ def get_filenames_from_assessment(assessment_item): ) -qti_reference_attributes = ("src", "href", "data") -qti_checksum_filename_regex = re.compile(r"^[a-f0-9]{32}\.[0-9a-z]+$") - - def get_filenames_from_qti_item(assessment_item): - # QTI attribute values are bare . references, matching - # the TipTap editor's existing permanentSrc="." convention - # for round-tripped tags -- not the markdown-only - # ${CONTENT_STORAGE}/. placeholder legacy types use. - checksums = set() - try: - doc = parse_qti_xml(assessment_item.raw_data.encode("utf-8")) - except etree.XMLSyntaxError: - return checksums - for element in doc.xpath("//*[@src or @href or @data or @srcset]"): - for attribute in qti_reference_attributes: - value = element.attrib.get(attribute) - if value and qti_checksum_filename_regex.match(value): - checksums.add(value) - srcset = element.attrib.get("srcset") - if srcset: - for entry in re.findall(srcset_entry_pattern, srcset): - token = entry[0].strip() - if token and qti_checksum_filename_regex.match(token): - checksums.add(token) - return checksums + return get_qti_media_references(assessment_item.raw_data) class AssessmentListSerializer(BulkListSerializer): From 26b4b4a3aeafb61204218bb2b2f1ca9124bcda47 Mon Sep 17 00:00:00 2001 From: rtibblesbot Date: Fri, 3 Jul 2026 11:05:08 -0700 Subject: [PATCH 2/4] feat: route nodes without perseus_question items to QTI packaging Dispatch to QTIExerciseGenerator unless a node has a perseus_question item, instead of only when it has a free_response item. QTIExerciseGenerator already synthesizes valid QTI XML for every legacy structured-field type; perseus_question is the only type it can't represent. Co-Authored-By: Claude Sonnet 5 --- .../tests/test_exportchannel.py | 93 ++++++++++++++++++- .../contentcuration/utils/publish.py | 10 +- 2 files changed, 95 insertions(+), 8 deletions(-) diff --git a/contentcuration/contentcuration/tests/test_exportchannel.py b/contentcuration/contentcuration/tests/test_exportchannel.py index 786c4ccec1..8fd3b03e7a 100644 --- a/contentcuration/contentcuration/tests/test_exportchannel.py +++ b/contentcuration/contentcuration/tests/test_exportchannel.py @@ -35,6 +35,7 @@ from .testdata import slideshow from .testdata import thumbnail_bytes from .testdata import tree +from .utils.qti.test_validation import VALID_CHOICE_ITEM from .utils.restricted_filesystemstorage import RestrictedFileSystemStorage from contentcuration import models as cc from contentcuration.models import CustomTaskMetadata @@ -258,6 +259,75 @@ def setUp(self): ai.contentnode = qti_exercise ai.save() + # Native QTI item, no perseus_question present -> must route to QTI packaging + native_qti_exercise = create_node( + { + "kind_id": "exercise", + "title": "Native QTI Exercise", + "extra_fields": qti_extra_fields, + } + ) + native_qti_exercise.complete = True + native_qti_exercise.parent = current_exercise.parent + native_qti_exercise.save() + cc.AssessmentItem.objects.create( + contentnode=native_qti_exercise, + assessment_id=uuid.uuid4().hex, + type=exercises.QTI, + question="", + answers="[]", + hints="[]", + raw_data=VALID_CHOICE_ITEM, + order=1, + randomize=False, + ) + + # Only legacy structured-field items, no perseus_question -> must now route to QTI (was Perseus) + legacy_no_perseus_exercise = create_node( + { + "kind_id": "exercise", + "title": "Legacy No Perseus Exercise", + "extra_fields": qti_extra_fields, + } + ) + legacy_no_perseus_exercise.complete = True + legacy_no_perseus_exercise.parent = current_exercise.parent + legacy_no_perseus_exercise.save() + cc.AssessmentItem.objects.create( + contentnode=legacy_no_perseus_exercise, + assessment_id=uuid.uuid4().hex, + type=exercises.SINGLE_SELECTION, + question="What is 2+2?", + answers=json.dumps([{"answer": "4", "correct": True, "order": 1}]), + hints=json.dumps([]), + raw_data="{}", + order=1, + randomize=False, + ) + + # A perseus_question item present -> must still route to Perseus packaging + perseus_only_exercise = create_node( + { + "kind_id": "exercise", + "title": "Perseus Only Exercise", + "extra_fields": qti_extra_fields, + } + ) + perseus_only_exercise.complete = True + perseus_only_exercise.parent = current_exercise.parent + perseus_only_exercise.save() + cc.AssessmentItem.objects.create( + contentnode=perseus_only_exercise, + assessment_id=uuid.uuid4().hex, + type=exercises.PERSEUS_QUESTION, + question="", + answers="[]", + hints="[]", + raw_data="{}", + order=1, + randomize=False, + ) + first_topic = self.content_channel.main_tree.get_descendants().first() # Add a publishable topic to ensure it does not inherit but that its children do @@ -709,6 +779,21 @@ def test_qti_exercise_generates_qti_archive(self): "QTI file should be a zip archive", ) + def test_native_qti_item_routes_to_qti_packaging(self): + node = cc.ContentNode.objects.get(title="Native QTI Exercise") + self.assertTrue(node.files.filter(preset_id=format_presets.QTI_ZIP).exists()) + self.assertFalse(node.files.filter(preset_id=format_presets.EXERCISE).exists()) + + def test_legacy_items_without_perseus_question_route_to_qti_packaging(self): + node = cc.ContentNode.objects.get(title="Legacy No Perseus Exercise") + self.assertTrue(node.files.filter(preset_id=format_presets.QTI_ZIP).exists()) + self.assertFalse(node.files.filter(preset_id=format_presets.EXERCISE).exists()) + + def test_perseus_question_item_routes_to_perseus_packaging(self): + node = cc.ContentNode.objects.get(title="Perseus Only Exercise") + self.assertTrue(node.files.filter(preset_id=format_presets.EXERCISE).exists()) + self.assertFalse(node.files.filter(preset_id=format_presets.QTI_ZIP).exists()) + def test_qti_archive_contains_manifest_and_assessment_ids(self): published_qti_exercise = kolibri_models.ContentNode.objects.get( @@ -733,15 +818,17 @@ def test_unit_topic_publishes_with_exercise_zip(self): attached assessment items compiled into a zip file during publishing.""" unit_topic = cc.ContentNode.objects.get(title="Test Unit Topic") - # Assert UNIT topic has exercise file in Studio + # Assert UNIT topic has a QTI archive file in Studio. Its assessment + # items are legacy structured-field (SINGLE_SELECTION) with no + # perseus_question item, so they route to QTI packaging. unit_files = cc.File.objects.filter( contentnode=unit_topic, - preset_id=format_presets.EXERCISE, + preset_id=format_presets.QTI_ZIP, ) self.assertEqual( unit_files.count(), 1, - "UNIT topic should have exactly one exercise archive file", + "UNIT topic should have exactly one QTI archive file", ) # Assert NO assessment metadata in Kolibri export for UNIT topics diff --git a/contentcuration/contentcuration/utils/publish.py b/contentcuration/contentcuration/utils/publish.py index 6e18dea57e..c3db2ad4d9 100644 --- a/contentcuration/contentcuration/utils/publish.py +++ b/contentcuration/contentcuration/utils/publish.py @@ -363,14 +363,14 @@ def recurse_nodes(self, node, inherited_fields): # noqa C901 if has_assessments(node): exercise_data = process_assessment_metadata(node) - any_free_response = any( - t == exercises.FREE_RESPONSE + any_perseus_question = any( + t == exercises.PERSEUS_QUESTION for t in exercise_data["assessment_mapping"].values() ) generator_class = ( - QTIExerciseGenerator - if any_free_response - else PerseusExerciseGenerator + PerseusExerciseGenerator + if any_perseus_question + else QTIExerciseGenerator ) # If this exercise previously had a file generated by a different From bc6d5f02c802833753cff44d526b875af9ff7dc0 Mon Sep 17 00:00:00 2001 From: rtibblesbot Date: Fri, 3 Jul 2026 13:44:55 -0700 Subject: [PATCH 3/4] fix: don't abort channel publish on native QTI item/media failures A ValueError from QTIExerciseGenerator wasn't caught anywhere in publish.py's recurse_nodes/publish_channel, so any single failure aborted the entire channel publish rather than just the offending item. - Schema-invalid native items are now logged and excluded from the package instead of raising. - Media references with no linked File are now logged and dropped from the manifest instead of raising. Co-Authored-By: Claude Sonnet 5 --- .../tests/utils/test_exercise_creation.py | 26 +++++++++++++---- .../contentcuration/utils/assessment/base.py | 9 ++++-- .../utils/assessment/qti/archive.py | 28 +++++++++++-------- 3 files changed, 43 insertions(+), 20 deletions(-) diff --git a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py index 2c92d47616..32fb779321 100644 --- a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py +++ b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py @@ -2197,7 +2197,8 @@ def test_native_qti_item_media_included_and_addressed(self): manifest = zip_file.read("imsmanifest.xml").decode("utf-8") self.assertIn(media_filename, manifest) - def test_native_qti_item_invalid_raw_data_raises(self): + def test_native_qti_item_invalid_raw_data_is_skipped(self): + """An item that fails schema validation is logged and excluded, not fatal to publish.""" invalid_raw_data = VALID_CHOICE_ITEM.replace( 'orientation="vertical"', 'orientation="sideways"' ) @@ -2210,13 +2211,19 @@ def test_native_qti_item_invalid_raw_data_raises(self): "all_assessment_items": [item.assessment_id], "assessment_mapping": {item.assessment_id: exercises.QTI}, } - with self.assertRaises(ValueError): + with self.assertLogs(level="ERROR") as logs: self._create_qti_zip(exercise_data) - self.assertFalse( - self.exercise_node.files.filter(preset_id=format_presets.QTI_ZIP).exists() + self.assertTrue( + any("failed schema validation" in message for message in logs.output) + ) + exercise_file = self.exercise_node.files.get(preset_id=format_presets.QTI_ZIP) + zip_file = self._validate_qti_zip_structure(exercise_file) + self.assertEqual( + [name for name in zip_file.namelist() if name.startswith("items/")], [] ) - def test_native_qti_item_missing_media_file_raises(self): + def test_native_qti_item_missing_media_file_is_logged_and_omitted(self): + """A dangling media reference is logged and skipped, not fatal to publish.""" raw_data = self.NATIVE_ITEM_XML.format(checksum="b" * 32, ext="png") item = self._create_native_qti_item(raw_data) # no File row linked exercise_data = { @@ -2227,8 +2234,15 @@ def test_native_qti_item_missing_media_file_raises(self): "all_assessment_items": [item.assessment_id], "assessment_mapping": {item.assessment_id: exercises.QTI}, } - with self.assertRaises(ValueError): + with self.assertLogs(level="ERROR") as logs: self._create_qti_zip(exercise_data) + self.assertTrue( + any("no matching File record linked" in message for message in logs.output) + ) + exercise_file = self.exercise_node.files.get(preset_id=format_presets.QTI_ZIP) + zip_file = self._validate_qti_zip_structure(exercise_file) + self.assertIn("items/native_item_1.xml", zip_file.namelist()) + self.assertNotIn(f"items/{'b' * 32}.png", zip_file.namelist()) def test_native_qti_duplicate_identifier_raises(self): item1 = self._create_native_qti_item(VALID_CHOICE_ITEM) diff --git a/contentcuration/contentcuration/utils/assessment/base.py b/contentcuration/contentcuration/utils/assessment/base.py index 368d9ad1c4..ead266e6ba 100644 --- a/contentcuration/contentcuration/utils/assessment/base.py +++ b/contentcuration/contentcuration/utils/assessment/base.py @@ -84,8 +84,8 @@ def create_assessment_item(self, assessment_item, processed_data): assessment_item: The assessment item to process processed_data: Data processed from the assessment item Returns: - filepath: Path for the created assessment item file - file_content: Content of the assessment item file + A (filepath, file_content) tuple, or None if the item should be + excluded from the archive (e.g. it failed validation). """ pass @@ -327,7 +327,10 @@ def process_assessment_item(self, assessment_item): "hints": processed_hints, "randomize": assessment_item.randomize, } - filepath, file_content = self.create_assessment_item(assessment_item, context) + result = self.create_assessment_item(assessment_item, context) + if result is None: + return + filepath, file_content = result self.add_file_to_write(filepath, file_content) def handle_before_assessment_items(self): diff --git a/contentcuration/contentcuration/utils/assessment/qti/archive.py b/contentcuration/contentcuration/utils/assessment/qti/archive.py index 18aa6c151e..e3774b2033 100644 --- a/contentcuration/contentcuration/utils/assessment/qti/archive.py +++ b/contentcuration/contentcuration/utils/assessment/qti/archive.py @@ -1,8 +1,10 @@ import base64 +import logging from dataclasses import dataclass from typing import Any from typing import Dict from typing import List +from typing import Optional from typing import Tuple from le_utils.constants import exercises @@ -197,7 +199,7 @@ def _add_resource(self, resource: QTIResource) -> None: ) self.qti_resources.append(resource) - def _create_native_qti_item(self, assessment_item) -> tuple[str, bytes]: + def _create_native_qti_item(self, assessment_item) -> Optional[Tuple[str, bytes]]: raw_bytes = assessment_item.raw_data.encode("utf-8") result = validate_qti_item(raw_bytes) @@ -205,10 +207,12 @@ def _create_native_qti_item(self, assessment_item) -> tuple[str, bytes]: error_messages = "; ".join( f"line {e.line}, column {e.column}: {e.message}" for e in result.errors ) - raise ValueError( + logging.error( f"QTI item {assessment_item.assessment_id} on node {self.ccnode.pk} " - f"failed schema validation: {error_messages}" + f"failed schema validation and will be excluded from the package: " + f"{error_messages}" ) + return None identifier = parse_qti_xml(raw_bytes).getroot().get("identifier") if not identifier: @@ -238,23 +242,25 @@ def _write_qti_media_files(self, assessment_item) -> List[str]: files_by_name = { f"{f.checksum}.{f.file_format_id}": f for f in assessment_item.files.all() } - missing = sorted(filenames - files_by_name.keys()) + missing = filenames - files_by_name.keys() if missing: - raise ValueError( - f"QTI item {assessment_item.assessment_id} references media files " - f"not linked to it: {', '.join(missing)}" + logging.error( + f"QTI item {assessment_item.assessment_id} on node {self.ccnode.pk} " + f"references media files with no matching File record linked to " + f"the item, so they will be omitted from the package: " + f"{', '.join(sorted(missing))}" ) - sorted_filenames = sorted(filenames) - for filename in sorted_filenames: + present_filenames = sorted(filenames - missing) + for filename in present_filenames: file_obj = files_by_name[filename] self._add_original_image(file_obj.checksum, filename, "items") - return sorted_filenames + return present_filenames def create_assessment_item( self, assessment_item, processed_data: Dict[str, Any] - ) -> tuple[str, bytes]: + ) -> Optional[Tuple[str, bytes]]: """Create QTI assessment item XML.""" if assessment_item.type == exercises.QTI: From b7d140669a4221310cddfb1ce70ca65f289375a0 Mon Sep 17 00:00:00 2001 From: rtibblesbot Date: Fri, 3 Jul 2026 16:18:37 -0700 Subject: [PATCH 4/4] fix: relocate native QTI media to items/images/ and remap XML references Media files referenced by native QTI items were being written bare into items/ alongside the item XML, unlike the legacy generator's items/images/ layout. Write them to items/images/ instead, and rewrite the item XML's src/href/data/srcset values to the images/ prefix that resolves to that directory. The rewrite is a targeted text substitution (not a parse/ serialize round-trip) so it changes only the remapped path values - everything else in the authored raw_data is preserved byte-for-byte. Co-Authored-By: Claude Sonnet 5 --- .../tests/utils/qti/test_media.py | 50 +++++++++++++++++++ .../tests/utils/test_exercise_creation.py | 12 ++++- .../utils/assessment/qti/archive.py | 28 +++++++---- .../utils/assessment/qti/media.py | 33 ++++++++++++ 4 files changed, 112 insertions(+), 11 deletions(-) diff --git a/contentcuration/contentcuration/tests/utils/qti/test_media.py b/contentcuration/contentcuration/tests/utils/qti/test_media.py index 1e204b43c6..430c1db9f5 100644 --- a/contentcuration/contentcuration/tests/utils/qti/test_media.py +++ b/contentcuration/contentcuration/tests/utils/qti/test_media.py @@ -1,4 +1,5 @@ from contentcuration.utils.assessment.qti.media import get_qti_media_references +from contentcuration.utils.assessment.qti.media import rewrite_qti_media_paths CHECKSUM_A = "a" * 32 CHECKSUM_B = "b" * 32 @@ -30,3 +31,52 @@ def test_returns_empty_set_for_malformed_xml(): def test_accepts_bytes(): xml = f''.encode("utf-8") assert get_qti_media_references(xml) == {f"{CHECKSUM_A}.png"} + + +def test_rewrite_leaves_input_untouched_with_no_mapping(): + xml = f'' + assert rewrite_qti_media_paths(xml, {}) == xml + + +def test_rewrite_remaps_src_href_data_and_preserves_formatting(): + xml = ( + f'diagram' + f'x' + f'' + ) + result = rewrite_qti_media_paths( + xml, + { + f"{CHECKSUM_A}.png": f"images/{CHECKSUM_A}.png", + f"{CHECKSUM_B}.pdf": f"images/{CHECKSUM_B}.pdf", + }, + ) + assert result == ( + f'diagram' + f'x' + f'' + ) + + +def test_rewrite_remaps_srcset_entries_preserving_descriptors(): + xml = f'' + result = rewrite_qti_media_paths( + xml, + { + f"{CHECKSUM_A}.png": f"images/{CHECKSUM_A}.png", + f"{CHECKSUM_B}.png": f"images/{CHECKSUM_B}.png", + }, + ) + assert result == ( + f'' + ) + + +def test_rewrite_ignores_values_not_in_mapping(): + xml = f'x' + result = rewrite_qti_media_paths( + xml, {f"{CHECKSUM_A}.png": f"images/{CHECKSUM_A}.png"} + ) + assert result == ( + f'x' + ) diff --git a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py index 32fb779321..868dbeef64 100644 --- a/contentcuration/contentcuration/tests/utils/test_exercise_creation.py +++ b/contentcuration/contentcuration/tests/utils/test_exercise_creation.py @@ -2193,9 +2193,17 @@ def test_native_qti_item_media_included_and_addressed(self): exercise_file = self.exercise_node.files.get(preset_id=format_presets.QTI_ZIP) zip_file = self._validate_qti_zip_structure(exercise_file) media_filename = f"{image_file.checksum}.{image_file.file_format_id}" - self.assertIn(f"items/{media_filename}", zip_file.namelist()) + # Media files can't sit bare alongside the item XML in items/ - they go in + # items/images/, matching the legacy generator's layout. + self.assertIn(f"items/images/{media_filename}", zip_file.namelist()) + self.assertNotIn(f"items/{media_filename}", zip_file.namelist()) manifest = zip_file.read("imsmanifest.xml").decode("utf-8") - self.assertIn(media_filename, manifest) + self.assertIn(f"images/{media_filename}", manifest) + + item_xml = zip_file.read("items/native_item_1.xml").decode("utf-8") + self.assertEqual( + item_xml, raw_data.replace(media_filename, f"images/{media_filename}") + ) def test_native_qti_item_invalid_raw_data_is_skipped(self): """An item that fails schema validation is logged and excluded, not fatal to publish.""" diff --git a/contentcuration/contentcuration/utils/assessment/qti/archive.py b/contentcuration/contentcuration/utils/assessment/qti/archive.py index e3774b2033..64cf094bc5 100644 --- a/contentcuration/contentcuration/utils/assessment/qti/archive.py +++ b/contentcuration/contentcuration/utils/assessment/qti/archive.py @@ -41,6 +41,7 @@ TextEntryInteraction, ) from contentcuration.utils.assessment.qti.media import get_qti_media_references +from contentcuration.utils.assessment.qti.media import rewrite_qti_media_paths from contentcuration.utils.assessment.qti.prompt import Prompt from contentcuration.utils.assessment.qti.validation import parse_qti_xml from contentcuration.utils.assessment.qti.validation import validate_qti_item @@ -221,7 +222,7 @@ def _create_native_qti_item(self, assessment_item) -> Optional[Tuple[str, bytes] ) filepath = f"items/{identifier}.xml" - file_dependencies = self._write_qti_media_files(assessment_item) + item_xml, file_dependencies = self._write_qti_media_files(assessment_item) self._add_resource( QTIResource( @@ -231,12 +232,13 @@ def _create_native_qti_item(self, assessment_item) -> Optional[Tuple[str, bytes] ) ) - return filepath, raw_bytes + return filepath, item_xml.encode("utf-8") - def _write_qti_media_files(self, assessment_item) -> List[str]: - filenames = get_qti_media_references(assessment_item.raw_data) + def _write_qti_media_files(self, assessment_item) -> Tuple[str, List[str]]: + raw_data = assessment_item.raw_data + filenames = get_qti_media_references(raw_data) if not filenames: - return [] + return raw_data, [] # assessment_item.files is prefetched by ExerciseArchiveGenerator.create_exercise_archive files_by_name = { @@ -251,12 +253,20 @@ def _write_qti_media_files(self, assessment_item) -> List[str]: f"{', '.join(sorted(missing))}" ) - present_filenames = sorted(filenames - missing) - for filename in present_filenames: + # Media files can't stay bare in items/ alongside the item XML - they're + # written to items/images/ (matching the legacy generator's layout) and + # the item XML's references are remapped to the images/ prefix that + # resolves to that directory relative to the item file. + path_by_filename = {} + for filename in sorted(filenames - missing): file_obj = files_by_name[filename] - self._add_original_image(file_obj.checksum, filename, "items") + self._add_original_image( + file_obj.checksum, filename, self.get_image_file_path() + ) + path_by_filename[filename] = f"{self.get_image_ref_prefix()}/{filename}" - return present_filenames + item_xml = rewrite_qti_media_paths(raw_data, path_by_filename) + return item_xml, sorted(path_by_filename.values()) def create_assessment_item( self, assessment_item, processed_data: Dict[str, Any] diff --git a/contentcuration/contentcuration/utils/assessment/qti/media.py b/contentcuration/contentcuration/utils/assessment/qti/media.py index 0522d7fe59..660bd10a6d 100644 --- a/contentcuration/contentcuration/utils/assessment/qti/media.py +++ b/contentcuration/contentcuration/utils/assessment/qti/media.py @@ -14,6 +14,10 @@ + " or ".join(f"@{attribute}" for attribute in QTI_REFERENCE_ATTRIBUTES) + " or @srcset]" ) +QTI_MEDIA_ATTRIBUTE_VALUE_REGEX = re.compile( + r"(?P" + "|".join(QTI_REFERENCE_ATTRIBUTES + ("srcset",)) + r")" + r'(?P\s*=\s*)(?P["\'])(?P[^"\']*)(?P=quote)' +) def get_qti_media_references(raw_data): @@ -43,3 +47,32 @@ def get_qti_media_references(raw_data): if value and QTI_CHECKSUM_FILENAME_REGEX.match(value) ) return checksums + + +def rewrite_qti_media_paths(raw_data, path_by_filename): + """ + Rewrite src/href/data/srcset attribute values referencing keys of + `path_by_filename` to the corresponding new path. Operates as a targeted + text substitution rather than a parse/serialize round-trip, so every other + byte of `raw_data` (formatting, attribute order, self-closing tag style, + etc.) is left untouched. + """ + if not path_by_filename: + return raw_data + + def _replace_srcset_entry(match): + filename = match.group(1) + new_path = path_by_filename.get(filename) + if new_path is None: + return match.group(0) + return match.group(0).replace(filename, new_path, 1) + + def _replace_attribute(match): + attribute, eq, quote, value = match.group("attr", "eq", "quote", "value") + if attribute == "srcset": + value = re.sub(srcset_entry_pattern, _replace_srcset_entry, value) + elif value in path_by_filename: + value = path_by_filename[value] + return f"{attribute}{eq}{quote}{value}{quote}" + + return QTI_MEDIA_ATTRIBUTE_VALUE_REGEX.sub(_replace_attribute, raw_data)