From 13ff6ed3911106ac25aa2caf8ed1a0dfbfb9efab Mon Sep 17 00:00:00 2001 From: Richard Tibbles Date: Mon, 9 Feb 2026 06:25:24 -0800 Subject: [PATCH 1/5] Add SCORM/LOM metadata mapping utilities Add utilities to map SCORM metadata fields (from IMS Content Packages) to Kolibri content node fields: title, description, language, tags, learning_activities, resource_types, and learner_needs. Co-Authored-By: Claude Opus 4.6 --- ricecooker/utils/SCORM_metadata.py | 341 +++++++++++++++++++++++++++++ tests/test_SCORM_metadata.py | 199 +++++++++++++++++ 2 files changed, 540 insertions(+) create mode 100644 ricecooker/utils/SCORM_metadata.py create mode 100644 tests/test_SCORM_metadata.py diff --git a/ricecooker/utils/SCORM_metadata.py b/ricecooker/utils/SCORM_metadata.py new file mode 100644 index 00000000..98e62998 --- /dev/null +++ b/ricecooker/utils/SCORM_metadata.py @@ -0,0 +1,341 @@ +""" +Utilities for mapping from SCORM metadata to LE Utils metadata. +""" +import re + +from le_utils.constants import licenses +from le_utils.constants.labels import learning_activities +from le_utils.constants.labels import needs +from le_utils.constants.labels import resource_type +from le_utils.constants.languages import getlang + + +imscp_metadata_keys = { + "general": ["title", "description", "language", "keyword"], + "rights": ["cost", "copyrightAndOtherRestrictions", "description"], + "educational": [ + "interactivityType", + "interactivityLevel", + "learningResourceType", + "intendedEndUserRole", + ], + "lifeCycle": ["contribute"], +} + + +# Define mappings from SCORM educational types to LE Utils activity types +SCORM_to_learning_activities_mappings = { + "exercise": learning_activities.PRACTICE, + "simulation": learning_activities.EXPLORE, + "questionnaire": learning_activities.PRACTICE, + "diagram": learning_activities.EXPLORE, + "figure": learning_activities.EXPLORE, + "graph": learning_activities.EXPLORE, + "index": learning_activities.READ, + "slide": learning_activities.READ, + "table": learning_activities.READ, + "narrative text": learning_activities.READ, + "exam": learning_activities.PRACTICE, + "experiment": learning_activities.EXPLORE, + "problem statement": learning_activities.REFLECT, + "self assessment": learning_activities.REFLECT, + "lecture": learning_activities.WATCH, +} + + +def _ensure_list(value): + """Normalize a value to a list: None -> [], str -> [str], other -> list.""" + if value is None: + return [] + if isinstance(value, str): + return [value] + return list(value) + + +def map_scorm_to_le_utils_activities(metadata_dict): + le_utils_activities = [] + + # Adjustments based on interactivity + interactive_adjustments = { + learning_activities.EXPLORE: ( + learning_activities.READ, + learning_activities.WATCH, + ) + } + + # Determine the interactivity level and type + interactivity_type = metadata_dict.get("interactivityType") + interactivity_level = metadata_dict.get("interactivityLevel") + + is_interactive = ( + interactivity_type + in [ + "active", + "mixed", + ] + or interactivity_level in ["medium", "high"] + ) + + # Extract the learning resource types from the SCORM data + learning_resource_types = _ensure_list(metadata_dict.get("learningResourceType")) + + # Map each SCORM type to an LE Utils activity type + for learning_resource_type in learning_resource_types: + le_utils_type = SCORM_to_learning_activities_mappings.get( + learning_resource_type + ) + # Adjust based on interactivity + if not is_interactive and le_utils_type in interactive_adjustments: + le_utils_type = ( + interactive_adjustments[le_utils_type][0] + if learning_resource_type == "simulation" + else interactive_adjustments[le_utils_type][1] + ) + + if le_utils_type and le_utils_type not in le_utils_activities: + le_utils_activities.append(le_utils_type) + + return le_utils_activities + + +# Define mappings from SCORM educational types to educator-focused resource types +SCORM_to_resource_type_mappings = { + "exercise": resource_type.EXERCISE, + "simulation": resource_type.ACTIVITY, + "questionnaire": resource_type.ACTIVITY, + "diagram": resource_type.MEDIA, + "figure": resource_type.MEDIA, + "graph": resource_type.MEDIA, + "index": resource_type.GUIDE, + "slide": resource_type.LESSON, + "table": resource_type.TUTORIAL, + "narrative text": resource_type.TEXTBOOK, + "exam": resource_type.EXERCISE, + "experiment": resource_type.ACTIVITY, + "problem statement": resource_type.ACTIVITY, + "self assessment": resource_type.ACTIVITY, + "lecture": resource_type.LESSON, +} + + +# Mapping for intendedEndUserRole when the resource is for educators +SCORM_intended_role_to_resource_type_mapping = { + "teacher": resource_type.LESSON_PLAN, + "author": resource_type.GUIDE, + "manager": resource_type.GUIDE, +} + + +def map_scorm_to_educator_resource_types(metadata_dict): + educator_resource_types = [] + + # Extract the learning resource types and intended end user role from the SCORM data + learning_resource_types = _ensure_list(metadata_dict.get("learningResourceType")) + intended_roles = _ensure_list(metadata_dict.get("intendedEndUserRole")) + + # Map each SCORM type to an educator-focused resource type + for learning_resource_type in learning_resource_types: + mapped_type = SCORM_to_resource_type_mappings.get(learning_resource_type) + if mapped_type and mapped_type not in educator_resource_types: + educator_resource_types.append(mapped_type) + + # Check if the intended end user role matches any educator roles + for role in intended_roles: + if ( + role in SCORM_intended_role_to_resource_type_mapping + and SCORM_intended_role_to_resource_type_mapping[role] + not in educator_resource_types + ): + educator_resource_types.append( + SCORM_intended_role_to_resource_type_mapping[role] + ) + + return educator_resource_types + + +def infer_beginner_level_from_difficulty(metadata_dict): + # Beginner difficulty levels + beginner_difficulties = {"very easy", "easy"} + + # Check if the difficulty level indicates beginner content + difficulty = metadata_dict.get("difficulty") + if difficulty in beginner_difficulties: + return [needs.FOR_BEGINNERS] + + return [] + + +def parse_vcard_fn(vcard_text): + """Extract FN (Full Name) from a VCARD string.""" + if not vcard_text: + return None + match = re.search(r"^FN:(.+)$", vcard_text, re.MULTILINE) + return match.group(1).strip() if match else None + + +def parse_vcard_org(vcard_text): + """Extract ORG (Organization) from a VCARD string.""" + if not vcard_text: + return None + match = re.search(r"^ORG:(.+)$", vcard_text, re.MULTILINE) + return match.group(1).strip() if match else None + + +# CC license patterns, ordered most specific first to avoid partial matches +_CC_LICENSE_PATTERNS = [ + ("Attribution-NonCommercial-ShareAlike", licenses.CC_BY_NC_SA), + ("Attribution-NonCommercial-NoDerivs", licenses.CC_BY_NC_ND), + ("Attribution-ShareAlike", licenses.CC_BY_SA), + ("Attribution-NoDerivs", licenses.CC_BY_ND), + ("Attribution-NonCommercial", licenses.CC_BY_NC), + ("Attribution", licenses.CC_BY), +] + + +def infer_license_from_rights(metadata_dict): + """Infer a license ID from rights metadata fields. + + Args: + metadata_dict: dict with rights_description, copyrightAndOtherRestrictions, etc. + + Returns: + (license_id, license_description) tuple. Either or both may be None. + """ + description = metadata_dict.get("rights_description") + copyright_restrictions = metadata_dict.get("copyrightAndOtherRestrictions") + + if description: + for pattern, license_id in _CC_LICENSE_PATTERNS: + if pattern in description: + return license_id, description + + if copyright_restrictions == "no": + return licenses.PUBLIC_DOMAIN, description + + return None, description + + +def _resolve_entity_name(entity, prefer_org=False): + """Extract a name from a VCARD entity string. + + If prefer_org is True, tries ORG first, then falls back to FN. + Otherwise returns FN directly. + """ + if prefer_org: + return parse_vcard_org(entity) or parse_vcard_fn(entity) + return parse_vcard_fn(entity) + + +# Maps contribute role -> (result field name, whether to prefer ORG over FN) +_ROLE_TO_FIELD = { + "author": ("author", False), + "publisher": ("provider", True), + "content provider": ("copyright_holder", True), +} + + +def extract_lifecycle_contributors(metadata_dict): + """Extract author, provider, copyright_holder from lifeCycle contribute data. + + Args: + metadata_dict: dict that may contain a 'contribute' key with VCARD entity data + + Returns: + dict with any of: author, provider, copyright_holder + """ + result = {} + contribute = metadata_dict.get("contribute") + if not contribute: + return result + + if isinstance(contribute, dict): + contribute = [contribute] + + for entry in contribute: + role_value = entry.get("role", {}) + if isinstance(role_value, dict): + role_value = role_value.get("value", "") + entity = entry.get("entity", "") + + field_config = _ROLE_TO_FIELD.get(role_value) + if not field_config: + continue + field_name, prefer_org = field_config + name = _resolve_entity_name(entity, prefer_org) + if name: + result[field_name] = name + + return result + + +def _normalize_language(lang_code): + """Normalize a language code, returning None if unrecognized.""" + if not lang_code: + return None + if getlang(lang_code) is None: + lang_code = lang_code.split("-")[0].lower() + return lang_code if getlang(lang_code) else None + + +def _normalize_keywords(keyword): + """Normalize keyword field to a list, or None if empty.""" + if not keyword: + return None + if isinstance(keyword, str): + return [keyword] + return keyword + + +def metadata_dict_to_content_node_fields(metadata_dict): + """Convert a SCORM/IMSCP metadata dict to fields suitable for ContentNodeMetadata. + + Maps: + title -> title + description -> description + language -> language (normalized: e.g. en-US -> en) + keyword -> tags (as list) + educational fields -> learning_activities, resource_types, learner_needs + rights fields -> license, license_description + lifeCycle contribute -> author, provider, copyright_holder + + Returns: + dict of fields (only includes non-empty values) + """ + result = {} + + if metadata_dict.get("title"): + result["title"] = metadata_dict["title"] + + if metadata_dict.get("description"): + result["description"] = metadata_dict["description"] + + language = _normalize_language(metadata_dict.get("language", "")) + if language: + result["language"] = language + + tags = _normalize_keywords(metadata_dict.get("keyword", [])) + if tags: + result["tags"] = tags + + activities = map_scorm_to_le_utils_activities(metadata_dict) + if activities: + result["learning_activities"] = activities + + resource_types = map_scorm_to_educator_resource_types(metadata_dict) + if resource_types: + result["resource_types"] = resource_types + + learner_needs = infer_beginner_level_from_difficulty(metadata_dict) + if learner_needs: + result["learner_needs"] = learner_needs + + license_id, license_description = infer_license_from_rights(metadata_dict) + if license_id: + result["license"] = license_id + if license_description: + result["license_description"] = license_description + + contributors = extract_lifecycle_contributors(metadata_dict) + result.update(contributors) + + return result diff --git a/tests/test_SCORM_metadata.py b/tests/test_SCORM_metadata.py new file mode 100644 index 00000000..3dd33b98 --- /dev/null +++ b/tests/test_SCORM_metadata.py @@ -0,0 +1,199 @@ +import pytest +from le_utils.constants import licenses +from le_utils.constants.labels import learning_activities +from le_utils.constants.labels import needs +from le_utils.constants.labels import resource_type + +from ricecooker.utils.SCORM_metadata import extract_lifecycle_contributors +from ricecooker.utils.SCORM_metadata import infer_beginner_level_from_difficulty +from ricecooker.utils.SCORM_metadata import infer_license_from_rights +from ricecooker.utils.SCORM_metadata import map_scorm_to_educator_resource_types +from ricecooker.utils.SCORM_metadata import map_scorm_to_le_utils_activities +from ricecooker.utils.SCORM_metadata import parse_vcard_fn +from ricecooker.utils.SCORM_metadata import parse_vcard_org + + +@pytest.mark.parametrize( + "scorm_dict, expected_result", + [ + ( + { + "interactivityType": "active", + "interactivityLevel": "high", + "learningResourceType": ["exercise", "simulation"], + }, + [learning_activities.PRACTICE, learning_activities.EXPLORE], + ), + ( + {"learningResourceType": ["lecture", "self assessment"]}, + [learning_activities.REFLECT, learning_activities.WATCH], + ), + ( + { + "interactivityType": "mixed", + "interactivityLevel": "medium", + "learningResourceType": ["simulation", "graph"], + }, + [learning_activities.EXPLORE], + ), + ( + { + "interactivityType": "expositive", + "interactivityLevel": "low", + "learningResourceType": ["simulation", "graph"], + }, + [learning_activities.READ, learning_activities.WATCH], + ), + ], +) +def test_map_scorm_to_le_utils_activities(scorm_dict, expected_result): + assert set(map_scorm_to_le_utils_activities(scorm_dict)) == set(expected_result) + + +@pytest.mark.parametrize( + "scorm_dict, expected_result", + [ + ( + { + "learningResourceType": ["exercise", "lecture"], + "intendedEndUserRole": ["teacher"], + }, + [resource_type.EXERCISE, resource_type.LESSON, resource_type.LESSON_PLAN], + ), + ( + { + "learningResourceType": ["simulation", "figure"], + "intendedEndUserRole": ["author"], + }, + [resource_type.ACTIVITY, resource_type.MEDIA, resource_type.GUIDE], + ), + ], +) +def test_map_scorm_to_educator_resource_types(scorm_dict, expected_result): + assert set(map_scorm_to_educator_resource_types(scorm_dict)) == set(expected_result) + + +def test_infer_beginner_level_from_difficulty(): + scorm_dict_easy = {"difficulty": "easy"} + assert infer_beginner_level_from_difficulty(scorm_dict_easy) == [ + needs.FOR_BEGINNERS + ] + + scorm_dict_hard = {"difficulty": "difficult"} + assert infer_beginner_level_from_difficulty(scorm_dict_hard) == [] + + +# --- VCARD parsing tests --- + + +def test_parse_vcard_fn(): + vcard = "BEGIN:VCARD\nVERSION:2.1\nFN:John Doe\nORG:Example Org\nEND:VCARD" + assert parse_vcard_fn(vcard) == "John Doe" + + +def test_parse_vcard_fn_none(): + assert parse_vcard_fn(None) is None + assert parse_vcard_fn("BEGIN:VCARD\nVERSION:2.1\nEND:VCARD") is None + + +def test_parse_vcard_org(): + vcard = "BEGIN:VCARD\nVERSION:2.1\nFN:John Doe\nORG:Example Organization\nEND:VCARD" + assert parse_vcard_org(vcard) == "Example Organization" + + +def test_parse_vcard_org_none(): + assert parse_vcard_org(None) is None + assert parse_vcard_org("BEGIN:VCARD\nVERSION:2.1\nFN:John\nEND:VCARD") is None + + +# --- License inference tests --- + + +def test_infer_license_cc_by_nc_sa(): + metadata = { + "rights_description": "Content is protected under Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License.", + "copyrightAndOtherRestrictions": "yes", + } + license_id, license_desc = infer_license_from_rights(metadata) + assert license_id == licenses.CC_BY_NC_SA + + +def test_infer_license_cc_by(): + metadata = { + "rights_description": "Licensed under Creative Commons Attribution 4.0", + "copyrightAndOtherRestrictions": "yes", + } + license_id, license_desc = infer_license_from_rights(metadata) + assert license_id == licenses.CC_BY + + +def test_infer_license_public_domain(): + metadata = { + "copyrightAndOtherRestrictions": "no", + } + license_id, license_desc = infer_license_from_rights(metadata) + assert license_id == licenses.PUBLIC_DOMAIN + + +def test_infer_license_no_match(): + metadata = { + "rights_description": "All rights reserved. Custom license.", + "copyrightAndOtherRestrictions": "yes", + } + license_id, license_desc = infer_license_from_rights(metadata) + assert license_id is None + assert license_desc == "All rights reserved. Custom license." + + +def test_infer_license_empty(): + license_id, license_desc = infer_license_from_rights({}) + assert license_id is None + assert license_desc is None + + +# --- Lifecycle contributor extraction tests --- + + +def test_extract_lifecycle_publisher(): + metadata = { + "contribute": { + "role": {"value": "publisher"}, + "entity": "BEGIN:VCARD\nVERSION:2.1\nFN:John Doe\nORG:Example Organization\nEND:VCARD", + } + } + result = extract_lifecycle_contributors(metadata) + assert result["provider"] == "Example Organization" + + +def test_extract_lifecycle_author(): + metadata = { + "contribute": { + "role": {"value": "author"}, + "entity": "BEGIN:VCARD\nVERSION:2.1\nFN:Jane Smith\nEND:VCARD", + } + } + result = extract_lifecycle_contributors(metadata) + assert result["author"] == "Jane Smith" + + +def test_extract_lifecycle_multiple_contributors(): + metadata = { + "contribute": [ + { + "role": {"value": "author"}, + "entity": "BEGIN:VCARD\nVERSION:2.1\nFN:Jane Smith\nEND:VCARD", + }, + { + "role": {"value": "publisher"}, + "entity": "BEGIN:VCARD\nVERSION:2.1\nFN:John Doe\nORG:Pub Corp\nEND:VCARD", + }, + ] + } + result = extract_lifecycle_contributors(metadata) + assert result["author"] == "Jane Smith" + assert result["provider"] == "Pub Corp" + + +def test_extract_lifecycle_empty(): + result = extract_lifecycle_contributors({}) + assert result == {} From 01bfef2acad3ea5e00cdb2a5b1b69e2ab40bd1d0 Mon Sep 17 00:00:00 2001 From: Richard Tibbles Date: Mon, 9 Feb 2026 06:25:31 -0800 Subject: [PATCH 2/5] Add standalone IMSCP manifest parsing Add ricecooker/utils/imscp.py with functions for parsing imsmanifest.xml from IMSCP zip files. Includes recursive organization tree building, resource linking, external metadata resolution, encoding fallback via chardet, and single-child topic chain flattening to reduce unnecessary nesting levels. Adds xmltodict dependency to setup.py. Co-Authored-By: Claude Opus 4.6 --- ricecooker/utils/imscp.py | 369 ++++++++++++++++ tests/test_imscp.py | 411 ++++++++++++++++++ ...mplete_manifest_with_external_metadata.xml | 103 +++++ .../samples/ims_xml/manifest_dangling_ref.xml | 27 ++ .../ims_xml/metadata_hummingbirds_course.xml | 232 ++++++++++ .../metadata_hummingbirds_organization.xml | 14 + .../samples/ims_xml/simple_manifest.xml | 63 +++ 7 files changed, 1219 insertions(+) create mode 100644 ricecooker/utils/imscp.py create mode 100644 tests/test_imscp.py create mode 100644 tests/testcontent/samples/ims_xml/complete_manifest_with_external_metadata.xml create mode 100644 tests/testcontent/samples/ims_xml/manifest_dangling_ref.xml create mode 100644 tests/testcontent/samples/ims_xml/metadata_hummingbirds_course.xml create mode 100644 tests/testcontent/samples/ims_xml/metadata_hummingbirds_organization.xml create mode 100644 tests/testcontent/samples/ims_xml/simple_manifest.xml diff --git a/ricecooker/utils/imscp.py b/ricecooker/utils/imscp.py new file mode 100644 index 00000000..1a7ffc8d --- /dev/null +++ b/ricecooker/utils/imscp.py @@ -0,0 +1,369 @@ +""" +Standalone IMSCP manifest parsing utilities. + +Used by the file pipeline to parse imsmanifest.xml from IMSCP zip files. +""" +import io +import logging +import re +import zipfile + +import chardet +from lxml import etree + +from ricecooker.utils.SCORM_metadata import imscp_metadata_keys + +logger = logging.getLogger(__name__) + + +class ManifestParseError(Exception): + """Raised when an imsmanifest.xml cannot be parsed.""" + + +def strip_ns_prefix(tree): + """Strip namespace prefixes from an LXML tree. + From https://stackoverflow.com/a/30233635 + """ + for element in tree.xpath("descendant-or-self::*[namespace-uri()!='']"): + element.tag = etree.QName(element).localname + + +def _get_elem_for_tag(root, tag): + elem = root.find("lom/%s" % tag) + if elem is not None: + return elem + return root.find(tag) + + +def _extract_lom_text(elem, preferred_language): + """Extract text from a LOM element using direct lxml traversal. + + Handles common LOM XML patterns: + - text + - text + - LOMv1.0text + - plain text + """ + strings = elem.findall("string") or elem.findall("langstring") + if strings: + if preferred_language is not None: + for s in strings: + lang = s.get("language", "") or s.get( + "{http://www.w3.org/XML/1998/namespace}lang", "" + ) + if lang.startswith(preferred_language): + return s.text + if len(strings) == 1: + return strings[0].text + return [s.text for s in strings] + + value_elem = elem.find("value") + if value_elem is not None: + return value_elem.text + + return elem.text.strip() if elem.text and elem.text.strip() else None + + +def _extract_contribute(contrib_elem): + """Extract a lifeCycle contribute entry as a dict with role and entity.""" + result = {} + role_elem = contrib_elem.find("role") + if role_elem is not None: + value_elem = role_elem.find("value") + if value_elem is not None: + result["role"] = {"value": value_elem.text} + entity_elem = contrib_elem.find("entity") + if entity_elem is not None and entity_elem.text: + result["entity"] = entity_elem.text + return result + + +def _resolve_metadata_elem(root, zip_file): + """Find and resolve the metadata element, following external references.""" + metadata_elem = root.find("metadata", root.nsmap) + if metadata_elem is None: + return None + + external_ref = metadata_elem.find( + "adlcp:location", + namespaces={"adlcp": "http://www.adlnet.org/xsd/adlcp_v1p3"}, + ) + if external_ref is not None and zip_file is not None: + with zip_file.open(external_ref.text) as external_file: + metadata_elem = etree.parse(external_file).getroot() + + strip_ns_prefix(metadata_elem) + return metadata_elem + + +def _detect_language(metadata_elem): + """Detect the language from the general section of a metadata element.""" + gen_elem = _get_elem_for_tag(metadata_elem, "general") + if gen_elem is not None: + lang_elem = gen_elem.find("language") + if lang_elem is not None and lang_elem.text: + return lang_elem.text.strip() + return None + + +def _collect_field(elem, tag, field, preferred_language): + """Extract a single field from a LOM section element. + + Returns (key, value) or None if the field is not present. + """ + if field == "contribute": + contrib_elems = elem.findall("contribute") + if len(contrib_elems) == 1: + return ("contribute", _extract_contribute(contrib_elems[0])) + elif contrib_elems: + return ("contribute", [_extract_contribute(c) for c in contrib_elems]) + return None + + field_elems = elem.findall(field) + if not field_elems: + return None + # Prefix rights fields to avoid collision with general.description + key = "rights_{}".format(field) if tag == "rights" else field + if len(field_elems) == 1: + return (key, _extract_lom_text(field_elems[0], preferred_language)) + return (key, [_extract_lom_text(fe, preferred_language) for fe in field_elems]) + + +def collect_metadata(root, zip_file=None, language=None): + """Extract LOM metadata from a manifest element. + + Args: + root: lxml Element (manifest or organization root) + zip_file: open ZipFile for resolving external metadata refs + language: preferred language code + + Returns: + dict of metadata fields + """ + metadata_elem = _resolve_metadata_elem(root, zip_file) + if metadata_elem is None: + return {} + + preferred_language = language or _detect_language(metadata_elem) + + metadata_dict = {} + for tag, fields in imscp_metadata_keys.items(): + elem = _get_elem_for_tag(metadata_elem, tag) + if elem is None: + continue + for field in fields: + result = _collect_field(elem, tag, field, preferred_language) + if result is not None: + metadata_dict[result[0]] = result[1] + + return metadata_dict + + +def parse_manifest_from_zip(zf): + """Parse imsmanifest.xml from an already-open ZipFile, with chardet fallback. + + Args: + zf: an open zipfile.ZipFile instance + + Returns: + lxml Element root of the manifest + """ + try: + with zf.open("imsmanifest.xml") as manifest_file: + return etree.parse(manifest_file).getroot() + except (etree.XMLSyntaxError, OSError): + pass + + # Handle XML files that are marked as UTF-8 but have non-UTF-8 chars. + # Detect the real encoding with chardet and re-encode as UTF-8. + try: + f = zf.open("imsmanifest.xml", "r") + data = f.read() + f.close() + + info = chardet.detect(data) + encoding = info["encoding"] or "utf-8" + data = data.decode(encoding) + return etree.parse(io.BytesIO(data.encode("utf-8"))).getroot() + except (etree.XMLSyntaxError, OSError, UnicodeDecodeError) as e: + raise ManifestParseError(str(e)) from e + + +def get_manifest(zip_path): + """Parse imsmanifest.xml from a zip file path, with chardet fallback. + + Args: + zip_path: path to the zip file + + Returns: + lxml Element root of the manifest + """ + with zipfile.ZipFile(zip_path) as zf: + return parse_manifest_from_zip(zf) + + +def walk_items(root, zip_file=None, language=None): + """Recursively walk item elements in a manifest, collecting metadata. + + Args: + root: lxml Element for an organization or item + zip_file: open ZipFile for resolving external metadata refs + language: preferred language code + + Returns: + dict with title, metadata, children, and item attributes + """ + root_dict = dict(root.items()) + + title_elem = root.find("title", root.nsmap) + if title_elem is not None: + text = "" + for child in title_elem.iter(): + if child.text: + text += child.text + if child.tail: + text += child.tail + if not text.strip(): + raise ManifestParseError( + "Title element has no title: {}".format( + etree.tostring(title_elem, pretty_print=True) + ) + ) + root_dict["title"] = text.strip() + + root_dict["metadata"] = collect_metadata(root, zip_file=zip_file, language=language) + + children = [] + for item in root.findall("item", root.nsmap): + children.append(walk_items(item, zip_file=zip_file, language=language)) + + if children: + root_dict["children"] = children + + return root_dict + + +def derive_content_files_dict(resource_elem, resources_dict): + """Collect all file paths referenced by a resource element, including dependencies.""" + nsmap = resource_elem.nsmap + file_elements = resource_elem.findall("file", nsmap) + base = resource_elem.get("{http://www.w3.org/XML/1998/namespace}base") or "" + file_paths = [base + fe.get("href") for fe in file_elements] + dep_elements = resource_elem.findall("dependency", nsmap) + dep_paths = [] + for de in dep_elements: + dep_ref = de.get("identifierref") + dre = resources_dict.get(dep_ref) + if dre is None: + logger.warning("Dangling dependency identifierref: %s", dep_ref) + continue + dep_paths.extend(derive_content_files_dict(dre, resources_dict)) + return file_paths + dep_paths + + +def collect_resources(item, resources_dict): + """Link items to their resource data (href, type, files). + + Modifies item dict in-place. + """ + if item.get("children"): + for child in item["children"]: + collect_resources(child, resources_dict) + elif item.get("identifierref"): + resource_elem = resources_dict.get(item["identifierref"]) + if resource_elem is None: + logger.warning("Dangling identifierref: %s", item["identifierref"]) + return + + for key, value in resource_elem.items(): + key_stripped = re.sub("^{.*}", "", key) + if key_stripped not in item: + item[key_stripped] = value + + if resource_elem.get("type") == "webcontent": + item["files"] = derive_content_files_dict(resource_elem, resources_dict) + + +def flatten_single_child_topics(item): + """Collapse single-child topic chains. + + When a topic has exactly one child that is also a topic (has children), + merge them by replacing the parent with the child, keeping the parent's + title only if the child doesn't have one. + + Returns the (possibly modified) item. + """ + if "children" not in item: + return item + + # First flatten all children recursively + item["children"] = [ + flatten_single_child_topics(child) for child in item["children"] + ] + + # Then check if this item has exactly one child that is also a topic + if len(item["children"]) == 1: + only_child = item["children"][0] + if "children" in only_child: + # Merge: keep child's structure but preserve parent's metadata if richer + if not only_child.get("metadata"): + only_child["metadata"] = item.get("metadata", {}) + return only_child + + return item + + +def parse_imscp_manifest(zip_path, language=None): + """Parse an IMSCP manifest from a zip file. + + Top-level entry point that opens the zip, parses the manifest XML, + collects metadata and organization tree, links resources, + and optionally flattens single-child topic chains. + + Args: + zip_path: path to the IMSCP zip file + language: preferred language code for metadata extraction + + Returns: + dict with keys: + - identifier: manifest identifier string + - metadata: dict of LOM metadata fields + - organizations: list of organization dicts with nested items + """ + manifest = get_manifest(zip_path) + nsmap = manifest.nsmap + + with zipfile.ZipFile(zip_path) as zf: + metadata = collect_metadata(manifest, zip_file=zf, language=language) + + resources_elem = manifest.find("resources", nsmap) + resources_dict = dict((r.get("identifier"), r) for r in resources_elem) + + organizations = [] + for org_elem in manifest.findall("organizations/organization", nsmap): + item_tree = walk_items(org_elem, zip_file=zf, language=language) + collect_resources(item_tree, resources_dict) + item_tree = flatten_single_child_topics(item_tree) + organizations.append(item_tree) + + return { + "identifier": manifest.get("identifier"), + "metadata": metadata, + "organizations": organizations, + } + + +def has_imscp_manifest(zip_path): + """Check whether a zip file contains imsmanifest.xml. + + Args: + zip_path: path to the zip file + + Returns: + bool + """ + try: + with zipfile.ZipFile(zip_path) as zf: + return "imsmanifest.xml" in zf.namelist() + except (zipfile.BadZipFile, FileNotFoundError): + return False diff --git a/tests/test_imscp.py b/tests/test_imscp.py new file mode 100644 index 00000000..b1a596e8 --- /dev/null +++ b/tests/test_imscp.py @@ -0,0 +1,411 @@ +"""Tests for IMSCP manifest parsing (ricecooker.utils.imscp).""" +import os +import tempfile +import zipfile +from contextlib import contextmanager + +import pytest + +from ricecooker.utils.imscp import flatten_single_child_topics +from ricecooker.utils.imscp import has_imscp_manifest +from ricecooker.utils.imscp import parse_imscp_manifest +from ricecooker.utils.SCORM_metadata import metadata_dict_to_content_node_fields + + +BASE_PATH = os.path.dirname(os.path.abspath(__file__)) +IMS_XML_DIR = os.path.join(BASE_PATH, "testcontent", "samples", "ims_xml") + + +@contextmanager +def create_zip_with_manifest(manifest_filename, *additional_files): + """Create a temp zip with a manifest and optional additional files.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + manifest_file = os.path.join(IMS_XML_DIR, manifest_filename) + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.write(manifest_file, "imsmanifest.xml") + for additional_file in additional_files: + zf.write(os.path.join(IMS_XML_DIR, additional_file), additional_file) + yield temp_zip_path + finally: + try: + os.remove(temp_zip_path) + except (FileNotFoundError, OSError): + pass + + +# --- Manifest parsing tests --- + + +def test_parse_simple_manifest(): + """Simple IMSCP zip produces correct identifier, metadata, organizations.""" + with create_zip_with_manifest("simple_manifest.xml") as zip_path: + result = parse_imscp_manifest(zip_path) + + assert result["identifier"] is None # simple_manifest.xml has no identifier attr + assert result["metadata"]["title"] == "Test File" + assert result["metadata"]["description"] == "Example of test file" + assert result["metadata"]["language"] == "en" + + # Should have 2 organizations + assert len(result["organizations"]) == 2 + + # First organization: "Folder 1" with children + org1 = result["organizations"][0] + assert org1["title"] == "Folder 1" + assert "children" in org1 + + # Check leaf items have href and files from resource linking + leaves = [c for c in org1["children"] if "children" not in c] + assert len(leaves) >= 2 + assert leaves[0]["title"] == "Test File1" + assert leaves[0]["href"] == "file1.html" + + +def test_parse_complex_manifest_with_external_metadata(): + """Complex manifest with external metadata references resolves metadata properly.""" + with create_zip_with_manifest( + "complete_manifest_with_external_metadata.xml", + "metadata_hummingbirds_course.xml", + "metadata_hummingbirds_organization.xml", + ) as zip_path: + result = parse_imscp_manifest(zip_path) + + assert ( + result["identifier"] + == "com.example.hummingbirds.contentpackaging.metadata.2024" + ) + assert result["metadata"]["title"] == "Discovering Hummingbirds" + assert "hummingbirds" in result["metadata"]["keyword"] + assert result["metadata"]["language"] == "en" + + # Should have 1 organization with 3 items + assert len(result["organizations"]) == 1 + org = result["organizations"][0] + assert org["title"] == "Discovering Hummingbirds" + assert len(org["children"]) == 3 + + # Check items have resources linked + item1 = org["children"][0] + assert item1["title"] == "Introduction to Hummingbirds" + assert item1["href"] == "intro.html" + assert "intro.html" in item1["files"] + + +def test_parse_manifest_language_detection(): + """Language codes like en-US in metadata are detected.""" + with create_zip_with_manifest( + "complete_manifest_with_external_metadata.xml", + "metadata_hummingbirds_course.xml", + "metadata_hummingbirds_organization.xml", + ) as zip_path: + result = parse_imscp_manifest(zip_path) + + # Item 1 has language "en-US" in its metadata + org = result["organizations"][0] + item1 = org["children"][0] + assert item1["metadata"].get("language") == "en-US" + + +def test_parse_manifest_with_explicit_language(): + """Passing language parameter uses it for preferred denesting.""" + with create_zip_with_manifest("simple_manifest.xml") as zip_path: + result = parse_imscp_manifest(zip_path, language="en") + + assert result["metadata"]["title"] == "Test File" + + +# --- Flattening tests --- + + +def test_flatten_single_child_organizations(): + """Single-child topic chains are collapsed.""" + item = { + "title": "Outer", + "metadata": {}, + "children": [ + { + "title": "Inner", + "metadata": {"description": "inner desc"}, + "children": [ + {"title": "Leaf1", "metadata": {}, "href": "page1.html"}, + {"title": "Leaf2", "metadata": {}, "href": "page2.html"}, + ], + } + ], + } + result = flatten_single_child_topics(item) + + # Outer should be collapsed into Inner + assert result["title"] == "Inner" + assert len(result["children"]) == 2 + assert result["children"][0]["title"] == "Leaf1" + + +def test_flatten_preserves_multi_child(): + """Multi-child structures are not flattened.""" + item = { + "title": "Parent", + "metadata": {}, + "children": [ + {"title": "Child1", "metadata": {}, "href": "page1.html"}, + {"title": "Child2", "metadata": {}, "href": "page2.html"}, + ], + } + result = flatten_single_child_topics(item) + + assert result["title"] == "Parent" + assert len(result["children"]) == 2 + + +def test_flatten_deep_single_chain(): + """Deep single-child chains are fully collapsed.""" + item = { + "title": "L1", + "metadata": {}, + "children": [ + { + "title": "L2", + "metadata": {}, + "children": [ + { + "title": "L3", + "metadata": {}, + "children": [ + {"title": "Leaf", "metadata": {}, "href": "page.html"}, + ], + } + ], + } + ], + } + result = flatten_single_child_topics(item) + + # L1 -> L2 -> L3 (single child that is a leaf, not a topic) should stop at L3 + # L3 has one child that is a leaf (no children key), so L3 stays + # L2 has one child (L3) that is a topic, so L2 collapses to L3 + # L1 has one child (L2->L3) that is a topic, so L1 collapses to L3 + assert result["title"] == "L3" + assert len(result["children"]) == 1 + assert result["children"][0]["title"] == "Leaf" + + +def test_flatten_preserves_leaf(): + """Leaf items (no children) are unchanged.""" + item = {"title": "Leaf", "metadata": {}, "href": "page.html"} + result = flatten_single_child_topics(item) + assert result["title"] == "Leaf" + assert result["href"] == "page.html" + + +def test_collect_resources_dangling_identifierref(): + """Dangling identifierref is handled gracefully.""" + with create_zip_with_manifest("manifest_dangling_ref.xml") as zip_path: + result = parse_imscp_manifest(zip_path) + org = result["organizations"][0] + valid_item = next(c for c in org["children"] if c["title"] == "Valid Item") + assert valid_item.get("href") == "valid.html" + dangling_item = next(c for c in org["children"] if c["title"] == "Dangling Item") + assert "href" not in dangling_item + + +# --- Manifest with bad encoding test --- + + +def test_manifest_with_bad_encoding(): + """chardet fallback handles non-UTF-8 manifest files.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + # Create a manifest with Latin-1 encoded content + manifest_content = b""" + + + + + + <langstring xml:lang="de">Einf\xfchrung</langstring> + + de + + + + + + Test + + + +""" + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("imsmanifest.xml", manifest_content) + + result = parse_imscp_manifest(temp_zip_path) + assert result["metadata"]["title"] == "Einführung" + finally: + os.remove(temp_zip_path) + + +# --- has_imscp_manifest tests --- + + +def test_has_imscp_manifest_true(): + with create_zip_with_manifest("simple_manifest.xml") as zip_path: + assert has_imscp_manifest(zip_path) is True + + +def test_has_imscp_manifest_false(): + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("index.html", "Hello") + assert has_imscp_manifest(temp_zip_path) is False + finally: + os.remove(temp_zip_path) + + +def test_has_imscp_manifest_nonexistent_file(): + assert has_imscp_manifest("/nonexistent/path.zip") is False + + +# --- SCORM metadata to content node fields tests --- + + +def test_metadata_dict_to_content_node_fields_basic(): + """Title and description are mapped.""" + metadata = {"title": "My Course", "description": "Course description"} + result = metadata_dict_to_content_node_fields(metadata) + assert result["title"] == "My Course" + assert result["description"] == "Course description" + + +def test_metadata_dict_to_content_node_fields_educational(): + """Learning activities and resource types are mapped from SCORM educational metadata.""" + metadata = { + "learningResourceType": ["exercise", "simulation"], + "interactivityType": "active", + "interactivityLevel": "high", + } + result = metadata_dict_to_content_node_fields(metadata) + assert "learning_activities" in result + assert "resource_types" in result + assert len(result["learning_activities"]) > 0 + assert len(result["resource_types"]) > 0 + + +def test_metadata_dict_to_content_node_fields_keywords(): + """Keywords are mapped to extra_fields.tags.""" + metadata = {"keyword": ["hummingbirds", "birds"]} + result = metadata_dict_to_content_node_fields(metadata) + assert result["tags"] == ["hummingbirds", "birds"] + + +def test_metadata_dict_to_content_node_fields_keywords_string(): + """Single keyword string is wrapped in a list.""" + metadata = {"keyword": "biology"} + result = metadata_dict_to_content_node_fields(metadata) + assert result["tags"] == ["biology"] + + +def test_metadata_dict_to_content_node_fields_empty(): + """Empty dict returns empty result.""" + result = metadata_dict_to_content_node_fields({}) + assert result == {} + + +def test_metadata_dict_to_content_node_fields_language(): + """Language code is mapped and normalized.""" + metadata = {"language": "en-US"} + result = metadata_dict_to_content_node_fields(metadata) + assert result["language"] == "en" + + +def test_metadata_dict_to_content_node_fields_difficulty(): + """Easy difficulty infers FOR_BEGINNERS learner need.""" + metadata = {"difficulty": "easy"} + result = metadata_dict_to_content_node_fields(metadata) + assert "learner_needs" in result + assert len(result["learner_needs"]) > 0 + + +# --- Rights/lifecycle integration tests --- + + +def test_metadata_dict_to_content_node_fields_rights(): + """CC license is inferred from rights_description.""" + from le_utils.constants import licenses + + metadata = { + "rights_description": "Content is protected under Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License.", + "copyrightAndOtherRestrictions": "yes", + } + result = metadata_dict_to_content_node_fields(metadata) + assert result["license"] == licenses.CC_BY_NC_SA + + +def test_scorm_learning_activities_single_string(): + """Single string learningResourceType should map to one activity, not iterate chars.""" + from ricecooker.utils.SCORM_metadata import map_scorm_to_le_utils_activities + from le_utils.constants.labels import learning_activities + + result = map_scorm_to_le_utils_activities({"learningResourceType": "exercise"}) + assert result == [learning_activities.PRACTICE] + + +def test_scorm_resource_types_single_string(): + """Single string learningResourceType should map to one resource type, not iterate chars.""" + from ricecooker.utils.SCORM_metadata import map_scorm_to_educator_resource_types + from le_utils.constants.labels import resource_type + + result = map_scorm_to_educator_resource_types( + {"learningResourceType": "exercise"} + ) + assert result == [resource_type.EXERCISE] + + +def test_empty_title_raises_manifest_parse_error(): + """Whitespace-only raises ManifestParseError, not AssertionError.""" + from ricecooker.utils.imscp import ManifestParseError + + manifest_content = b"""<?xml version="1.0" encoding="UTF-8"?> +<manifest> + <metadata><lom><general><title><string>Test</string> + + + + Item + + + + + + + +""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("imsmanifest.xml", manifest_content) + with pytest.raises(ManifestParseError): + parse_imscp_manifest(temp_zip_path) + finally: + os.remove(temp_zip_path) + + +def test_metadata_dict_to_content_node_fields_lifecycle(): + """Provider is extracted from lifecycle contribute data.""" + metadata = { + "contribute": { + "role": {"value": "publisher"}, + "entity": "BEGIN:VCARD\nVERSION:2.1\nFN:John Doe\nORG:Example Organization\nEND:VCARD", + } + } + result = metadata_dict_to_content_node_fields(metadata) + assert result["provider"] == "Example Organization" diff --git a/tests/testcontent/samples/ims_xml/complete_manifest_with_external_metadata.xml b/tests/testcontent/samples/ims_xml/complete_manifest_with_external_metadata.xml new file mode 100644 index 00000000..804647ec --- /dev/null +++ b/tests/testcontent/samples/ims_xml/complete_manifest_with_external_metadata.xml @@ -0,0 +1,103 @@ + + + + + + ADL SCORM + 2004 3rd Edition + + metadata_hummingbirds_course.xml + + + + + Discovering Hummingbirds + + + metadata_hummingbirds_organization.xml + + + + + Introduction to Hummingbirds + + + + + Explore the fascinating world of hummingbirds, their unique characteristics and behaviors. + + en-US + + + + + + + + Hummingbird Habitats + + + + + Learn about various habitats of hummingbirds, from tropical jungles to backyard gardens. + + + + + + + + + Hummingbird Species + + + + + Discover the diversity of hummingbird species and their unique adaptations. + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/testcontent/samples/ims_xml/manifest_dangling_ref.xml b/tests/testcontent/samples/ims_xml/manifest_dangling_ref.xml new file mode 100644 index 00000000..4adace59 --- /dev/null +++ b/tests/testcontent/samples/ims_xml/manifest_dangling_ref.xml @@ -0,0 +1,27 @@ + + + + + + <string language="en">Dangling Ref Test</string> + en + + + + + + Test Org + + Valid Item + + + Dangling Item + + + + + + + + + diff --git a/tests/testcontent/samples/ims_xml/metadata_hummingbirds_course.xml b/tests/testcontent/samples/ims_xml/metadata_hummingbirds_course.xml new file mode 100644 index 00000000..cb3c7346 --- /dev/null +++ b/tests/testcontent/samples/ims_xml/metadata_hummingbirds_course.xml @@ -0,0 +1,232 @@ + + + + + URI + com.example.hummingbirds.contentpackaging.metadata.2024 + + + <string language="en-US">Discovering Hummingbirds</string> + <string language="es">Descubriendo Colibríes</string> + + en + + An engaging overview of hummingbirds, covering their biology, behavior, and habitats. This course is designed for enthusiasts and bird watchers of all levels. + + + hummingbirds + + + bird watching + + + ornithology + + + Global range, with a focus on the Americas. + + + LOMv1.0 + hierarchical + + + LOMv1.0 + 2 + + + + + + 1.0 + + + LOMv1.0 + final + + + + LOMv1.0 + publisher + + BEGIN:VCARD +VERSION:2.1 +FN:John Doe +ORG:Example Organization +TEL;WORK;VOICE:(555) 123-4567 +ADR;WORK:;;1234 Main St;Anytown;AN;12345;Country +EMAIL;PREF;INTERNET:john.doe@example.com +END:VCARD + + + 2024-01-01 + + Initial publication date of the course. + + + + + + + + URI + com.example.hummingbirds.metadata.2024 + + LOMv1.0 + SCORM_CAM_v1.3 + en-us + + + + text/html + image/jpeg + video/mp4 + 1048576 + http://www.example.com/hummingbirds + + + + LOMv1.0 + browser + + + LOMv1.0 + any + + + + + No specific installation requirements, accessible via standard web browsers. + + + Optimized for both desktop and mobile platforms. + + + P1H + + The average time to complete the course is approximately 1 hour. + + + + + + + LOMv1.0 + documentary + + + LOMv1.0 + interactive lesson + + + LOMv1.0 + medium + + + LOMv1.0 + medium + + + LOMv1.0 + learner + + + LOMv1.0 + training + + + Age 12 and up + + + LOMv1.0 + easy + + + P1H + + Approximately 1 hour is needed to complete the course. + + + + This course provides an introduction to hummingbirds, suitable for beginners in ornithology and bird watching. + + en-us + + + + + LOMv1.0 + no + + + LOMv1.0 + yes + + + Content is protected under Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License. + + + + + + LOMv1.0 + ispartof + + + + URI + com.example.biologyseries + + + Part of a larger series on biology and wildlife. + + + + + + BEGIN:VCARD +VERSION:2.1 +FN:Jane Smith +ORG:Example Organization +TEL;WORK;VOICE:(555) 987-6543 +ADR;WORK:;;4321 Side St;Anytown;AN;54321;Country +EMAIL;PREF;INTERNET:jane.smith@example.com +END:VCARD + + + 2024-02-01 + + Annotation added to provide additional insights into course content. + + + + This course includes interactive elements to enhance learner engagement with the subject of hummingbirds. + + + + + + LOMv1.0 + discipline + + + + Example Organization's catalog of biology courses + + + ornithology_basics + + Basic Ornithology + + + + + This course serves as an introductory module in the Basic Ornithology series, focusing on hummingbirds. + + + biology + + + wildlife + + + diff --git a/tests/testcontent/samples/ims_xml/metadata_hummingbirds_organization.xml b/tests/testcontent/samples/ims_xml/metadata_hummingbirds_organization.xml new file mode 100644 index 00000000..87018dfa --- /dev/null +++ b/tests/testcontent/samples/ims_xml/metadata_hummingbirds_organization.xml @@ -0,0 +1,14 @@ + + + + + + This metadata provides information about the structure and organization of the Discovering Hummingbirds course. The course is organized in a hierarchical manner, guiding learners from basic concepts to more detailed aspects of hummingbird biology and behavior. + + + + LOMv1.0 + hierarchical + + + diff --git a/tests/testcontent/samples/ims_xml/simple_manifest.xml b/tests/testcontent/samples/ims_xml/simple_manifest.xml new file mode 100644 index 00000000..bcbb2f3a --- /dev/null +++ b/tests/testcontent/samples/ims_xml/simple_manifest.xml @@ -0,0 +1,63 @@ + + + + + + Test File + + en + + Example of test file + + + + + + + Folder 1 + + Test File1 + + + Test File2 + + + Folder 2 + + Test File1 Nested + + + Test File2 Nested + + + + Folder 3 + + Test File3 Nested + + + Test File4 Nested + + + + + Folder 4 + + Test File3 + + + Test File4 + + + + + + + + + + + + + + From 46635accba8a416db48a1bb1883e108f7f147ee1 Mon Sep 17 00:00:00 2001 From: Richard Tibbles Date: Mon, 9 Feb 2026 06:25:41 -0800 Subject: [PATCH 3/5] Add ContentNode support for building nested children from pipeline metadata Extend ContentNode._process_uri() to build child TopicNode/ContentNode trees when the pipeline produces nested ContentNodeMetadata (e.g. from IMSCP manifests with organization hierarchies). Add children and tags fields to ContentNodeMetadata. Simplify IMSCPZipFile and QTIZipFile to plain DownloadFile subclasses with only a preset override, since all parsing now happens in the pipeline. Co-Authored-By: Claude Opus 4.6 --- ricecooker/classes/files.py | 20 ++++++ ricecooker/classes/nodes.py | 104 ++++++++++++++++++++++++++- ricecooker/utils/pipeline/context.py | 3 + 3 files changed, 125 insertions(+), 2 deletions(-) diff --git a/ricecooker/classes/files.py b/ricecooker/classes/files.py index c824a829..f2ca7e33 100644 --- a/ricecooker/classes/files.py +++ b/ricecooker/classes/files.py @@ -288,6 +288,26 @@ class HTMLZipFile(DownloadFile): default_preset = format_presets.HTML5_ZIP +class IMSCPZipFile(DownloadFile): + default_ext = file_formats.HTML5 + allowed_formats = {file_formats.HTML5} + is_primary = True + + def get_preset(self): + return self.preset or format_presets.IMSCP_ZIP + + +class QTIZipFile(DownloadFile): + """Placeholder for QTI zip file support. QTI pipeline handlers are not yet implemented.""" + + default_ext = file_formats.HTML5 + allowed_formats = {file_formats.HTML5} + is_primary = True + + def get_preset(self): + return self.preset or format_presets.QTI_ZIP + + class H5PFile(DownloadFile): default_ext = file_formats.H5P allowed_formats = {file_formats.H5P} diff --git a/ricecooker/classes/nodes.py b/ricecooker/classes/nodes.py index ea3dd33b..814deae6 100644 --- a/ricecooker/classes/nodes.py +++ b/ricecooker/classes/nodes.py @@ -135,7 +135,7 @@ def __init__( thumbnail=None, files=None, derive_thumbnail=False, - node_modifications={}, + node_modifications=None, extra_fields=None, suggested_duration=None, ): @@ -159,7 +159,7 @@ def __init__( self.set_thumbnail(thumbnail) # save modifications passed in by csv - self.node_modifications = node_modifications + self.node_modifications = node_modifications or {} self.author = author or "" self.aggregator = aggregator or "" @@ -795,6 +795,73 @@ def gather_ancestor_metadata(self): metadata = self.parent.gather_ancestor_metadata() return self.get_metadata_dict(metadata) + def _build_children_from_metadata(self, children_list, parent_files=None, license=None): + """Build child TopicNode/ContentNode tree from nested ContentNodeMetadata dicts. + + Called when the pipeline produces nested metadata (e.g. IMSCP manifests). + + Args: + children_list: list of child metadata dicts + parent_files: files to propagate to leaf children (defaults to self.files) + license: license to propagate to leaf ContentNode children (defaults to self.license) + """ + if parent_files is None: + parent_files = list(self.files) + license = license or getattr(self, "license", None) + + for child_data in children_list: + source_id = child_data.get("source_id", child_data.get("title", "unknown")) + title = child_data.get("title", "Untitled") + + nested_children = child_data.get("children") + + if nested_children: + # This is a topic node + node = TopicNode( + source_id, + title, + ) + node._build_children_from_metadata( + nested_children, parent_files=parent_files, license=license + ) + else: + # This is a leaf content node — filter files by preset if specified + file_presets = child_data.get("file_presets", []) + if file_presets: + files = [f for f in parent_files if f.get_preset() in file_presets] + else: + files = [] + extra_fields = child_data.get("extra_fields", {}) + node = ContentNode( + source_id, + title, + license, + files=list(files), + extra_fields=extra_fields, + ) + node.kind = child_data.get("kind", content_kinds.HTML5) + node._files_processed = True + + # Apply additional metadata fields + metadata_fields = [ + "description", + "learning_activities", + "resource_types", + "learner_needs", + "tags", + "language", + ] + for field in metadata_fields: + if field in child_data: + val = child_data[field] + existing = getattr(node, field, None) + if isinstance(existing, list) and isinstance(val, list): + setattr(node, field, existing + val) + elif val: + setattr(node, field, val) + + self.add_child(node) + class TopicNode(TreeNode): """Model representing topic nodes for organizing channel content. @@ -929,9 +996,15 @@ def _process_uri(self): metadata_dict.pop("path", None) file_obj = File(**metadata_dict) self.add_file(file_obj) + + # Pop children before applying metadata to self + children_data = content_metadata.pop("children", None) + for key, value in content_metadata.items(): if key == "extra_fields": self.extra_fields.update(value) + elif key == "tags": + self.tags = self.tags + value else: if key == "kind" and self.kind is not None and self.kind != value: raise InvalidNodeException( @@ -939,10 +1012,37 @@ def _process_uri(self): ) setattr(self, key, value) + # Build child nodes from nested metadata (e.g. IMSCP organizations) + if children_data: + self._dynamic_children = True + self._build_children_from_metadata(children_data) + + def _collect_dynamic_filenames(self, node): + """Recursively collect filenames from dynamically created children. + + Validates ContentNode leaves and recurses into TopicNode children. + """ + filenames = [] + for child in node.children: + if isinstance(child, ContentNode) and not child._files_processed: + child._files_processed = True + child.validate() + for f in child.files: + if f.filename: + filenames.append(f.filename) + if isinstance(child, TopicNode): + filenames.extend(self._collect_dynamic_filenames(child)) + return filenames + def process_files(self): if self.uri: self._process_uri() filenames = super().process_files() + + # Process dynamically created children (recursively) + if getattr(self, "_dynamic_children", False): + filenames.extend(self._collect_dynamic_filenames(self)) + # Now that we have set all the metadata, and files, we validate the node # again to ensure that the metadata is valid self._files_processed = True diff --git a/ricecooker/utils/pipeline/context.py b/ricecooker/utils/pipeline/context.py index 386e177b..ed794484 100644 --- a/ricecooker/utils/pipeline/context.py +++ b/ricecooker/utils/pipeline/context.py @@ -35,6 +35,9 @@ class ContentNodeMetadata: source_id: Optional[str] = None kind: Optional[str] = None extra_fields: Optional[dict] = None + tags: Optional[list[str]] = None + children: Optional[list] = None # list of ContentNodeMetadata dicts + file_presets: Optional[list[str]] = None # presets for filtering parent files def _recursive_update(target, source): From b87317103bfd1dfb29901199e172d0099c7a6b67 Mon Sep 17 00:00:00 2001 From: Richard Tibbles Date: Mon, 9 Feb 2026 06:25:48 -0800 Subject: [PATCH 4/5] Add IMSCP pipeline handlers for conversion and metadata extraction Add IMSCPConversionHandler and IMSCPMetadataExtractor to the file pipeline, ordered before their HTML5 counterparts so IMSCP zips (detected by presence of imsmanifest.xml) are handled correctly. The metadata extractor uses parse_imscp_manifest() and metadata_dict_to_content_node_fields() to produce nested ContentNodeMetadata trees, which ContentNode._process_uri() then expands into TopicNode/ContentNode children. Co-Authored-By: Claude Opus 4.6 --- ricecooker/utils/pipeline/convert.py | 30 + ricecooker/utils/pipeline/extract_metadata.py | 73 ++ tests/test_imscp_pipeline.py | 694 ++++++++++++++++++ 3 files changed, 797 insertions(+) create mode 100644 tests/test_imscp_pipeline.py diff --git a/ricecooker/utils/pipeline/convert.py b/ricecooker/utils/pipeline/convert.py index f1dab505..4ab4d3e9 100644 --- a/ricecooker/utils/pipeline/convert.py +++ b/ricecooker/utils/pipeline/convert.py @@ -35,6 +35,9 @@ from ricecooker.utils.audio import AudioCompressionError from ricecooker.utils.audio import compress_audio from ricecooker.utils.caching import generate_key +from ricecooker.utils.imscp import has_imscp_manifest +from ricecooker.utils.imscp import ManifestParseError +from ricecooker.utils.imscp import parse_manifest_from_zip from ricecooker.utils.pipeline.context import ContextMetadata from ricecooker.utils.pipeline.context import FileMetadata from ricecooker.utils.pipeline.exceptions import InvalidFileException @@ -308,6 +311,32 @@ def _read_and_compress_archive_file( return reader(filepath) +class IMSCPConversionHandler(ArchiveProcessingBaseHandler): + """Validates IMSCP content packages (zip files containing imsmanifest.xml).""" + + EXTENSIONS = {file_formats.HTML5} + FILE_TYPE = "IMSCP" + + def should_handle(self, path: str) -> bool: + if not super().should_handle(path): + return False + return has_imscp_manifest(path) + + def validate_archive(self, path: str): + with self.open_and_verify_archive(path) as zf: + if "imsmanifest.xml" not in zf.namelist(): + raise InvalidFileException( + f"File {path} is not a valid IMSCP file, imsmanifest.xml is missing." + ) + # Verify the manifest is parseable using the already-open zip + try: + parse_manifest_from_zip(zf) + except ManifestParseError as e: + raise InvalidFileException( + f"File {path} has an invalid imsmanifest.xml: {e}" + ) + + class HTML5ConversionHandler(ArchiveProcessingBaseHandler): EXTENSIONS = {file_formats.HTML5} FILE_TYPE = "HTML5" @@ -605,6 +634,7 @@ class ConversionStageHandler(StageHandler): BloomConversionHandler, EPUBConversionHandler, H5PConversionHandler, + IMSCPConversionHandler, HTML5ConversionHandler, KPUBConversionHandler, VideoCompressionHandler, diff --git a/ricecooker/utils/pipeline/extract_metadata.py b/ricecooker/utils/pipeline/extract_metadata.py index 13d3db9c..984c01d8 100644 --- a/ricecooker/utils/pipeline/extract_metadata.py +++ b/ricecooker/utils/pipeline/extract_metadata.py @@ -3,8 +3,11 @@ from .file_handler import ExtensionMatchingHandler from .file_handler import StageHandler +from ricecooker.utils.imscp import has_imscp_manifest +from ricecooker.utils.imscp import parse_imscp_manifest from ricecooker.utils.pipeline.context import ContentNodeMetadata from ricecooker.utils.pipeline.context import FileMetadata +from ricecooker.utils.SCORM_metadata import metadata_dict_to_content_node_fields from ricecooker.utils.utils import extract_path_ext from ricecooker.utils.videos import extract_duration_of_media from ricecooker.utils.videos import guess_video_preset_by_resolution @@ -65,6 +68,75 @@ class PDFMetadataExtractor(MetadataExtractor): EXTENSIONS = {file_formats.PDF} +class IMSCPMetadataExtractor(MetadataExtractor): + """Extracts metadata from IMSCP content packages, producing nested ContentNodeMetadata.""" + + EXTENSIONS = {file_formats.HTML5} + + def should_handle(self, path: str) -> bool: + if not super().should_handle(path): + return False + return has_imscp_manifest(path) + + def infer_preset(self, path): + return format_presets.IMSCP_ZIP + + def _build_children_metadata(self, items): + """Recursively build ContentNodeMetadata list from parsed item dicts.""" + children = [] + for item in items: + fields = metadata_dict_to_content_node_fields(item.get("metadata", {})) + fields["title"] = item.get("title", fields.get("title")) + fields["source_id"] = item.get("identifier", item.get("title")) + + if "children" in item: + # This is a topic node + fields["kind"] = "topic" + fields["children"] = self._build_children_metadata(item["children"]) + else: + # This is a leaf content node + fields["kind"] = "html5" + fields["file_presets"] = [format_presets.IMSCP_ZIP] + entry = item.get("href", "") + if item.get("parameters"): + entry += item["parameters"] + fields["extra_fields"] = {"options": {"entry": entry}} + + children.append( + ContentNodeMetadata( + **{ + k: v + for k, v in fields.items() + if k in ContentNodeMetadata.__dataclass_fields__ + } + ) + ) + return children + + def handle_file(self, path): + manifest = parse_imscp_manifest(path) + root_fields = metadata_dict_to_content_node_fields(manifest.get("metadata", {})) + + organizations = manifest.get("organizations", []) + if organizations: + root_fields["kind"] = "topic" + root_fields["children"] = self._build_children_metadata(organizations) + + preset = self.infer_preset(path) + content_node_metadata = ContentNodeMetadata( + **{ + k: v + for k, v in root_fields.items() + if k in ContentNodeMetadata.__dataclass_fields__ + } + ) + + return FileMetadata( + preset=preset, + content_node_metadata=content_node_metadata, + ) + + class HTML5MetadataExtractor(MetadataExtractor): EXTENSIONS = {file_formats.HTML5} @@ -95,6 +167,7 @@ class ExtractMetadataStageHandler(StageHandler): EPUBMetadataExtractor, PDFMetadataExtractor, H5PMetadataExtractor, + IMSCPMetadataExtractor, HTML5MetadataExtractor, BloomPubMetadataExtractor, KPUBMetadataExtractor, diff --git a/tests/test_imscp_pipeline.py b/tests/test_imscp_pipeline.py new file mode 100644 index 00000000..00245138 --- /dev/null +++ b/tests/test_imscp_pipeline.py @@ -0,0 +1,694 @@ +"""Tests for IMSCP pipeline integration (handlers + ContentNode support).""" +import os +import tempfile +import zipfile +from unittest.mock import MagicMock + +import pytest +from le_utils.constants import content_kinds +from le_utils.constants import format_presets +from le_utils.constants import licenses + +from ricecooker.classes.files import File +from ricecooker.classes.nodes import ContentNode +from ricecooker.classes.nodes import TopicNode +from ricecooker.utils.pipeline.context import ContentNodeMetadata +from ricecooker.utils.pipeline.context import FileMetadata +from ricecooker.utils.pipeline.convert import IMSCPConversionHandler +from ricecooker.utils.pipeline.exceptions import InvalidFileException +from ricecooker.utils.pipeline.extract_metadata import IMSCPMetadataExtractor + + +BASE_PATH = os.path.dirname(os.path.abspath(__file__)) +IMS_XML_DIR = os.path.join(BASE_PATH, "testcontent", "samples", "ims_xml") + + +# --- Fixtures --- + + +@pytest.fixture +def simple_imscp_zip(): + """Create a simple IMSCP zip with manifest.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + manifest_file = os.path.join(IMS_XML_DIR, "simple_manifest.xml") + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.write(manifest_file, "imsmanifest.xml") + # Add index.html so it also qualifies as HTML5 if needed + zf.writestr("index.html", "IMSCP Content") + yield temp_zip_path + try: + os.remove(temp_zip_path) + except (FileNotFoundError, OSError): + pass + + +@pytest.fixture +def complex_imscp_zip(): + """Create a complex IMSCP zip with external metadata.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.write( + os.path.join(IMS_XML_DIR, "complete_manifest_with_external_metadata.xml"), + "imsmanifest.xml", + ) + zf.write( + os.path.join(IMS_XML_DIR, "metadata_hummingbirds_course.xml"), + "metadata_hummingbirds_course.xml", + ) + zf.write( + os.path.join(IMS_XML_DIR, "metadata_hummingbirds_organization.xml"), + "metadata_hummingbirds_organization.xml", + ) + zf.writestr("index.html", "Hummingbirds") + yield temp_zip_path + try: + os.remove(temp_zip_path) + except (FileNotFoundError, OSError): + pass + + +@pytest.fixture +def regular_html5_zip(): + """Create a regular HTML5 zip without manifest.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("index.html", "Regular HTML5") + yield temp_zip_path + try: + os.remove(temp_zip_path) + except (FileNotFoundError, OSError): + pass + + +# --- ConversionHandler tests --- + + +def test_imscp_conversion_handler_should_handle_imscp_zip(simple_imscp_zip): + handler = IMSCPConversionHandler() + assert handler.should_handle(simple_imscp_zip) is True + + +def test_imscp_conversion_handler_should_not_handle_regular_html5(regular_html5_zip): + handler = IMSCPConversionHandler() + assert handler.should_handle(regular_html5_zip) is False + + +def test_imscp_conversion_handler_validates_manifest_exists(simple_imscp_zip): + """Validation passes when imsmanifest.xml is present and parseable.""" + handler = IMSCPConversionHandler() + # Should not raise + handler.validate_archive(simple_imscp_zip) + + +def test_imscp_conversion_handler_validates_invalid_manifest(): + """Validation fails for zip without imsmanifest.xml.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("index.html", "No manifest") + handler = IMSCPConversionHandler() + with pytest.raises(InvalidFileException): + handler.validate_archive(temp_zip_path) + finally: + os.remove(temp_zip_path) + + +# --- MetadataExtractor tests --- + + +def test_imscp_metadata_extractor_should_handle(simple_imscp_zip): + extractor = IMSCPMetadataExtractor() + assert extractor.should_handle(simple_imscp_zip) is True + + +def test_imscp_metadata_extractor_should_not_handle_regular_html5(regular_html5_zip): + extractor = IMSCPMetadataExtractor() + assert extractor.should_handle(regular_html5_zip) is False + + +def test_imscp_metadata_extractor_preset(simple_imscp_zip): + extractor = IMSCPMetadataExtractor() + assert extractor.infer_preset(simple_imscp_zip) == format_presets.IMSCP_ZIP + + +def test_imscp_metadata_extractor_simple_manifest(simple_imscp_zip): + """Simple manifest produces nested ContentNodeMetadata with children.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(simple_imscp_zip) + + assert isinstance(result, FileMetadata) + assert result.preset == format_presets.IMSCP_ZIP + + metadata = result.content_node_metadata + assert isinstance(metadata, ContentNodeMetadata) + assert metadata.title == "Test File" + assert metadata.kind == "topic" + assert metadata.children is not None + assert len(metadata.children) == 2 # 2 organizations + + +def test_imscp_metadata_extractor_complex_manifest(complex_imscp_zip): + """Complex manifest with external metadata produces rich nested metadata.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(complex_imscp_zip) + + metadata = result.content_node_metadata + assert metadata.title == "Discovering Hummingbirds" + assert metadata.kind == "topic" + assert metadata.children is not None + assert len(metadata.children) == 1 # 1 organization + + org = metadata.children[0] + assert isinstance(org, ContentNodeMetadata) + assert org.title == "Discovering Hummingbirds" + assert org.children is not None + assert len(org.children) == 3 + + +def test_imscp_metadata_extractor_leaf_entry_points(complex_imscp_zip): + """Leaf nodes have extra_fields.options.entry set.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(complex_imscp_zip) + + org = result.content_node_metadata.children[0] + leaf = org.children[0] # "Introduction to Hummingbirds" + assert leaf.kind == "html5" + assert leaf.extra_fields is not None + assert leaf.extra_fields["options"]["entry"] == "intro.html" + + +def test_imscp_metadata_extractor_scorm_metadata_applied(complex_imscp_zip): + """SCORM educational metadata is mapped to learning_activities etc.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(complex_imscp_zip) + + metadata = result.content_node_metadata + # The complex manifest has keyword metadata + assert metadata.tags is not None + assert "hummingbirds" in metadata.tags + + +# --- ContentNodeMetadata children field tests --- + + +def test_content_node_metadata_children_field_defaults_none(): + metadata = ContentNodeMetadata() + assert metadata.children is None + + +def test_content_node_metadata_with_children_serializes(): + """FileMetadata.to_dict() recursively serializes nested children.""" + child = ContentNodeMetadata(title="Child", kind="html5") + parent = ContentNodeMetadata(title="Parent", kind="topic", children=[child]) + fm = FileMetadata(content_node_metadata=parent) + result = fm.to_dict() + + assert "content_node_metadata" in result + assert result["content_node_metadata"]["title"] == "Parent" + assert len(result["content_node_metadata"]["children"]) == 1 + assert result["content_node_metadata"]["children"][0]["title"] == "Child" + + +def test_file_metadata_merge_with_children(): + """Merge preserves children structure.""" + child = ContentNodeMetadata(title="Child", kind="html5") + parent = ContentNodeMetadata(title="Parent", kind="topic", children=[child]) + fm1 = FileMetadata(content_node_metadata=parent) + fm2 = FileMetadata(preset="some_preset") + + merged = fm1.merge(fm2) + assert merged.preset == "some_preset" + # children should be preserved through merge + merged_dict = merged.to_dict() + assert len(merged_dict["content_node_metadata"]["children"]) == 1 + + +# --- ContentNode nested children support tests (Step 5) --- + + +def _make_license(): + """Create a license object for testing.""" + from ricecooker.classes.licenses import get_license + + return get_license(licenses.CC_BY, copyright_holder="Test Author") + + +def _make_mock_file(preset=format_presets.IMSCP_ZIP, filename="test.zip"): + """Create a mock File object for testing.""" + f = File(preset=preset) + f.filename = filename + return f + + +def test_content_node_builds_children_from_nested_metadata(): + """_build_children_from_metadata creates TopicNode/ContentNode children.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Topic 1", + "source_id": "topic1", + "kind": "topic", + "children": [ + { + "title": "Leaf 1", + "source_id": "leaf1", + "kind": "html5", + "extra_fields": {"options": {"entry": "page1.html"}}, + }, + ], + }, + { + "title": "Leaf 2", + "source_id": "leaf2", + "kind": "html5", + "extra_fields": {"options": {"entry": "page2.html"}}, + }, + ] + + node._build_children_from_metadata(children_data) + + assert len(node.children) == 2 + assert isinstance(node.children[0], TopicNode) + assert node.children[0].title == "Topic 1" + assert node.children[0].source_id == "topic1" + assert isinstance(node.children[1], ContentNode) + assert node.children[1].title == "Leaf 2" + assert node.children[1].source_id == "leaf2" + + +def test_content_node_becomes_topic_when_has_children(): + """When pipeline returns nested metadata, root ContentNode gets kind=topic.""" + license = _make_license() + + # Build a ContentNodeMetadata with children + child_meta = ContentNodeMetadata( + title="Leaf", + kind="html5", + source_id="leaf1", + extra_fields={"options": {"entry": "index.html"}}, + ) + root_meta = ContentNodeMetadata( + title="Root Topic", kind="topic", children=[child_meta] + ) + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + # Mock pipeline to return our metadata + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + assert node.kind == "topic" + assert node.title == "Root Topic" + + +def test_content_node_children_have_entry_points(): + """Leaf children have correct extra_fields.options.entry.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Page A", + "source_id": "page_a", + "kind": "html5", + "extra_fields": {"options": {"entry": "content/page_a.html"}}, + }, + ] + + node._build_children_from_metadata(children_data) + + child = node.children[0] + assert isinstance(child, ContentNode) + assert child.extra_fields == {"options": {"entry": "content/page_a.html"}} + assert child.kind == content_kinds.HTML5 + + +def test_content_node_children_share_files(): + """Leaf children with file_presets get parent files; topic children do not.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Topic", + "source_id": "topic1", + "kind": "topic", + "children": [ + { + "title": "Leaf", + "source_id": "leaf1", + "kind": "html5", + "file_presets": [format_presets.IMSCP_ZIP], + "extra_fields": {"options": {"entry": "index.html"}}, + }, + ], + }, + ] + + node._build_children_from_metadata(children_data) + + topic_child = node.children[0] + leaf_child = topic_child.children[0] + # Topic children should not get files + assert len(topic_child.files) == 0 + # Leaf children with matching file_presets should get files + assert len(leaf_child.files) > 0 + + +def test_content_node_children_metadata_fields_applied(): + """Metadata fields like description, tags are applied to child nodes.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Leaf", + "source_id": "leaf1", + "kind": "html5", + "description": "A test description", + "tags": ["tag1", "tag2"], + "extra_fields": {"options": {"entry": "index.html"}}, + }, + ] + + node._build_children_from_metadata(children_data) + + child = node.children[0] + assert child.description == "A test description" + assert "tag1" in child.tags + assert "tag2" in child.tags + + +def test_content_node_dynamic_children_flag(): + """_process_uri sets _dynamic_children flag when children data exists.""" + license = _make_license() + + child_meta = ContentNodeMetadata( + title="Leaf", + kind="html5", + source_id="leaf1", + extra_fields={"options": {"entry": "index.html"}}, + ) + root_meta = ContentNodeMetadata(title="Root", kind="topic", children=[child_meta]) + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + assert node._dynamic_children is True + assert len(node.children) == 1 + + +def test_content_node_no_children_no_dynamic_flag(): + """_process_uri does not set _dynamic_children when no children data.""" + license = _make_license() + + root_meta = ContentNodeMetadata(title="Leaf Content", kind="html5") + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + assert not getattr(node, "_dynamic_children", False) + + +def test_content_node_recursive_children_structure(): + """_build_children_from_metadata handles deeply nested structures.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Level 1", + "source_id": "l1", + "kind": "topic", + "children": [ + { + "title": "Level 2", + "source_id": "l2", + "kind": "topic", + "children": [ + { + "title": "Deep Leaf", + "source_id": "deep", + "kind": "html5", + "extra_fields": {"options": {"entry": "deep.html"}}, + }, + ], + }, + ], + }, + ] + + node._build_children_from_metadata(children_data) + + l1 = node.children[0] + assert isinstance(l1, TopicNode) + assert l1.title == "Level 1" + + l2 = l1.children[0] + assert isinstance(l2, TopicNode) + assert l2.title == "Level 2" + + deep = l2.children[0] + assert isinstance(deep, ContentNode) + assert deep.title == "Deep Leaf" + assert deep.extra_fields["options"]["entry"] == "deep.html" + + +def test_topic_children_have_no_files(): + """TopicNode children should not inherit parent files.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Topic", + "source_id": "topic1", + "kind": "topic", + "children": [ + { + "title": "Leaf", + "source_id": "leaf1", + "kind": "html5", + "extra_fields": {"options": {"entry": "index.html"}}, + }, + ], + }, + ] + + node._build_children_from_metadata(children_data) + + topic_child = node.children[0] + assert isinstance(topic_child, TopicNode) + assert len(topic_child.files) == 0 + + +def test_leaf_children_get_files_by_preset(): + """Leaf children with file_presets get only files matching their presets.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + imscp_file = _make_mock_file(preset=format_presets.IMSCP_ZIP, filename="test.zip") + node.add_file(imscp_file) + + children_data = [ + { + "title": "Leaf", + "source_id": "leaf1", + "kind": "html5", + "file_presets": [format_presets.IMSCP_ZIP], + "extra_fields": {"options": {"entry": "index.html"}}, + }, + ] + + node._build_children_from_metadata(children_data) + + child = node.children[0] + assert len(child.files) == 1 + assert child.files[0].get_preset() == format_presets.IMSCP_ZIP + + +def test_children_without_file_presets_get_no_files(): + """Children without file_presets get no files.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Leaf", + "source_id": "leaf1", + "kind": "html5", + "extra_fields": {"options": {"entry": "index.html"}}, + }, + ] + + node._build_children_from_metadata(children_data) + + child = node.children[0] + assert len(child.files) == 0 + + +# --- License propagation tests (Issue 1) --- + + +def test_license_propagated_to_nested_content_nodes(): + """Leaf ContentNodes inside TopicNodes inherit the parent's license.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Topic 1", + "source_id": "topic1", + "kind": "topic", + "children": [ + { + "title": "Leaf 1", + "source_id": "leaf1", + "kind": "html5", + "extra_fields": {"options": {"entry": "page1.html"}}, + }, + ], + }, + ] + + node._build_children_from_metadata(children_data) + + topic = node.children[0] + leaf = topic.children[0] + assert isinstance(leaf, ContentNode) + assert leaf.license == license + + +def test_license_propagated_through_deep_nesting(): + """License propagates through Topic->Topic->Leaf (3 levels).""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + mock_file = _make_mock_file() + node.add_file(mock_file) + + children_data = [ + { + "title": "Level 1", + "source_id": "l1", + "kind": "topic", + "children": [ + { + "title": "Level 2", + "source_id": "l2", + "kind": "topic", + "children": [ + { + "title": "Deep Leaf", + "source_id": "deep", + "kind": "html5", + "extra_fields": {"options": {"entry": "deep.html"}}, + }, + ], + }, + ], + }, + ] + + node._build_children_from_metadata(children_data) + + l1 = node.children[0] + l2 = l1.children[0] + deep = l2.children[0] + assert isinstance(deep, ContentNode) + assert deep.license == license + + +# --- process_files recursive collection tests (Issue 2) --- + + +def test_process_files_collects_grandchild_filenames(): + """process_files collects filenames from grandchildren inside TopicNodes.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + node.kind = content_kinds.TOPIC + node._dynamic_children = True + + # Build Topic -> Leaf structure + children_data = [ + { + "title": "Topic 1", + "source_id": "topic1", + "kind": "topic", + "children": [ + { + "title": "Leaf 1", + "source_id": "leaf1", + "kind": "html5", + "extra_fields": {"options": {"entry": "page1.html"}}, + }, + ], + }, + ] + node._build_children_from_metadata(children_data) + + # Give the grandchild a file with a filename (simulating a processed file) + grandchild = node.children[0].children[0] + mock_file = _make_mock_file(filename="abc123.zip") + grandchild.add_file(mock_file) + + # Simulate what process_files does for dynamic children: + # The current flat loop only checks direct children, missing grandchildren + filenames = node._collect_dynamic_filenames(node) + + assert "abc123.zip" in filenames From 1b191bc3e268f30a8320d74dd473bf8e1cc5b76e Mon Sep 17 00:00:00 2001 From: Richard Tibbles Date: Wed, 11 Feb 2026 15:35:54 -0800 Subject: [PATCH 5/5] Add QTI resource detection and routing in IMSCP pipeline Extend IMSCP manifest parsing to recognize QTI resources (imsqti_* prefix) alongside webcontent. QTI leaf items are mapped to exercise kind with QTI_ZIP preset, while webcontent items retain html5/IMSCP_ZIP. Mixed packages route to IMSCP_ZIP at the file level with per-child preset overrides via shallow copy. Also hardens XML parsing (safe parser, null guards), migrates child metadata from raw dicts to typed ContentNodeMetadata objects, and extracts shared test helpers. Co-Authored-By: Claude Opus 4.6 --- ricecooker/classes/files.py | 6 +- ricecooker/classes/nodes.py | 167 +++- ricecooker/utils/imscp.py | 121 ++- ricecooker/utils/pipeline/context.py | 37 +- ricecooker/utils/pipeline/extract_metadata.py | 51 +- tests/helpers.py | 39 + tests/test_imscp.py | 310 ++++++- tests/test_imscp_pipeline.py | 818 ++++++++++++++---- .../samples/ims_xml/mixed_manifest.xml | 48 + .../samples/ims_xml/qti_manifest.xml | 34 + 10 files changed, 1358 insertions(+), 273 deletions(-) create mode 100644 tests/helpers.py create mode 100644 tests/testcontent/samples/ims_xml/mixed_manifest.xml create mode 100644 tests/testcontent/samples/ims_xml/qti_manifest.xml diff --git a/ricecooker/classes/files.py b/ricecooker/classes/files.py index f2ca7e33..f2fc63cf 100644 --- a/ricecooker/classes/files.py +++ b/ricecooker/classes/files.py @@ -298,7 +298,11 @@ def get_preset(self): class QTIZipFile(DownloadFile): - """Placeholder for QTI zip file support. QTI pipeline handlers are not yet implemented.""" + """File class for QTI (Question and Test Interoperability) zip packages. + + Used for exercise-kind leaf nodes within IMSCP packages that contain + QTI assessment resources (resource type prefix ``imsqti_``). + """ default_ext = file_formats.HTML5 allowed_formats = {file_formats.HTML5} diff --git a/ricecooker/classes/nodes.py b/ricecooker/classes/nodes.py index 814deae6..b671c0b9 100644 --- a/ricecooker/classes/nodes.py +++ b/ricecooker/classes/nodes.py @@ -1,4 +1,5 @@ # Node models to represent channel's tree +import copy import json import re import uuid @@ -795,13 +796,27 @@ def gather_ancestor_metadata(self): metadata = self.parent.gather_ancestor_metadata() return self.get_metadata_dict(metadata) - def _build_children_from_metadata(self, children_list, parent_files=None, license=None): - """Build child TopicNode/ContentNode tree from nested ContentNodeMetadata dicts. + def _build_children_from_metadata( + self, children_list, parent_files=None, license=None + ): + """Build child TopicNode/ContentNode tree from nested ContentNodeMetadata. Called when the pipeline produces nested metadata (e.g. IMSCP manifests). + File copy semantics: + - Leaf children with a ``file_preset`` receive shallow copies + (``copy.copy``) of the parent files, with the copy's preset + overridden to the child's ``file_preset``. + - Leaf children without a ``file_preset`` receive no files. + - TopicNode children never receive files. + + Note: Leaf children are created as ContentNode which requires a license. + The license defaults to self.license, so this method is primarily suited + for calls from ContentNode (which has a license attribute). Calling from + TopicNode requires passing the license argument explicitly. + Args: - children_list: list of child metadata dicts + children_list: list of ContentNodeMetadata objects parent_files: files to propagate to leaf children (defaults to self.files) license: license to propagate to leaf ContentNode children (defaults to self.license) """ @@ -809,29 +824,33 @@ def _build_children_from_metadata(self, children_list, parent_files=None, licens parent_files = list(self.files) license = license or getattr(self, "license", None) - for child_data in children_list: - source_id = child_data.get("source_id", child_data.get("title", "unknown")) - title = child_data.get("title", "Untitled") - - nested_children = child_data.get("children") + for child_meta in children_list: + source_id = child_meta.source_id or child_meta.title or "unknown" + title = child_meta.title or "Untitled" - if nested_children: + if child_meta.children: # This is a topic node node = TopicNode( source_id, title, ) node._build_children_from_metadata( - nested_children, parent_files=parent_files, license=license + child_meta.children, parent_files=parent_files, license=license ) else: - # This is a leaf content node — filter files by preset if specified - file_presets = child_data.get("file_presets", []) - if file_presets: - files = [f for f in parent_files if f.get_preset() in file_presets] + # This is a leaf content node — copy files with child's preset. + # Shallow copy is safe here: File attributes are all simple + # types (str, int, None), so no mutable state is shared. + file_preset = child_meta.file_preset + if file_preset: + files = [] + for f in parent_files: + f_copy = copy.copy(f) + f_copy.preset = file_preset + files.append(f_copy) else: files = [] - extra_fields = child_data.get("extra_fields", {}) + extra_fields = child_meta.extra_fields or {} node = ContentNode( source_id, title, @@ -839,7 +858,7 @@ def _build_children_from_metadata(self, children_list, parent_files=None, licens files=list(files), extra_fields=extra_fields, ) - node.kind = child_data.get("kind", content_kinds.HTML5) + node.kind = child_meta.kind or content_kinds.HTML5 node._files_processed = True # Apply additional metadata fields @@ -852,8 +871,8 @@ def _build_children_from_metadata(self, children_list, parent_files=None, licens "language", ] for field in metadata_fields: - if field in child_data: - val = child_data[field] + val = getattr(child_meta, field, None) + if val is not None: existing = getattr(node, field, None) if isinstance(existing, list) and isinstance(val, list): setattr(node, field, existing + val) @@ -979,6 +998,37 @@ def _validate(self): self._validate_uri() return super(ContentNode, self)._validate() + _FILE_INIT_KEYS = { + "preset", + "language", + "default_ext", + "source_url", + "duration", + "original_filename", + "filename", + } + + # Fields from ContentNodeMetadata that _process_uri may apply to the node. + # Identity/legal fields (source_id, license, license_description, + # copyright_holder) are intentionally excluded to prevent pipeline + # metadata from overwriting values set in the constructor. + _METADATA_APPLY_FIELDS = ( + "title", + "description", + "thumbnail", + "author", + "aggregator", + "provider", + "language", + "grade_levels", + "categories", + "resource_types", + "learning_activities", + "accessibility_labels", + "learner_needs", + "role", + ) + def _process_uri(self): try: file_metadata_list = self.pipeline.execute( @@ -987,40 +1037,61 @@ def _process_uri(self): except (InvalidFileException, ExpectedFileException) as e: config.LOGGER.error(f"Error processing path: {self.uri} with error: {e}") return None - content_metadata = {} + content_node_metadata = None for file_metadata in file_metadata_list: - metadata_dict = file_metadata.to_dict() - if "content_node_metadata" in metadata_dict: - content_metadata.update(metadata_dict.pop("content_node_metadata")) - # Remove path from metadata_dict as it is not needed for the File object - metadata_dict.pop("path", None) - file_obj = File(**metadata_dict) - self.add_file(file_obj) - - # Pop children before applying metadata to self - children_data = content_metadata.pop("children", None) - - for key, value in content_metadata.items(): - if key == "extra_fields": - self.extra_fields.update(value) - elif key == "tags": - self.tags = self.tags + value - else: - if key == "kind" and self.kind is not None and self.kind != value: - raise InvalidNodeException( - "Inferred kind is different from content node class kind." - ) - setattr(self, key, value) + # Extract ContentNodeMetadata as typed object (not via to_dict) + if file_metadata.content_node_metadata is not None: + content_node_metadata = file_metadata.content_node_metadata + self._add_file_from_metadata(file_metadata) + + if content_node_metadata is not None: + self._apply_content_node_metadata(content_node_metadata) + + def _add_file_from_metadata(self, file_metadata): + """Create a File object from FileMetadata and add it to this node.""" + metadata_dict = file_metadata.to_dict() + metadata_dict.pop("content_node_metadata", None) + metadata_dict.pop("path", None) + # Filter to only keys accepted by File.__init__ + dropped = set(metadata_dict.keys()) - self._FILE_INIT_KEYS + if dropped: + config.LOGGER.debug( + "Dropped keys from FileMetadata when creating File: %s", dropped + ) + file_kwargs = { + k: v for k, v in metadata_dict.items() if k in self._FILE_INIT_KEYS + } + self.add_file(File(**file_kwargs)) + + def _apply_content_node_metadata(self, content_node_metadata): + """Apply ContentNodeMetadata fields to this node and build children. + See _METADATA_APPLY_FIELDS for the allowlist and exclusion rationale. + """ + for field_name in self._METADATA_APPLY_FIELDS: + value = getattr(content_node_metadata, field_name, None) + if value is not None: + setattr(self, field_name, value) + if content_node_metadata.extra_fields is not None: + self.extra_fields.update(content_node_metadata.extra_fields) + if content_node_metadata.tags is not None: + self.tags = self.tags + content_node_metadata.tags + kind = content_node_metadata.kind + if kind is not None: + if self.kind is not None and self.kind != kind: + raise InvalidNodeException( + "Inferred kind is different from content node class kind." + ) + self.kind = kind # Build child nodes from nested metadata (e.g. IMSCP organizations) - if children_data: + if content_node_metadata.children: self._dynamic_children = True - self._build_children_from_metadata(children_data) + self._build_children_from_metadata(content_node_metadata.children) - def _collect_dynamic_filenames(self, node): + def _validate_and_collect_dynamic_filenames(self, node): """Recursively collect filenames from dynamically created children. - Validates ContentNode leaves and recurses into TopicNode children. + Validates ContentNode leaves and recurses into any node with children. """ filenames = [] for child in node.children: @@ -1030,8 +1101,8 @@ def _collect_dynamic_filenames(self, node): for f in child.files: if f.filename: filenames.append(f.filename) - if isinstance(child, TopicNode): - filenames.extend(self._collect_dynamic_filenames(child)) + if child.children: + filenames.extend(self._validate_and_collect_dynamic_filenames(child)) return filenames def process_files(self): @@ -1041,7 +1112,7 @@ def process_files(self): # Process dynamically created children (recursively) if getattr(self, "_dynamic_children", False): - filenames.extend(self._collect_dynamic_filenames(self)) + filenames.extend(self._validate_and_collect_dynamic_filenames(self)) # Now that we have set all the metadata, and files, we validate the node # again to ensure that the metadata is valid diff --git a/ricecooker/utils/imscp.py b/ricecooker/utils/imscp.py index 1a7ffc8d..81846899 100644 --- a/ricecooker/utils/imscp.py +++ b/ricecooker/utils/imscp.py @@ -13,8 +13,24 @@ from ricecooker.utils.SCORM_metadata import imscp_metadata_keys +_SAFE_PARSER = etree.XMLParser(resolve_entities=False, no_network=True) + logger = logging.getLogger(__name__) +QTI_RESOURCE_TYPE_PREFIX = "imsqti_" + + +def is_qti_resource(resource_type): + """Check whether a resource type string indicates a QTI resource. + + Args: + resource_type: the type attribute from a element + + Returns: + bool + """ + return bool(resource_type) and resource_type.startswith(QTI_RESOURCE_TYPE_PREFIX) + class ManifestParseError(Exception): """Raised when an imsmanifest.xml cannot be parsed.""" @@ -35,7 +51,7 @@ def _get_elem_for_tag(root, tag): return root.find(tag) -def _extract_lom_text(elem, preferred_language): +def _extract_lom_text(elem, preferred_language) -> "str | list[str] | None": """Extract text from a LOM element using direct lxml traversal. Handles common LOM XML patterns: @@ -43,6 +59,11 @@ def _extract_lom_text(elem, preferred_language): - text - LOMv1.0text - plain text + + Returns: + A single string when one match is found (or a preferred-language match), + a list of strings when multiple matches exist and none is preferred, + or None when the element has no extractable text. """ strings = elem.findall("string") or elem.findall("langstring") if strings: @@ -89,8 +110,13 @@ def _resolve_metadata_elem(root, zip_file): namespaces={"adlcp": "http://www.adlnet.org/xsd/adlcp_v1p3"}, ) if external_ref is not None and zip_file is not None: - with zip_file.open(external_ref.text) as external_file: - metadata_elem = etree.parse(external_file).getroot() + try: + with zip_file.open(external_ref.text) as external_file: + metadata_elem = etree.parse(external_file, _SAFE_PARSER).getroot() + except KeyError: + raise ManifestParseError( + "External metadata file not found in zip: {}".format(external_ref.text) + ) strip_ns_prefix(metadata_elem) return metadata_elem @@ -170,7 +196,7 @@ def parse_manifest_from_zip(zf): """ try: with zf.open("imsmanifest.xml") as manifest_file: - return etree.parse(manifest_file).getroot() + return etree.parse(manifest_file, _SAFE_PARSER).getroot() except (etree.XMLSyntaxError, OSError): pass @@ -184,7 +210,7 @@ def parse_manifest_from_zip(zf): info = chardet.detect(data) encoding = info["encoding"] or "utf-8" data = data.decode(encoding) - return etree.parse(io.BytesIO(data.encode("utf-8"))).getroot() + return etree.parse(io.BytesIO(data.encode("utf-8")), _SAFE_PARSER).getroot() except (etree.XMLSyntaxError, OSError, UnicodeDecodeError) as e: raise ManifestParseError(str(e)) from e @@ -280,7 +306,8 @@ def collect_resources(item, resources_dict): if key_stripped not in item: item[key_stripped] = value - if resource_elem.get("type") == "webcontent": + resource_type = resource_elem.get("type", "") + if resource_type == "webcontent" or is_qti_resource(resource_type): item["files"] = derive_content_files_dict(resource_elem, resources_dict) @@ -305,7 +332,9 @@ def flatten_single_child_topics(item): if len(item["children"]) == 1: only_child = item["children"][0] if "children" in only_child: - # Merge: keep child's structure but preserve parent's metadata if richer + # Merge: keep child's structure but use parent's title when child lacks one + if not only_child.get("title"): + only_child["title"] = item.get("title", "") if not only_child.get("metadata"): only_child["metadata"] = item.get("metadata", {}) return only_child @@ -337,7 +366,11 @@ def parse_imscp_manifest(zip_path, language=None): metadata = collect_metadata(manifest, zip_file=zf, language=language) resources_elem = manifest.find("resources", nsmap) - resources_dict = dict((r.get("identifier"), r) for r in resources_elem) + resources_dict = ( + {r.get("identifier"): r for r in resources_elem} + if resources_elem is not None + else {} + ) organizations = [] for org_elem in manifest.findall("organizations/organization", nsmap): @@ -367,3 +400,75 @@ def has_imscp_manifest(zip_path): return "imsmanifest.xml" in zf.namelist() except (zipfile.BadZipFile, FileNotFoundError): return False + + +def has_qti_resources(zip_path): + """Check whether an IMSCP zip contains any QTI resources. + + This scans raw elements without full manifest parsing. + Prefer has_qti_items() when you already have a parsed manifest dict. + + Args: + zip_path: path to the zip file + + Returns: + bool + """ + try: + manifest = get_manifest(zip_path) + except (zipfile.BadZipFile, FileNotFoundError, KeyError, ManifestParseError): + return False + + nsmap = manifest.nsmap + resources_elem = manifest.find("resources", nsmap) + if resources_elem is None: + return False + + for resource in resources_elem: + if is_qti_resource(resource.get("type", "")): + return True + return False + + +def _collect_leaf_types(item): + """Recursively yield resource type strings from leaf items in a parsed manifest tree.""" + if "children" in item: + for child in item["children"]: + yield from _collect_leaf_types(child) + elif "type" in item: + yield item["type"] + + +def has_qti_items(parsed_manifest): + """Check whether a parsed manifest dict contains any QTI resource items. + + Operates on an already-parsed manifest. Prefer this over has_qti_resources() + when you already have the result of parse_imscp_manifest(). + + Args: + parsed_manifest: dict returned by parse_imscp_manifest() + + Returns: + bool + """ + for org in parsed_manifest.get("organizations", []): + for leaf_type in _collect_leaf_types(org): + if is_qti_resource(leaf_type): + return True + return False + + +def has_webcontent_items(parsed_manifest): + """Check whether a parsed manifest dict contains any webcontent resource items. + + Args: + parsed_manifest: dict returned by parse_imscp_manifest() + + Returns: + bool + """ + for org in parsed_manifest.get("organizations", []): + for leaf_type in _collect_leaf_types(org): + if leaf_type == "webcontent": + return True + return False diff --git a/ricecooker/utils/pipeline/context.py b/ricecooker/utils/pipeline/context.py index ed794484..7fceb011 100644 --- a/ricecooker/utils/pipeline/context.py +++ b/ricecooker/utils/pipeline/context.py @@ -1,8 +1,13 @@ +from __future__ import annotations + +import logging from dataclasses import asdict from dataclasses import dataclass from typing import Optional from typing import Type +logger = logging.getLogger(__name__) + class AutoDataClassMetaClass(type): def __new__(mcs, name: str, bases: tuple, namespace: dict) -> Type: @@ -14,6 +19,10 @@ def __new__(mcs, name: str, bases: tuple, namespace: dict) -> Type: class ContentNodeMetadata: """ A dataclass for storing metadata about a content node. + + Intentionally mutable (not frozen) because MetadataExtractor.handle_file() + sets .kind in place after construction (see extract_metadata.py line ~50). + This differs from ContextMetadata which uses frozen=True. """ title: Optional[str] = None @@ -32,12 +41,32 @@ class ContentNodeMetadata: accessibility_labels: Optional[list[str]] = None learner_needs: Optional[list[str]] = None role: Optional[str] = None + language: Optional[str] = None source_id: Optional[str] = None kind: Optional[str] = None extra_fields: Optional[dict] = None tags: Optional[list[str]] = None - children: Optional[list] = None # list of ContentNodeMetadata dicts - file_presets: Optional[list[str]] = None # presets for filtering parent files + children: Optional[list[ContentNodeMetadata]] = None + file_preset: Optional[str] = None # preset to assign to parent file for this child + + +def _content_node_metadata_from_dict(d): + """Reconstruct a ContentNodeMetadata (with nested children) from a dict.""" + children = d.get("children") + if children is not None: + d = dict( + d, + children=[ + _content_node_metadata_from_dict(c) if isinstance(c, dict) else c + for c in children + ], + ) + valid_keys = ContentNodeMetadata.__dataclass_fields__ + dropped = {k for k in d if k not in valid_keys} + if dropped: + logger.debug("Dropped unknown keys from ContentNodeMetadata dict: %s", dropped) + valid = {k: v for k, v in d.items() if k in valid_keys} + return ContentNodeMetadata(**valid) def _recursive_update(target, source): @@ -72,6 +101,10 @@ def merge(self, other): fields with other fields when defined. """ new_dict = _recursive_update(self.to_dict(), other.to_dict()) + # Reconstruct ContentNodeMetadata from dict if present + cnm = new_dict.get("content_node_metadata") + if isinstance(cnm, dict): + new_dict["content_node_metadata"] = _content_node_metadata_from_dict(cnm) return self.__class__(**new_dict) diff --git a/ricecooker/utils/pipeline/extract_metadata.py b/ricecooker/utils/pipeline/extract_metadata.py index 984c01d8..dd077ace 100644 --- a/ricecooker/utils/pipeline/extract_metadata.py +++ b/ricecooker/utils/pipeline/extract_metadata.py @@ -1,10 +1,15 @@ +from le_utils.constants import content_kinds from le_utils.constants import file_formats from le_utils.constants import format_presets from .file_handler import ExtensionMatchingHandler from .file_handler import StageHandler from ricecooker.utils.imscp import has_imscp_manifest +from ricecooker.utils.imscp import has_qti_items +from ricecooker.utils.imscp import has_webcontent_items +from ricecooker.utils.imscp import is_qti_resource from ricecooker.utils.imscp import parse_imscp_manifest +from ricecooker.utils.pipeline.context import _content_node_metadata_from_dict from ricecooker.utils.pipeline.context import ContentNodeMetadata from ricecooker.utils.pipeline.context import FileMetadata from ricecooker.utils.SCORM_metadata import metadata_dict_to_content_node_fields @@ -78,7 +83,20 @@ def should_handle(self, path: str) -> bool: return False return has_imscp_manifest(path) - def infer_preset(self, path): + def _infer_preset_from_manifest(self, manifest): + """Infer the primary preset from a parsed manifest dict. + + Returns QTI_ZIP for pure QTI packages, IMSCP_ZIP for everything else + (pure webcontent, mixed QTI+webcontent, or manifests with no + organizations/leaf items). + """ + has_qti = has_qti_items(manifest) + has_webcontent = has_webcontent_items(manifest) + if has_qti and not has_webcontent: + # Pure QTI package (assessments only) + return format_presets.QTI_ZIP + # IMSCP for: pure webcontent, mixed (QTI + webcontent), or + # unknown resource types (safe default for any IMS package) return format_presets.IMSCP_ZIP def _build_children_metadata(self, items): @@ -94,23 +112,20 @@ def _build_children_metadata(self, items): fields["kind"] = "topic" fields["children"] = self._build_children_metadata(item["children"]) else: - # This is a leaf content node - fields["kind"] = "html5" - fields["file_presets"] = [format_presets.IMSCP_ZIP] + # This is a leaf content node — detect QTI vs webcontent + resource_type = item.get("type", "") + if is_qti_resource(resource_type): + fields["kind"] = content_kinds.EXERCISE + fields["file_preset"] = format_presets.QTI_ZIP + else: + fields["kind"] = content_kinds.HTML5 + fields["file_preset"] = format_presets.IMSCP_ZIP entry = item.get("href", "") if item.get("parameters"): entry += item["parameters"] fields["extra_fields"] = {"options": {"entry": entry}} - children.append( - ContentNodeMetadata( - **{ - k: v - for k, v in fields.items() - if k in ContentNodeMetadata.__dataclass_fields__ - } - ) - ) + children.append(_content_node_metadata_from_dict(fields)) return children def handle_file(self, path): @@ -122,14 +137,8 @@ def handle_file(self, path): root_fields["kind"] = "topic" root_fields["children"] = self._build_children_metadata(organizations) - preset = self.infer_preset(path) - content_node_metadata = ContentNodeMetadata( - **{ - k: v - for k, v in root_fields.items() - if k in ContentNodeMetadata.__dataclass_fields__ - } - ) + preset = self._infer_preset_from_manifest(manifest) + content_node_metadata = _content_node_metadata_from_dict(root_fields) return FileMetadata( preset=preset, diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 00000000..7f02a786 --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,39 @@ +"""Shared test helper utilities.""" +import os +import tempfile +import zipfile +from contextlib import contextmanager + + +IMS_XML_DIR = os.path.join( + os.path.dirname(os.path.abspath(__file__)), "testcontent", "samples", "ims_xml" +) + + +@contextmanager +def create_zip_with_manifest(manifest_filename, *additional_files, extra_entries=None): + """Create a temp zip with a manifest and optional additional files. + + Args: + manifest_filename: XML file in IMS_XML_DIR to use as imsmanifest.xml + *additional_files: additional files from IMS_XML_DIR to include + extra_entries: dict of {arcname: content_string} for in-memory entries + """ + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + manifest_file = os.path.join(IMS_XML_DIR, manifest_filename) + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.write(manifest_file, "imsmanifest.xml") + for additional_file in additional_files: + zf.write(os.path.join(IMS_XML_DIR, additional_file), additional_file) + if extra_entries: + for arcname, content in extra_entries.items(): + zf.writestr(arcname, content) + yield temp_zip_path + finally: + try: + os.remove(temp_zip_path) + except (FileNotFoundError, OSError): + pass diff --git a/tests/test_imscp.py b/tests/test_imscp.py index b1a596e8..50d888a6 100644 --- a/tests/test_imscp.py +++ b/tests/test_imscp.py @@ -2,38 +2,92 @@ import os import tempfile import zipfile -from contextlib import contextmanager import pytest +from helpers import create_zip_with_manifest +from ricecooker.utils.imscp import _SAFE_PARSER from ricecooker.utils.imscp import flatten_single_child_topics from ricecooker.utils.imscp import has_imscp_manifest +from ricecooker.utils.imscp import has_qti_items +from ricecooker.utils.imscp import has_qti_resources +from ricecooker.utils.imscp import has_webcontent_items +from ricecooker.utils.imscp import is_qti_resource +from ricecooker.utils.imscp import ManifestParseError from ricecooker.utils.imscp import parse_imscp_manifest from ricecooker.utils.SCORM_metadata import metadata_dict_to_content_node_fields -BASE_PATH = os.path.dirname(os.path.abspath(__file__)) -IMS_XML_DIR = os.path.join(BASE_PATH, "testcontent", "samples", "ims_xml") +# --- Safe parser tests --- -@contextmanager -def create_zip_with_manifest(manifest_filename, *additional_files): - """Create a temp zip with a manifest and optional additional files.""" +def test_safe_parser_disables_entity_resolution(): + """_SAFE_PARSER does not resolve XML entities.""" + from lxml import etree + import io + + # XML with an internal entity reference + xml_bytes = b""" + ]> + &xxe;""" + tree = etree.parse(io.BytesIO(xml_bytes), _SAFE_PARSER) + # With resolve_entities=False the entity reference is NOT resolved into text + assert tree.getroot().text is None + + +# --- Null guard / error handling tests --- + + +def test_parse_manifest_without_resources_element(): + """Manifest with but no should not crash.""" + manifest_content = b""" + + <string>No Resources</string> + + + Org + Item 1 + + +""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("imsmanifest.xml", manifest_content) + result = parse_imscp_manifest(temp_zip_path) + # Should not crash; the item with identifierref won't resolve + assert result["metadata"]["title"] == "No Resources" + assert len(result["organizations"]) == 1 + finally: + os.remove(temp_zip_path) + + +def test_resolve_metadata_with_missing_external_file(): + """ pointing to nonexistent file in zip raises ManifestParseError.""" + manifest_content = b""" + + + nonexistent_metadata.xml + + + + Test + + + +""" temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) temp_zip_path = temp_zip.name temp_zip.close() try: - manifest_file = os.path.join(IMS_XML_DIR, manifest_filename) with zipfile.ZipFile(temp_zip_path, "w") as zf: - zf.write(manifest_file, "imsmanifest.xml") - for additional_file in additional_files: - zf.write(os.path.join(IMS_XML_DIR, additional_file), additional_file) - yield temp_zip_path + zf.writestr("imsmanifest.xml", manifest_content) + with pytest.raises(ManifestParseError, match="nonexistent_metadata.xml"): + parse_imscp_manifest(temp_zip_path) finally: - try: - os.remove(temp_zip_path) - except (FileNotFoundError, OSError): - pass + os.remove(temp_zip_path) # --- Manifest parsing tests --- @@ -192,6 +246,48 @@ def test_flatten_deep_single_chain(): assert result["children"][0]["title"] == "Leaf" +def test_flatten_preserves_parent_title_when_child_has_none(): + """When child has no title, parent's title is preserved.""" + item = { + "title": "Parent Title", + "metadata": {}, + "children": [ + { + "metadata": {"description": "child desc"}, + "children": [ + {"title": "Leaf1", "metadata": {}, "href": "page1.html"}, + {"title": "Leaf2", "metadata": {}, "href": "page2.html"}, + ], + } + ], + } + result = flatten_single_child_topics(item) + + # Child has no title, so parent's title should be used + assert result["title"] == "Parent Title" + assert len(result["children"]) == 2 + + +def test_flatten_replaces_empty_string_title_with_parent(): + """When child has an empty-string title, parent's title is used.""" + item = { + "title": "Parent Title", + "metadata": {}, + "children": [ + { + "title": "", + "metadata": {}, + "children": [ + {"title": "Leaf", "metadata": {}, "href": "page.html"}, + ], + } + ], + } + result = flatten_single_child_topics(item) + # Empty string is not a meaningful title — parent's title should be used + assert result["title"] == "Parent Title" + + def test_flatten_preserves_leaf(): """Leaf items (no children) are unchanged.""" item = {"title": "Leaf", "metadata": {}, "href": "page.html"} @@ -362,9 +458,7 @@ def test_scorm_resource_types_single_string(): from ricecooker.utils.SCORM_metadata import map_scorm_to_educator_resource_types from le_utils.constants.labels import resource_type - result = map_scorm_to_educator_resource_types( - {"learningResourceType": "exercise"} - ) + result = map_scorm_to_educator_resource_types({"learningResourceType": "exercise"}) assert result == [resource_type.EXERCISE] @@ -409,3 +503,183 @@ def test_metadata_dict_to_content_node_fields_lifecycle(): } result = metadata_dict_to_content_node_fields(metadata) assert result["provider"] == "Example Organization" + + +# --- QTI resource detection tests --- + + +@pytest.mark.parametrize( + "resource_type,expected", + [ + ("imsqti_item_xmlv2p1", True), + ("imsqti_test_xmlv2p1", True), + ("imsqti_assessment_xmlv2p1", True), + ("webcontent", False), + ("", False), + (None, False), + ("associatedcontent/imscc_xmlv1p1/learning-application-resource", False), + ], +) +def test_is_qti_resource(resource_type, expected): + assert is_qti_resource(resource_type) is expected + + +def test_collect_resources_qti(): + """QTI resources get files populated just like webcontent.""" + with create_zip_with_manifest("qti_manifest.xml") as zip_path: + result = parse_imscp_manifest(zip_path) + + org = result["organizations"][0] + # Both items reference QTI resources and should have files + for child in org["children"]: + assert "files" in child, f"QTI item '{child['title']}' should have files" + assert len(child["files"]) > 0 + + +def test_collect_resources_mixed(): + """Mixed package: both webcontent and QTI items get files.""" + with create_zip_with_manifest("mixed_manifest.xml") as zip_path: + result = parse_imscp_manifest(zip_path) + + org = result["organizations"][0] + lessons = next(c for c in org["children"] if c["title"] == "Lessons") + quizzes = next(c for c in org["children"] if c["title"] == "Quizzes") + + # Webcontent leaf + lesson1 = lessons["children"][0] + assert "files" in lesson1 + assert "lesson1.html" in lesson1["files"] + + # QTI leaves + for quiz in quizzes["children"]: + assert "files" in quiz, f"QTI item '{quiz['title']}' should have files" + + +def test_has_qti_resources_true(): + """Zip with QTI resource types returns True.""" + with create_zip_with_manifest("qti_manifest.xml") as zip_path: + assert has_qti_resources(zip_path) is True + + +def test_has_qti_resources_false(): + """Zip with only webcontent returns False.""" + with create_zip_with_manifest("simple_manifest.xml") as zip_path: + assert has_qti_resources(zip_path) is False + + +def test_has_qti_resources_mixed(): + """Mixed package with both webcontent and QTI returns True.""" + with create_zip_with_manifest("mixed_manifest.xml") as zip_path: + assert has_qti_resources(zip_path) is True + + +def test_has_qti_resources_nonexistent_file(): + assert has_qti_resources("/nonexistent/path.zip") is False + + +def test_has_qti_resources_bad_zip(): + """Garbage bytes file: has_qti_resources returns False.""" + temp = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp.write(b"this is not a zip file at all") + temp.close() + try: + assert has_qti_resources(temp.name) is False + finally: + os.remove(temp.name) + + +def test_has_qti_resources_invalid_manifest(): + """Zip with malformed XML as imsmanifest.xml: has_qti_resources returns False.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("imsmanifest.xml", "<<>>") + assert has_qti_resources(temp_zip_path) is False + finally: + os.remove(temp_zip_path) + + +def test_has_qti_resources_manifest_missing_from_zip(): + """Zip without imsmanifest.xml: has_qti_resources returns False.""" + temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) + temp_zip_path = temp_zip.name + temp_zip.close() + try: + with zipfile.ZipFile(temp_zip_path, "w") as zf: + zf.writestr("index.html", "") + assert has_qti_resources(temp_zip_path) is False + finally: + os.remove(temp_zip_path) + + +def test_collect_resources_qti_preserves_type(): + """QTI items have their resource type preserved in the item dict.""" + with create_zip_with_manifest("qti_manifest.xml") as zip_path: + result = parse_imscp_manifest(zip_path) + + org = result["organizations"][0] + for child in org["children"]: + assert child.get("type") == "imsqti_item_xmlv2p1" + + +def test_collect_resources_mixed_preserves_types(): + """Mixed items preserve their respective resource types.""" + with create_zip_with_manifest("mixed_manifest.xml") as zip_path: + result = parse_imscp_manifest(zip_path) + + org = result["organizations"][0] + lessons = next(c for c in org["children"] if c["title"] == "Lessons") + quizzes = next(c for c in org["children"] if c["title"] == "Quizzes") + + lesson1 = lessons["children"][0] + assert lesson1.get("type") == "webcontent" + + quiz1 = quizzes["children"][0] + assert is_qti_resource(quiz1.get("type", "")) + + +# --- has_qti_items / has_webcontent_items tests (parsed manifest dict) --- + + +def test_has_qti_items_pure_qti(): + """has_qti_items returns True for a pure QTI manifest.""" + with create_zip_with_manifest("qti_manifest.xml") as zip_path: + parsed = parse_imscp_manifest(zip_path) + assert has_qti_items(parsed) is True + + +def test_has_qti_items_pure_imscp(): + """has_qti_items returns False for a pure IMSCP (webcontent) manifest.""" + with create_zip_with_manifest("simple_manifest.xml") as zip_path: + parsed = parse_imscp_manifest(zip_path) + assert has_qti_items(parsed) is False + + +def test_has_qti_items_mixed(): + """has_qti_items returns True for a mixed manifest.""" + with create_zip_with_manifest("mixed_manifest.xml") as zip_path: + parsed = parse_imscp_manifest(zip_path) + assert has_qti_items(parsed) is True + + +def test_has_webcontent_items_pure_imscp(): + """has_webcontent_items returns True for a pure IMSCP (webcontent) manifest.""" + with create_zip_with_manifest("simple_manifest.xml") as zip_path: + parsed = parse_imscp_manifest(zip_path) + assert has_webcontent_items(parsed) is True + + +def test_has_webcontent_items_pure_qti(): + """has_webcontent_items returns False for a pure QTI manifest.""" + with create_zip_with_manifest("qti_manifest.xml") as zip_path: + parsed = parse_imscp_manifest(zip_path) + assert has_webcontent_items(parsed) is False + + +def test_has_webcontent_items_mixed(): + """has_webcontent_items returns True for a mixed manifest.""" + with create_zip_with_manifest("mixed_manifest.xml") as zip_path: + parsed = parse_imscp_manifest(zip_path) + assert has_webcontent_items(parsed) is True diff --git a/tests/test_imscp_pipeline.py b/tests/test_imscp_pipeline.py index 00245138..b58ce4a8 100644 --- a/tests/test_imscp_pipeline.py +++ b/tests/test_imscp_pipeline.py @@ -3,8 +3,10 @@ import tempfile import zipfile from unittest.mock import MagicMock +from unittest.mock import patch import pytest +from helpers import create_zip_with_manifest from le_utils.constants import content_kinds from le_utils.constants import format_presets from le_utils.constants import licenses @@ -19,8 +21,7 @@ from ricecooker.utils.pipeline.extract_metadata import IMSCPMetadataExtractor -BASE_PATH = os.path.dirname(os.path.abspath(__file__)) -IMS_XML_DIR = os.path.join(BASE_PATH, "testcontent", "samples", "ims_xml") +_INDEX_HTML = {"index.html": "Content"} # --- Fixtures --- @@ -29,46 +30,22 @@ @pytest.fixture def simple_imscp_zip(): """Create a simple IMSCP zip with manifest.""" - temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) - temp_zip_path = temp_zip.name - temp_zip.close() - manifest_file = os.path.join(IMS_XML_DIR, "simple_manifest.xml") - with zipfile.ZipFile(temp_zip_path, "w") as zf: - zf.write(manifest_file, "imsmanifest.xml") - # Add index.html so it also qualifies as HTML5 if needed - zf.writestr("index.html", "IMSCP Content") - yield temp_zip_path - try: - os.remove(temp_zip_path) - except (FileNotFoundError, OSError): - pass + with create_zip_with_manifest( + "simple_manifest.xml", extra_entries=_INDEX_HTML + ) as path: + yield path @pytest.fixture def complex_imscp_zip(): """Create a complex IMSCP zip with external metadata.""" - temp_zip = tempfile.NamedTemporaryFile(suffix=".zip", delete=False) - temp_zip_path = temp_zip.name - temp_zip.close() - with zipfile.ZipFile(temp_zip_path, "w") as zf: - zf.write( - os.path.join(IMS_XML_DIR, "complete_manifest_with_external_metadata.xml"), - "imsmanifest.xml", - ) - zf.write( - os.path.join(IMS_XML_DIR, "metadata_hummingbirds_course.xml"), - "metadata_hummingbirds_course.xml", - ) - zf.write( - os.path.join(IMS_XML_DIR, "metadata_hummingbirds_organization.xml"), - "metadata_hummingbirds_organization.xml", - ) - zf.writestr("index.html", "Hummingbirds") - yield temp_zip_path - try: - os.remove(temp_zip_path) - except (FileNotFoundError, OSError): - pass + with create_zip_with_manifest( + "complete_manifest_with_external_metadata.xml", + "metadata_hummingbirds_course.xml", + "metadata_hummingbirds_organization.xml", + extra_entries=_INDEX_HTML, + ) as path: + yield path @pytest.fixture @@ -135,8 +112,11 @@ def test_imscp_metadata_extractor_should_not_handle_regular_html5(regular_html5_ def test_imscp_metadata_extractor_preset(simple_imscp_zip): + from ricecooker.utils.imscp import parse_imscp_manifest + extractor = IMSCPMetadataExtractor() - assert extractor.infer_preset(simple_imscp_zip) == format_presets.IMSCP_ZIP + manifest = parse_imscp_manifest(simple_imscp_zip) + assert extractor._infer_preset_from_manifest(manifest) == format_presets.IMSCP_ZIP def test_imscp_metadata_extractor_simple_manifest(simple_imscp_zip): @@ -217,6 +197,21 @@ def test_content_node_metadata_with_children_serializes(): assert result["content_node_metadata"]["children"][0]["title"] == "Child" +def test_content_node_metadata_from_dict_logs_unknown_keys(): + """Unknown keys in dict are dropped with a debug log message.""" + from ricecooker.utils.pipeline.context import _content_node_metadata_from_dict + + with patch("ricecooker.utils.pipeline.context.logger") as mock_logger: + result = _content_node_metadata_from_dict( + {"title": "Test", "bogus_key": "value", "another_unknown": 42} + ) + assert result.title == "Test" + mock_logger.debug.assert_called_once() + logged_keys = mock_logger.debug.call_args[0][1] + assert "bogus_key" in logged_keys + assert "another_unknown" in logged_keys + + def test_file_metadata_merge_with_children(): """Merge preserves children structure.""" child = ContentNodeMetadata(title="Child", kind="html5") @@ -256,25 +251,25 @@ def test_content_node_builds_children_from_nested_metadata(): node.add_file(mock_file) children_data = [ - { - "title": "Topic 1", - "source_id": "topic1", - "kind": "topic", - "children": [ - { - "title": "Leaf 1", - "source_id": "leaf1", - "kind": "html5", - "extra_fields": {"options": {"entry": "page1.html"}}, - }, + ContentNodeMetadata( + title="Topic 1", + source_id="topic1", + kind="topic", + children=[ + ContentNodeMetadata( + title="Leaf 1", + source_id="leaf1", + kind="html5", + extra_fields={"options": {"entry": "page1.html"}}, + ), ], - }, - { - "title": "Leaf 2", - "source_id": "leaf2", - "kind": "html5", - "extra_fields": {"options": {"entry": "page2.html"}}, - }, + ), + ContentNodeMetadata( + title="Leaf 2", + source_id="leaf2", + kind="html5", + extra_fields={"options": {"entry": "page2.html"}}, + ), ] node._build_children_from_metadata(children_data) @@ -330,12 +325,12 @@ def test_content_node_children_have_entry_points(): node.add_file(mock_file) children_data = [ - { - "title": "Page A", - "source_id": "page_a", - "kind": "html5", - "extra_fields": {"options": {"entry": "content/page_a.html"}}, - }, + ContentNodeMetadata( + title="Page A", + source_id="page_a", + kind="html5", + extra_fields={"options": {"entry": "content/page_a.html"}}, + ), ] node._build_children_from_metadata(children_data) @@ -347,27 +342,27 @@ def test_content_node_children_have_entry_points(): def test_content_node_children_share_files(): - """Leaf children with file_presets get parent files; topic children do not.""" + """Leaf children with file_preset get parent files; topic children do not.""" license = _make_license() node = ContentNode("root_id", "Root", license) mock_file = _make_mock_file() node.add_file(mock_file) children_data = [ - { - "title": "Topic", - "source_id": "topic1", - "kind": "topic", - "children": [ - { - "title": "Leaf", - "source_id": "leaf1", - "kind": "html5", - "file_presets": [format_presets.IMSCP_ZIP], - "extra_fields": {"options": {"entry": "index.html"}}, - }, + ContentNodeMetadata( + title="Topic", + source_id="topic1", + kind="topic", + children=[ + ContentNodeMetadata( + title="Leaf", + source_id="leaf1", + kind="html5", + file_preset=format_presets.IMSCP_ZIP, + extra_fields={"options": {"entry": "index.html"}}, + ), ], - }, + ), ] node._build_children_from_metadata(children_data) @@ -376,7 +371,7 @@ def test_content_node_children_share_files(): leaf_child = topic_child.children[0] # Topic children should not get files assert len(topic_child.files) == 0 - # Leaf children with matching file_presets should get files + # Leaf children with file_preset get copies of parent files assert len(leaf_child.files) > 0 @@ -388,14 +383,14 @@ def test_content_node_children_metadata_fields_applied(): node.add_file(mock_file) children_data = [ - { - "title": "Leaf", - "source_id": "leaf1", - "kind": "html5", - "description": "A test description", - "tags": ["tag1", "tag2"], - "extra_fields": {"options": {"entry": "index.html"}}, - }, + ContentNodeMetadata( + title="Leaf", + source_id="leaf1", + kind="html5", + description="A test description", + tags=["tag1", "tag2"], + extra_fields={"options": {"entry": "index.html"}}, + ), ] node._build_children_from_metadata(children_data) @@ -467,26 +462,26 @@ def test_content_node_recursive_children_structure(): node.add_file(mock_file) children_data = [ - { - "title": "Level 1", - "source_id": "l1", - "kind": "topic", - "children": [ - { - "title": "Level 2", - "source_id": "l2", - "kind": "topic", - "children": [ - { - "title": "Deep Leaf", - "source_id": "deep", - "kind": "html5", - "extra_fields": {"options": {"entry": "deep.html"}}, - }, + ContentNodeMetadata( + title="Level 1", + source_id="l1", + kind="topic", + children=[ + ContentNodeMetadata( + title="Level 2", + source_id="l2", + kind="topic", + children=[ + ContentNodeMetadata( + title="Deep Leaf", + source_id="deep", + kind="html5", + extra_fields={"options": {"entry": "deep.html"}}, + ), ], - }, + ), ], - }, + ), ] node._build_children_from_metadata(children_data) @@ -513,19 +508,19 @@ def test_topic_children_have_no_files(): node.add_file(mock_file) children_data = [ - { - "title": "Topic", - "source_id": "topic1", - "kind": "topic", - "children": [ - { - "title": "Leaf", - "source_id": "leaf1", - "kind": "html5", - "extra_fields": {"options": {"entry": "index.html"}}, - }, + ContentNodeMetadata( + title="Topic", + source_id="topic1", + kind="topic", + children=[ + ContentNodeMetadata( + title="Leaf", + source_id="leaf1", + kind="html5", + extra_fields={"options": {"entry": "index.html"}}, + ), ], - }, + ), ] node._build_children_from_metadata(children_data) @@ -535,21 +530,21 @@ def test_topic_children_have_no_files(): assert len(topic_child.files) == 0 -def test_leaf_children_get_files_by_preset(): - """Leaf children with file_presets get only files matching their presets.""" +def test_leaf_children_get_files_with_preset(): + """Leaf children with file_preset get copies of parent files with that preset.""" license = _make_license() node = ContentNode("root_id", "Root", license) imscp_file = _make_mock_file(preset=format_presets.IMSCP_ZIP, filename="test.zip") node.add_file(imscp_file) children_data = [ - { - "title": "Leaf", - "source_id": "leaf1", - "kind": "html5", - "file_presets": [format_presets.IMSCP_ZIP], - "extra_fields": {"options": {"entry": "index.html"}}, - }, + ContentNodeMetadata( + title="Leaf", + source_id="leaf1", + kind="html5", + file_preset=format_presets.IMSCP_ZIP, + extra_fields={"options": {"entry": "index.html"}}, + ), ] node._build_children_from_metadata(children_data) @@ -557,22 +552,24 @@ def test_leaf_children_get_files_by_preset(): child = node.children[0] assert len(child.files) == 1 assert child.files[0].get_preset() == format_presets.IMSCP_ZIP + # Child gets a copy, not the same object + assert child.files[0] is not imscp_file -def test_children_without_file_presets_get_no_files(): - """Children without file_presets get no files.""" +def test_children_without_file_preset_get_no_files(): + """Children without file_preset get no files.""" license = _make_license() node = ContentNode("root_id", "Root", license) mock_file = _make_mock_file() node.add_file(mock_file) children_data = [ - { - "title": "Leaf", - "source_id": "leaf1", - "kind": "html5", - "extra_fields": {"options": {"entry": "index.html"}}, - }, + ContentNodeMetadata( + title="Leaf", + source_id="leaf1", + kind="html5", + extra_fields={"options": {"entry": "index.html"}}, + ), ] node._build_children_from_metadata(children_data) @@ -592,19 +589,19 @@ def test_license_propagated_to_nested_content_nodes(): node.add_file(mock_file) children_data = [ - { - "title": "Topic 1", - "source_id": "topic1", - "kind": "topic", - "children": [ - { - "title": "Leaf 1", - "source_id": "leaf1", - "kind": "html5", - "extra_fields": {"options": {"entry": "page1.html"}}, - }, + ContentNodeMetadata( + title="Topic 1", + source_id="topic1", + kind="topic", + children=[ + ContentNodeMetadata( + title="Leaf 1", + source_id="leaf1", + kind="html5", + extra_fields={"options": {"entry": "page1.html"}}, + ), ], - }, + ), ] node._build_children_from_metadata(children_data) @@ -623,26 +620,26 @@ def test_license_propagated_through_deep_nesting(): node.add_file(mock_file) children_data = [ - { - "title": "Level 1", - "source_id": "l1", - "kind": "topic", - "children": [ - { - "title": "Level 2", - "source_id": "l2", - "kind": "topic", - "children": [ - { - "title": "Deep Leaf", - "source_id": "deep", - "kind": "html5", - "extra_fields": {"options": {"entry": "deep.html"}}, - }, + ContentNodeMetadata( + title="Level 1", + source_id="l1", + kind="topic", + children=[ + ContentNodeMetadata( + title="Level 2", + source_id="l2", + kind="topic", + children=[ + ContentNodeMetadata( + title="Deep Leaf", + source_id="deep", + kind="html5", + extra_fields={"options": {"entry": "deep.html"}}, + ), ], - }, + ), ], - }, + ), ] node._build_children_from_metadata(children_data) @@ -666,19 +663,19 @@ def test_process_files_collects_grandchild_filenames(): # Build Topic -> Leaf structure children_data = [ - { - "title": "Topic 1", - "source_id": "topic1", - "kind": "topic", - "children": [ - { - "title": "Leaf 1", - "source_id": "leaf1", - "kind": "html5", - "extra_fields": {"options": {"entry": "page1.html"}}, - }, + ContentNodeMetadata( + title="Topic 1", + source_id="topic1", + kind="topic", + children=[ + ContentNodeMetadata( + title="Leaf 1", + source_id="leaf1", + kind="html5", + extra_fields={"options": {"entry": "page1.html"}}, + ), ], - }, + ), ] node._build_children_from_metadata(children_data) @@ -689,6 +686,477 @@ def test_process_files_collects_grandchild_filenames(): # Simulate what process_files does for dynamic children: # The current flat loop only checks direct children, missing grandchildren - filenames = node._collect_dynamic_filenames(node) + filenames = node._validate_and_collect_dynamic_filenames(node) assert "abc123.zip" in filenames + + +def test_file_init_keys_match_file_signature(): + """Guard: _FILE_INIT_KEYS must stay in sync with File.__init__ parameters.""" + import inspect + + expected = set(inspect.signature(File.__init__).parameters) - {"self"} + assert ContentNode._FILE_INIT_KEYS == expected, ( + f"_FILE_INIT_KEYS is out of sync with File.__init__. " + f"Missing: {expected - ContentNode._FILE_INIT_KEYS}, " + f"Extra: {ContentNode._FILE_INIT_KEYS - expected}" + ) + + +# --- QTI zip fixtures --- + + +@pytest.fixture +def qti_imscp_zip(): + """Create an IMSCP zip with pure QTI resources.""" + with create_zip_with_manifest( + "qti_manifest.xml", extra_entries=_INDEX_HTML + ) as path: + yield path + + +@pytest.fixture +def mixed_imscp_zip(): + """Create an IMSCP zip with both webcontent and QTI resources.""" + with create_zip_with_manifest( + "mixed_manifest.xml", extra_entries=_INDEX_HTML + ) as path: + yield path + + +# --- QTI metadata extraction tests (Step 3) --- + + +def test_metadata_extractor_sets_kind_on_content_node_metadata(simple_imscp_zip): + """MetadataExtractor.handle_file mutates ContentNodeMetadata.kind in place. + + This characterization test guards against accidentally freezing ContentNodeMetadata. + """ + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(simple_imscp_zip) + assert result.content_node_metadata.kind == "topic" + + +def test_pure_qti_metadata(qti_imscp_zip): + """All leaves in a pure QTI package have kind=exercise and QTI_ZIP file_preset.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(qti_imscp_zip) + + metadata = result.content_node_metadata + assert metadata.kind == "topic" + org = metadata.children[0] + + for child in org.children: + assert child.kind == content_kinds.EXERCISE + assert child.file_preset == format_presets.QTI_ZIP + + +def test_mixed_metadata(mixed_imscp_zip): + """Mixed package: webcontent leaves get html5/IMSCP_ZIP, QTI leaves get exercise/QTI_ZIP.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(mixed_imscp_zip) + + metadata = result.content_node_metadata + org = metadata.children[0] + + lessons = next(c for c in org.children if c.title == "Lessons") + quizzes = next(c for c in org.children if c.title == "Quizzes") + + # Webcontent leaf + lesson1 = lessons.children[0] + assert lesson1.kind == content_kinds.HTML5 + assert lesson1.file_preset == format_presets.IMSCP_ZIP + + # QTI leaves + for quiz in quizzes.children: + assert quiz.kind == content_kinds.EXERCISE + assert quiz.file_preset == format_presets.QTI_ZIP + + +def test_pure_imscp_metadata_unchanged(simple_imscp_zip): + """Regression: pure webcontent packages still produce html5/IMSCP_ZIP leaves.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(simple_imscp_zip) + + metadata = result.content_node_metadata + org = metadata.children[0] + + # All leaves should be html5/IMSCP_ZIP + def check_leaves(node): + if node.children: + for child in node.children: + check_leaves(child) + else: + assert node.kind == content_kinds.HTML5 + assert node.file_preset == format_presets.IMSCP_ZIP + + check_leaves(org) + + +def test_qti_entry_point(qti_imscp_zip): + """QTI leaf nodes have extra_fields.options.entry pointing to QTI XML.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(qti_imscp_zip) + + org = result.content_node_metadata.children[0] + q1 = org.children[0] + assert q1.extra_fields["options"]["entry"] == "question1.xml" + q2 = org.children[1] + assert q2.extra_fields["options"]["entry"] == "question2.xml" + + +def test_mixed_package_single_file_preset(mixed_imscp_zip): + """Mixed package still produces a single FileMetadata with IMSCP_ZIP preset.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(mixed_imscp_zip) + + # One preset per file — children override preset via copy + assert result.preset == format_presets.IMSCP_ZIP + + +def test_pure_qti_single_file_registration(qti_imscp_zip): + """Pure QTI package produces a single file with QTI_ZIP preset.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(qti_imscp_zip) + + assert result.preset == format_presets.QTI_ZIP + + +def test_pure_imscp_single_file_preset(simple_imscp_zip): + """Pure IMSCP package produces a single file with IMSCP_ZIP preset.""" + extractor = IMSCPMetadataExtractor() + result = extractor.handle_file(simple_imscp_zip) + + assert result.preset == format_presets.IMSCP_ZIP + + +def test_handle_file_opens_zip_at_most_twice(mixed_imscp_zip): + """handle_file should not redundantly open the zip file.""" + open_count = 0 + original_init = zipfile.ZipFile.__init__ + + def counting_init(self, *args, **kwargs): + nonlocal open_count + open_count += 1 + return original_init(self, *args, **kwargs) + + extractor = IMSCPMetadataExtractor() + with patch.object(zipfile.ZipFile, "__init__", counting_init): + extractor.handle_file(mixed_imscp_zip) + + assert open_count <= 2, f"Zip opened {open_count} times, expected at most 2" + + +# --- Node building tests for QTI (Step 5) --- + + +def test_qti_leaf_nodes_are_exercise_kind(): + """Dynamically created leaf nodes from QTI metadata have kind='exercise'.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + imscp_file = _make_mock_file(preset=format_presets.QTI_ZIP, filename="test.zip") + node.add_file(imscp_file) + + children_data = [ + ContentNodeMetadata( + title="Question 1", + source_id="q1", + kind=content_kinds.EXERCISE, + file_preset=format_presets.QTI_ZIP, + extra_fields={"options": {"entry": "question1.xml"}}, + ), + ContentNodeMetadata( + title="Question 2", + source_id="q2", + kind=content_kinds.EXERCISE, + file_preset=format_presets.QTI_ZIP, + extra_fields={"options": {"entry": "question2.xml"}}, + ), + ] + + node._build_children_from_metadata(children_data) + + for child in node.children: + assert isinstance(child, ContentNode) + assert child.kind == content_kinds.EXERCISE + assert len(child.files) == 1 + assert child.files[0].get_preset() == format_presets.QTI_ZIP + + +def test_mixed_tree_structure(): + """Mixed tree: one parent file, children get copies with correct preset.""" + license = _make_license() + node = ContentNode("root_id", "Root", license) + # Only ONE file — children copy it with the right preset + imscp_file = _make_mock_file(preset=format_presets.IMSCP_ZIP, filename="test.zip") + node.add_file(imscp_file) + + children_data = [ + ContentNodeMetadata( + title="Lessons", + source_id="lessons", + kind="topic", + children=[ + ContentNodeMetadata( + title="Lesson 1", + source_id="l1", + kind=content_kinds.HTML5, + file_preset=format_presets.IMSCP_ZIP, + extra_fields={"options": {"entry": "lesson1.html"}}, + ), + ], + ), + ContentNodeMetadata( + title="Quizzes", + source_id="quizzes", + kind="topic", + children=[ + ContentNodeMetadata( + title="Quiz 1", + source_id="quiz1", + kind=content_kinds.EXERCISE, + file_preset=format_presets.QTI_ZIP, + extra_fields={"options": {"entry": "quiz1.xml"}}, + ), + ], + ), + ] + + node._build_children_from_metadata(children_data) + + lessons = node.children[0] + quizzes = node.children[1] + + lesson1 = lessons.children[0] + assert lesson1.kind == content_kinds.HTML5 + assert lesson1.files[0].get_preset() == format_presets.IMSCP_ZIP + + quiz1 = quizzes.children[0] + assert quiz1.kind == content_kinds.EXERCISE + assert quiz1.files[0].get_preset() == format_presets.QTI_ZIP + + +def test_exercise_kind_in_activity_map(): + """Exercise kind maps to PRACTICE learning activity.""" + from ricecooker.classes.nodes import kind_activity_map + from le_utils.constants.labels import learning_activities + + assert content_kinds.EXERCISE in kind_activity_map + assert kind_activity_map[content_kinds.EXERCISE] == learning_activities.PRACTICE + + +def test_process_uri_logs_dropped_keys(): + """_process_uri logs debug message when FileMetadata keys are filtered out.""" + license = _make_license() + + root_meta = ContentNodeMetadata(title="Leaf Content", kind="html5") + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + license="CC BY", + license_description="Creative Commons Attribution", + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + + with patch("ricecooker.classes.nodes.config") as mock_config: + mock_config.UPDATE = False + mock_logger = MagicMock() + mock_config.LOGGER = mock_logger + node._process_uri() + + mock_logger.debug.assert_called() + # The log message should mention the dropped keys + assert "license" in str(mock_logger.debug.call_args) + # Title from pipeline metadata should have been applied + assert node.title == "Leaf Content" + + +def test_process_uri_handles_license_in_file_metadata(): + """_process_uri doesn't crash when FileMetadata has license fields set.""" + license = _make_license() + + root_meta = ContentNodeMetadata(title="Leaf Content", kind="html5") + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + license="CC BY", + license_description="Creative Commons Attribution", + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + assert len(node.files) == 1 + assert node.files[0].get_preset() == format_presets.IMSCP_ZIP + + +def test_process_uri_applies_language_from_metadata(): + """Pipeline metadata language should be applied to the node.""" + license = _make_license() + + root_meta = ContentNodeMetadata(title="Leaf Content", kind="html5", language="fr") + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + assert node.language == "fr" + + +def test_process_uri_does_not_overwrite_source_id(): + """Pipeline metadata source_id must not overwrite the node's constructor value.""" + license = _make_license() + + root_meta = ContentNodeMetadata( + title="New Title", kind="html5", source_id="pipeline_source_id" + ) + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "original_source_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + assert node.source_id == "original_source_id" + + +def test_process_uri_does_not_overwrite_license(): + """Pipeline metadata license must not overwrite the node's constructor value.""" + license = _make_license() + + root_meta = ContentNodeMetadata(title="New Title", kind="html5", license="CC BY-SA") + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + assert node.license == license + + +def test_metadata_apply_fields_is_class_constant(): + """_METADATA_APPLY_FIELDS should be accessible as a class-level constant.""" + assert hasattr(ContentNode, "_METADATA_APPLY_FIELDS") + assert isinstance(ContentNode._METADATA_APPLY_FIELDS, tuple) + + +def test_process_uri_does_not_apply_copyright_holder(): + """Pipeline metadata copyright_holder must not be applied to the node. + + copyright_holder is an identity/legal field like source_id and license, + so it is intentionally excluded from _METADATA_APPLY_FIELDS. + """ + license = _make_license() + + root_meta = ContentNodeMetadata( + title="New Title", kind="html5", copyright_holder="Pipeline Holder" + ) + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + # copyright_holder is not in _METADATA_APPLY_FIELDS, so it should + # never be set as a direct attribute on the node + assert not hasattr(node, "copyright_holder") + # The license object retains its original copyright_holder + assert node.license.copyright_holder == "Test Author" + + +def test_process_uri_children_get_correct_preset_via_copy(): + """_process_uri with mixed metadata: children get file copies with correct presets.""" + license = _make_license() + + child_meta = ContentNodeMetadata( + title="Lesson 1", + kind=content_kinds.HTML5, + source_id="l1", + file_preset=format_presets.IMSCP_ZIP, + extra_fields={"options": {"entry": "lesson1.html"}}, + ) + quiz_meta = ContentNodeMetadata( + title="Quiz 1", + kind=content_kinds.EXERCISE, + source_id="q1", + file_preset=format_presets.QTI_ZIP, + extra_fields={"options": {"entry": "quiz1.xml"}}, + ) + root_meta = ContentNodeMetadata( + title="Mixed", kind="topic", children=[child_meta, quiz_meta] + ) + file_metadata = FileMetadata( + preset=format_presets.IMSCP_ZIP, + content_node_metadata=root_meta, + path="test.zip", + ) + + mock_pipeline = MagicMock() + mock_pipeline.should_handle.return_value = True + mock_pipeline.execute.return_value = [file_metadata] + + node = ContentNode( + "root_id", "Root", license, uri="test.zip", pipeline=mock_pipeline + ) + node._process_uri() + + # Parent has single file + assert len(node.files) == 1 + assert node.files[0].get_preset() == format_presets.IMSCP_ZIP + + # Children get copies with the right presets + lesson = node.children[0] + quiz = node.children[1] + assert lesson.files[0].get_preset() == format_presets.IMSCP_ZIP + assert quiz.files[0].get_preset() == format_presets.QTI_ZIP diff --git a/tests/testcontent/samples/ims_xml/mixed_manifest.xml b/tests/testcontent/samples/ims_xml/mixed_manifest.xml new file mode 100644 index 00000000..3bf23fc5 --- /dev/null +++ b/tests/testcontent/samples/ims_xml/mixed_manifest.xml @@ -0,0 +1,48 @@ + + + + + + Mixed Content Package + + en + + Package with both webcontent and QTI resources + + + + + + + Course Content + + Lessons + + Lesson 1 + + + + Quizzes + + Quiz 1 + + + Quiz 2 + + + + + + + + + + + + + + + + + + diff --git a/tests/testcontent/samples/ims_xml/qti_manifest.xml b/tests/testcontent/samples/ims_xml/qti_manifest.xml new file mode 100644 index 00000000..7f06eaa8 --- /dev/null +++ b/tests/testcontent/samples/ims_xml/qti_manifest.xml @@ -0,0 +1,34 @@ + + + + + + QTI Assessment + + en + + A pure QTI assessment package + + + + + + + Assessments + + Question 1 + + + Question 2 + + + + + + + + + + + +