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/tests/utils/qti/test_media.py b/contentcuration/contentcuration/tests/utils/qti/test_media.py new file mode 100644 index 0000000000..430c1db9f5 --- /dev/null +++ b/contentcuration/contentcuration/tests/utils/qti/test_media.py @@ -0,0 +1,82 @@ +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 + + +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"} + + +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 ae6a295e41..868dbeef64 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,163 @@ 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}" + # 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(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.""" + 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.assertLogs(level="ERROR") as logs: + self._create_qti_zip(exercise_data) + 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_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 = { + "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.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) + 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/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 7574192a61..64cf094bc5 100644 --- a/contentcuration/contentcuration/utils/assessment/qti/archive.py +++ b/contentcuration/contentcuration/utils/assessment/qti/archive.py @@ -1,7 +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 @@ -37,7 +40,11 @@ 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.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 choice_interactions = { @@ -56,6 +63,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 +81,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 +193,89 @@ 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) -> Optional[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 + ) + logging.error( + f"QTI item {assessment_item.assessment_id} on node {self.ccnode.pk} " + 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: + raise ValueError( + f"QTI item {assessment_item.assessment_id} is missing a root identifier attribute" + ) + + filepath = f"items/{identifier}.xml" + item_xml, file_dependencies = self._write_qti_media_files(assessment_item) + + self._add_resource( + QTIResource( + identifier=identifier, + filepath=filepath, + file_dependencies=file_dependencies, + ) + ) + + return filepath, item_xml.encode("utf-8") + + 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 raw_data, [] + + # 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 = filenames - files_by_name.keys() + if 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))}" + ) + + # 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, self.get_image_file_path() + ) + path_by_filename[filename] = f"{self.get_image_ref_prefix()}/{filename}" + + 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] - ) -> tuple[str, bytes]: + ) -> Optional[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 +319,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 +329,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 +336,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..660bd10a6d --- /dev/null +++ b/contentcuration/contentcuration/utils/assessment/qti/media.py @@ -0,0 +1,78 @@ +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]" +) +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): + """ + 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 + + +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) 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 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):