Import warning

CHANGELOG.md

@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fix class_name warning: RemovedInWagtail60Warning
 - Helper texts for some ordered content sets
 - CMS Forms Flexible imports and Better Error Handling
+- Contentpage warnings for media_link field
 ### Removed
 - Locale field on exports
 - Menu app

home/content_import_export.py

@@ -15,6 +15,7 @@ def import_content(file, filetype, progress_queue, purge=True, locale=None) -> N
 
     importer = ContentImporter(file.read(), filetype, progress_queue, purge, locale)
     importer.perform_import()
+    return importer
 
 
 def export_xlsx_content(queryset: PageQuerySet, response: HttpResponse) -> None:

home/import_content_pages.py

@@ -17,7 +17,12 @@
 from wagtail.models.sites import Site  # type: ignore
 from wagtail.rich_text import RichText  # type: ignore
 
-from home.import_helpers import ImportException, parse_file, validate_using_form
+from home.import_helpers import (
+    ImportException,
+    ImportWarning,
+    parse_file,
+    validate_using_form,
+)
 
 from .models import (
     Assessment,
@@ -60,6 +65,7 @@ def __init__(
         self.go_to_page_list_items: dict[PageId, dict[int, list[dict[str, Any]]]] = (
             defaultdict(lambda: defaultdict(list))
         )
+        self.import_warnings: list[ImportWarning] = []
 
     def locale_from_display_name(self, langname: str) -> Locale:
         if langname not in self.locale_map:
@@ -89,6 +95,15 @@ def perform_import(self) -> None:
         self.link_related_pages()
         self.add_go_to_page_items(self.go_to_page_buttons, "buttons")
         self.add_go_to_page_items(self.go_to_page_list_items, "list_items")
+        self.add_media_link(rows)
+
+    def add_media_link(self, rows: list["ContentRow"]) -> None:
+        for row_num, row in enumerate(rows, start=2):
+            if row.media_link:
+                if row.media_link is not None or row.media_link != "":
+                    self.import_warnings.append(
+                        ImportWarning(f"{row.media_link}", row_num)
+                    )
 
     def process_rows(self, rows: list["ContentRow"]) -> None:
         # Non-page rows don't have a locale, so we need to remember the last
@@ -743,6 +758,7 @@ class ContentRow:
     related_pages: list[str] = field(default_factory=list)
     footer: str = ""
     language_code: str = ""
+    import_warnings: list[ImportWarning] = field(default_factory=list)
 
     @classmethod
     def from_flat(cls, row: dict[str, str], row_num: int) -> "ContentRow":
@@ -790,6 +806,7 @@ def from_flat(cls, row: dict[str, str], row_num: int) -> "ContentRow":
             list_items=list_items,
             footer=row.pop("footer", ""),
             **row,
+            import_warnings=[],
         )
 
     @property

home/import_helpers.py

@@ -44,6 +44,20 @@ def __init__(
         super().__init__()
 
 
+class ImportWarning:
+    """
+    Base warnings for all import related issues.
+    """
+
+    def __init__(
+        self,
+        message: str | list[str],
+        row_num: int | None = None,
+    ):
+        self.row_num = row_num
+        self.message = message
+
+
 def wagtail_to_formdata(val: Any) -> Any:
     """
     Convert a model dict field that may be a nested streamfield (or associated

home/tests/import-export-data/contentpage_media_link_warning.csv

@@ -0,0 +1,4 @@
+structure,message,page_id,Slug,parent,web_title,web_subtitle,web_body,whatsapp_title,whatsapp_body,whatsapp_template_name,whatsapp_template_category,example_values,variation_title,variation_body,list_title,list_items,sms_title,sms_body,ussd_title,ussd_body,messenger_title,messenger_body,viber_title,viber_body,translation_tag,tags,quick_replies,triggers,next_prompt,buttons,image_link,doc_link,media_link,related_pages,footer,language_code
+Menu 1,0,4,main-menu,,Main Menu,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,en
+Sub 1.1,1,5,main-menu-first-time-user,Main Menu,main menu first time user,,,main menu first time user,Test,,,5892bccd-8025-419d-9a8e-a6a37b755dbf,,,Main menu🏠,,,[],,,,,,,,,,,,[],,,http://test.com/image.png,,,en
+Sub 1.1,1,5,first-time-user,Main Menu,first time user,,,first time user,Test2,,,5892bccd-8025-419d-9a8e-a6a37b755dbf,,,Main menu🏠,,,[],,,,,,,,,,,,[],,,http://test.com/image22.png,,,en

home/tests/import-export-data/contentpage_without_warning.csv

@@ -0,0 +1,3 @@
+structure,message,page_id,Slug,parent,web_title,web_subtitle,web_body,whatsapp_title,whatsapp_body,whatsapp_template_name,whatsapp_template_category,example_values,variation_title,variation_body,list_title,list_items,sms_title,sms_body,ussd_title,ussd_body,messenger_title,messenger_body,viber_title,viber_body,translation_tag,tags,quick_replies,triggers,next_prompt,buttons,image_link,doc_link,media_link,related_pages,footer,language_code
+Menu 1,0,4,main-menu,,Main Menu,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,,en
+Sub 1.1,1,5,main-menu-first-time-user,Main Menu,main menu first time user,,,main menu first time user,Test,,,5892bccd-8025-419d-9a8e-a6a37b755dbf,,,Main menu🏠,,,[],,,,,,,,,,,,[],,,,,,en

home/tests/test_content_import_export.py

@@ -480,11 +480,13 @@ def export_ordered_content(self) -> bytes:
         content = b"".join(stream.streaming_content)
         return content
 
-    def import_content(self, content_bytes: bytes, **kw: Any) -> None:
+    def import_content(self, content_bytes: bytes, **kw: Any) -> Any:
         """
         Import given content in the configured format with the configured importer.
         """
-        import_content(BytesIO(content_bytes), self.format.upper(), Queue(), **kw)
+        return import_content(
+            BytesIO(content_bytes), self.format.upper(), Queue(), **kw
+        )
 
     def import_ordered_sets(self, content_bytes: bytes) -> None:
         import_ordered_sets(BytesIO(content_bytes), self.format.upper(), Queue())
@@ -1694,6 +1696,51 @@ def test_list_item_type_error_message(self, csv_impexp: ImportExport) -> None:
         assert e.value.row_num == 3
         assert e.value.message == ["list item with invalid type 'new_type'"]
 
+    def test_media_link_warning(self, csv_impexp: ImportExport) -> None:
+        """
+        Import a page with media link it should return a warning
+        no error should be raised
+        """
+
+        csv_impexp.import_file("contentpage_media_link_warning.csv")
+
+        page = Page.objects.all()
+
+        assert len(page) > 0
+
+    def test_media_link_warning_response(self, csv_impexp: ImportExport) -> None:
+        """
+        Import a page with media link it should return a warning
+        media_link will be excluded from uploaded data
+        """
+
+        resp = csv_impexp.import_file("contentpage_media_link_warning.csv")
+
+        content = csv_impexp.export_content()
+        src, dst = csv_impexp.csvs2dicts(resp, content)
+
+        assert "media_link" not in [item.keys() for item in dst]
+
+    def test_import_success_no_warnings(self, csv_impexp: ImportExport) -> None:
+        """
+        Import a page that will not have warnings list
+        """
+        content = csv_impexp.read_bytes("contentpage_without_warning.csv")
+        importer = csv_impexp.import_content(content)
+
+        assert importer.import_warnings == []
+
+    def test_import_success_with_warnings(self, csv_impexp: ImportExport) -> None:
+        """
+        Import a page that will return warnings if media_link is not empty
+        """
+        content = csv_impexp.read_bytes("contentpage_media_link_warning.csv")
+        importer = csv_impexp.import_content(content)
+
+        assert len(importer.import_warnings) == 2
+        assert importer.import_warnings[0].message == "http://test.com/image.png"
+        assert importer.import_warnings[0].row_num == 3
+
 
 @pytest.mark.django_db
 class TestExport:

home/views.py

@@ -180,10 +180,12 @@ def __init__(self, purge, locale, **kwargs):
         super().__init__(**kwargs)
 
     def run(self):
+        warnings = []
         try:
-            import_content(
+            importer = import_content(
                 self.file, self.file_type, self.progress_queue, self.purge, self.locale
             )
+
         except ImportException as e:
             self.result_queue.put(
                 (
@@ -198,7 +200,23 @@ def run(self):
             self.result_queue.put((messages.ERROR, ["Content import failed"]))
             logger.exception("Content import failed")
         else:
-            self.result_queue.put((messages.SUCCESS, ["Content import successful"]))
+            warnings = importer.import_warnings
+            if warnings:
+                self.result_queue.put(
+                    (
+                        messages.WARNING,
+                        [
+                            "Content import successful",
+                            *[
+                                f"row {warning.row_num}: {warning.message}"
+                                for warning in warnings
+                            ],
+                        ],
+                    )
+                )
+            else:
+                self.result_queue.put((messages.SUCCESS, ["Content import successful"]))
+
         # Wait until the user has fetched the result message to close the thread
         self.result_queue.join()