feat: handle file content logic when creating next component version

.coveragerc

@@ -6,7 +6,8 @@ source =
 	openedx_tagging
 omit =
     test_settings
-    *migrations*
+    **/migrations/*
     *admin.py
     *static*
     *templates*
+    **/tests/**

openedx_learning/apps/authoring/components/api.py

@@ -12,6 +12,7 @@
 """
 from __future__ import annotations
 
+import mimetypes
 from datetime import datetime, timezone
 from enum import StrEnum, auto
 from logging import getLogger
@@ -129,7 +130,7 @@ def create_component_version(
 def create_next_component_version(
     component_pk: int,
     /,
-    content_to_replace: dict[str, int | None],
+    content_to_replace: dict[str, int | None | bytes],
     created: datetime,
     title: str | None = None,
     created_by: int | None = None,
@@ -143,8 +144,8 @@ def create_next_component_version(
     API, since ``content_to_replace`` needs Content IDs for the values.
 
     The ``content_to_replace`` dict is a mapping of strings representing the
-    local path/key for a file, to ``Content.id`` values. Using a `None` for
-    a value in this dict means to delete that key in the next version.
+    local path/key for a file, to ``Content.id`` or content bytes values. Using
+    `None` for a value in this dict means to delete that key in the next version.
 
     It is okay to mark entries for deletion that don't exist. For instance, if a
     version has ``a.txt`` and ``b.txt``, sending a ``content_to_replace`` value
@@ -186,11 +187,27 @@ def create_next_component_version(
             component_id=component_pk,
         )
         # First copy the new stuff over...
-        for key, content_pk in content_to_replace.items():
+        for key, content_pk_or_bytes in content_to_replace.items():
             # If the content_pk is None, it means we want to remove the
             # content represented by our key from the next version. Otherwise,
             # we add our key->content_pk mapping to the next version.
-            if content_pk is not None:
+            if content_pk_or_bytes is not None:
+                if isinstance(content_pk_or_bytes, bytes):
+                    file_path, file_content = key, content_pk_or_bytes
+                    media_type_str, _encoding = mimetypes.guess_type(file_path)
+                    # We use "application/octet-stream" as a generic fallback media type, per
+                    # RFC 2046: https://datatracker.ietf.org/doc/html/rfc2046
+                    media_type_str = media_type_str or "application/octet-stream"
+                    media_type = contents_api.get_or_create_media_type(media_type_str)
+                    content = contents_api.get_or_create_file_content(
+                        component.learning_package.id,
+                        media_type.id,
+                        data=file_content,
+                        created=created,
+                    )
+                    content_pk = content.pk
+                else:
+                    content_pk = content_pk_or_bytes
                 ComponentVersionContent.objects.create(
                     content_id=content_pk,
                     component_version=component_version,

openedx_learning/apps/authoring/components/management/commands/add_assets_to_component.py

@@ -4,14 +4,11 @@
 This is mostly meant to be a debugging tool to let us to easily load some test
 asset data into the system.
 """
-import mimetypes
 import pathlib
 from datetime import datetime, timezone
 
 from django.core.management.base import BaseCommand
 
-from ....components.api import create_component_version_content
-from ....contents.api import get_or_create_file_content, get_or_create_media_type
 from ....publishing.api import get_learning_package_by_key
 from ...api import create_next_component_version, get_component_by_key
 
@@ -69,39 +66,18 @@ def handle(self, *args, **options):
         )
 
         created = datetime.now(tz=timezone.utc)
-        keys_to_remove = set()
-        local_keys_to_content = {}
+        local_keys_to_content_bytes = {}
 
         for file_mapping in file_mappings:
             local_key, file_path = file_mapping.split(":", 1)
 
-            # No file_path means to delete this entry from the next version.
-            if not file_path:
-                keys_to_remove.add(local_key)
-                continue
-
-            media_type_str, _encoding = mimetypes.guess_type(file_path)
-            media_type = get_or_create_media_type(media_type_str)
-            content = get_or_create_file_content(
-                learning_package.id,
-                media_type.id,
-                data=pathlib.Path(file_path).read_bytes(),
-                created=created,
-            )
-            local_keys_to_content[local_key] = content.id
+            local_keys_to_content_bytes[local_key] = pathlib.Path(file_path).read_bytes() if file_path else None
 
         next_version = create_next_component_version(
             component.pk,
-            content_to_replace={local_key: None for local_key in keys_to_remove},
+            content_to_replace=local_keys_to_content_bytes,
             created=created,
         )
-        for local_key, content_id in sorted(local_keys_to_content.items()):
-            create_component_version_content(
-                next_version.pk,
-                content_id,
-                key=local_key,
-                learner_downloadable=True,
-            )
 
         self.stdout.write(
             f"Created v{next_version.version_num} of "

tests/openedx_learning/apps/authoring/components/test_api.py

@@ -415,6 +415,30 @@ def test_add(self):
             new_version.contents.get(componentversioncontent__key="nested/path/hello.txt")
         )
 
+    def test_bytes_content(self):
+        bytes_content = b'raw content'
+
+        version_1 = components_api.create_next_component_version(
+            self.problem.pk,
+            title="Problem Version 1",
+            content_to_replace={
+                "raw.txt": bytes_content,
+                "no_ext": bytes_content,
+            },
+            created=self.now,
+        )
+
+        content_txt = version_1.contents.get(componentversioncontent__key="raw.txt")
+        content_raw_txt = version_1.contents.get(componentversioncontent__key="no_ext")
+
+        assert content_txt.size == len(bytes_content)
+        assert str(content_txt.media_type) == 'text/plain'
+        assert content_txt.read_file().read() == bytes_content
+
+        assert content_raw_txt.size == len(bytes_content)
+        assert str(content_raw_txt.media_type) == 'application/octet-stream'
+        assert content_raw_txt.read_file().read() == bytes_content
+
     def test_multiple_versions(self):
         hello_content = contents_api.get_or_create_text_content(
             self.learning_package.id,