Added tests for bulk import review. Fixed bug with campaign/project slug check in review

concordia/admin/views.py

@@ -370,14 +370,14 @@ def admin_bulk_import_review(request):
 
                     # Read Campaign slug value from excel
                     campaign_slug = row["Campaign Slug"]
-                    if campaign_slug and not pattern.match(campaign_slug):
+                    if campaign_slug and not pattern.fullmatch(campaign_slug):
                         messages.warning(
                             request, "Campaign slug doesn't match pattern."
                         )
 
                     # Read Project slug value from excel
                     project_slug = row["Project Slug"]
-                    if project_slug and not pattern.match(project_slug):
+                    if project_slug and not pattern.fullmatch(project_slug):
                         messages.warning(request, "Project slug doesn't match pattern.")
 
                     potential_urls = filter(None, re.split(r"[\s]+", import_url_blob))
@@ -389,20 +389,13 @@ def admin_bulk_import_review(request):
                             )
                             continue
 
-                        try:
-                            urls.append(url)
-                            url_counter = url_counter + 1
-
-                            if url_counter == 50:
-                                all_urls.append(urls)
-                                url_counter = 0
-                                urls = []
+                        urls.append(url)
+                        url_counter = url_counter + 1
 
-                        except Exception as exc:
-                            messages.error(
-                                request,
-                                f"Unhandled error attempting to count {url}: {exc}",
-                            )
+                        if url_counter == 50:
+                            all_urls.append(urls)
+                            url_counter = 0
+                            urls = []
 
                 all_urls.append(urls)
                 for _i, val in enumerate(all_urls):

concordia/tests/test_admin_views.py

@@ -688,6 +688,353 @@ def test_import_task_exception(self):
             )
 
 
+class TestAdminBulkImportReview(CreateTestUsers, TestCase):
+    def setUp(self):
+        self.login_user(is_staff=True, is_superuser=True)
+        self.path = reverse("admin:bulk-review")
+        self.campaign_title = "Test Campaign"
+        self.campaign_short_description = "Short description"
+        self.campaign_long_description = "Long description"
+        self.campaign_slug = "test-campaign"
+        self.project_slug = "test-project"
+        self.project_title = "Test Project"
+        self.project_description = "Project description"
+        self.url = "http://example.com"
+        self.spreadsheet_data = {
+            "Campaign": self.campaign_title,
+            "Campaign Short Description": self.campaign_short_description,
+            "Campaign Long Description": self.campaign_long_description,
+            "Campaign Slug": self.campaign_slug,
+            "Project Slug": self.project_slug,
+            "Project": self.project_title,
+            "Project Description": self.project_description,
+            "Import URLs": self.url,
+        }
+        self.post_data = {"spreadsheet_file": BytesIO()}
+
+    def test_get(self):
+        response = self.client.get(self.path)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("form", response.context)
+
+    def test_invalid_form(self):
+        response = self.client.post(self.path)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("form", response.context)
+
+    def test_fully_valid_form(self):
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [self.spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+            self.assertEqual(response.status_code, 200)
+            messages = [str(message) for message in get_messages(response.wsgi_request)]
+            self.assertEqual(len(messages), 3)
+            self.assertEqual(messages[0], "Fetch test message")
+            self.assertEqual(messages[1], "Total Asset\xa0Count:1")
+            self.assertEqual(messages[2], "All Processes Completed")
+
+    def test_missing_field(self):
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+        del spreadsheet_data["Campaign"]
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 4)
+        self.assertEqual(messages[0], "Skipping row 0: missing fields ['Campaign']")
+        self.assertEqual(messages[1], "Fetch test message")
+        self.assertEqual(messages[2], "Total Asset\xa0Count:1")
+        self.assertEqual(messages[3], "All Processes Completed")
+
+    def test_empty_field(self):
+        # Only three fields require values: Campaign, Projet and Import URLs.
+        # Other fields must be present but can be empty.
+        # This tests that blank value check
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            # Test one empty field
+            spreadsheet_data["Campaign"] = ""
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+            self.assertEqual(response.status_code, 200)
+            messages = [str(message) for message in get_messages(response.wsgi_request)]
+            self.assertEqual(len(messages), 4)
+            self.assertEqual(
+                messages[0],
+                "Skipping row 0: at least one required field "
+                "(Campaign, Project, Import URLs) is empty",
+            )
+            self.assertEqual(messages[1], "Fetch test message")
+            self.assertEqual(messages[2], "Total Asset\xa0Count:1")
+            self.assertEqual(messages[3], "All Processes Completed")
+
+    def test_all_empty_fields(self):
+        # If all values in a spreadsheet row are empty, the row is skipped silently
+        spreadsheet_data = {key: "" for key in self.spreadsheet_data.keys()}
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 3)
+        self.assertEqual(messages[0], "Fetch test message")
+        self.assertEqual(messages[1], "Total Asset\xa0Count:1")
+        self.assertEqual(messages[2], "All Processes Completed")
+
+    def test_empty_campaign_slug(self):
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+        spreadsheet_data["Campaign Slug"] = ""
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 3)
+        self.assertEqual(messages[0], "Fetch test message")
+        self.assertEqual(messages[1], "Total Asset\xa0Count:1")
+        self.assertEqual(messages[2], "All Processes Completed")
+
+    def test_bad_campaign_slug(self):
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+        spreadsheet_data["Campaign Slug"] = "bad#slug@"
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 4)
+        self.assertEqual(messages[0], "Campaign slug doesn't match pattern.")
+        self.assertEqual(messages[1], "Fetch test message")
+        self.assertEqual(messages[2], "Total Asset\xa0Count:1")
+        self.assertEqual(messages[3], "All Processes Completed")
+
+    def test_empty_project_slug(self):
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+        spreadsheet_data["Project Slug"] = ""
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 3)
+        self.assertEqual(messages[0], "Fetch test message")
+        self.assertEqual(messages[1], "Total Asset\xa0Count:1")
+        self.assertEqual(messages[2], "All Processes Completed")
+
+    def test_bad_project_slug(self):
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+        spreadsheet_data["Project Slug"] = "bad#slug@"
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 4)
+        self.assertEqual(messages[0], "Project slug doesn't match pattern.")
+        self.assertEqual(messages[1], "Fetch test message")
+        self.assertEqual(messages[2], "Total Asset\xa0Count:1")
+        self.assertEqual(messages[3], "All Processes Completed")
+
+    def test_bad_url(self):
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+        spreadsheet_data["Import URLs"] = bad_url = "ftp://example.com"
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 4)
+        self.assertEqual(messages[0], f"Skipping unrecognized URL value: {bad_url}")
+        self.assertEqual(messages[1], "Fetch test message")
+        self.assertEqual(messages[2], "Total Asset\xa0Count:1")
+        self.assertEqual(messages[3], "All Processes Completed")
+
+    def test_large_number_urls(self):
+        spreadsheet_data = copy.copy(self.spreadsheet_data)
+        spreadsheet_data["Import URLs"] = " ".join([self.url for i in range(51)])
+
+        with (
+            mock.patch(
+                "concordia.admin.views.AdminProjectBulkImportForm", autospec=True
+            ) as form_mock,
+            mock.patch(
+                "concordia.admin.views.slurp_excel", autospec=True
+            ) as slurp_mock,
+            mock.patch(
+                "concordia.admin.views.fetch_all_urls",
+                autospec=True,
+            ) as fetch_mock,
+        ):
+            form_mock.return_value.is_valid.return_value = True
+            form_mock.return_value.cleaned_data = {}
+            slurp_mock.return_value = [spreadsheet_data]
+            fetch_mock.return_value = [["Fetch test message"], 1]
+
+            response = self.client.post(self.path, data=self.post_data)
+
+        self.assertEqual(response.status_code, 200)
+        messages = [str(message) for message in get_messages(response.wsgi_request)]
+        self.assertEqual(len(messages), 4)
+        self.assertEqual(messages[0], "Fetch test message")
+        self.assertEqual(messages[1], "Fetch test message")
+        # This count is weird because we mock the fetch_all_urls function
+        self.assertEqual(messages[2], "Total Asset\xa0Count:2")
+        self.assertEqual(messages[3], "All Processes Completed")
+
+
 class TestSerializedObjectView(TestCase):
     def setUp(self):
         self.card = create_card()