Merge pull request #817 from moneebullah25/fix/lead-import-archived-tag

fix(leads): report archived tags in a CSV import as row errors

backend/leads/csv_import.py

@@ -224,10 +224,15 @@ def parse_and_validate(file_bytes: bytes, request, legacy=False) -> ImportResult
     profiles, teams = _reference_maps(parsed, org)
     valid: list[ValidatedRow] = []
     seen_emails: dict[str, int] = {}
+    # Restoring a tag is an org-admin action in the tags API, so an import
+    # (open to sales members) reports it instead of reactivating it.
+    archived = set(
+        Tags.objects.filter(org=org, is_active=False).values_list("slug", flat=True)
+    )
 
     for idx, record in parsed:
         row_errors, validated = _validate_row(
-            idx, record, request, profiles, teams, seen_emails, legacy
+            idx, record, request, profiles, teams, seen_emails, legacy, archived
         )
         if row_errors:
             errors.extend(row_errors)
@@ -272,7 +277,7 @@ def _reference_maps(parsed, org) -> tuple[dict[str, str], dict[str, str]]:
     return profiles, teams
 
 
-def _validate_row(idx, record, request, profiles, teams, seen_emails, legacy):
+def _validate_row(idx, record, request, profiles, teams, seen_emails, legacy, archived):
     errors: list[RowError] = []
 
     if legacy:
@@ -345,6 +350,14 @@ def _validate_row(idx, record, request, profiles, teams, seen_emails, legacy):
         name_error = Tags.name_error(name)
         if name_error:
             errors.append(RowError(idx, "tags", f"Tag '{name}': {name_error}"))
+        elif Tags.slug_for(name) in archived:
+            errors.append(
+                RowError(
+                    idx,
+                    "tags",
+                    f"Tag '{name}' is archived; restore it in Settings first",
+                )
+            )
 
     if errors:
         return errors, None
@@ -407,7 +420,6 @@ def _get_or_create_tag(name: str, org, cache: dict[str, Tags]) -> Tags:
     # SELECT-then-INSERT race when two imports name the same new tag.
     slug = Tags.slug_for(name)
     if slug not in cache:
-        cache[slug], _ = Tags.objects.get_or_create(
-            slug=slug, org=org, defaults={"name": name}
-        )
+        tag, _ = Tags.objects.get_or_create(slug=slug, org=org, defaults={"name": name})
+        cache[slug] = tag
     return cache[slug]

backend/leads/tests/test_lead_import.py

@@ -623,3 +623,27 @@ class TestTagSlugs:
         tag = Tags.objects.get(org=org_a)
         assert tag.slug == "trade-show"
         assert set(Lead.objects.values_list("tags", flat=True)) == {tag.id}
+
+    @pytest.mark.parametrize("url", [PREVIEW, COMMIT])
+    def test_archived_tag_is_a_row_error_for_an_admin(self, admin_client, org_a, url):
+        Tags.objects.create(name="VIP", slug="vip", org=org_a, is_active=False)
+        response = _post(admin_client, url, self._tagged("a@example.com", "vip "))
+        error = response.json()["errors"][0]
+        assert (error["row"], error["field"]) == (1, "tags")
+        assert error["message"] == "Tag 'vip' is archived; restore it in Settings first"
+        assert not Tags.objects.get(org=org_a).is_active
+        assert not Lead.objects.exists()
+
+    @pytest.mark.parametrize("url", [PREVIEW, COMMIT])
+    def test_archived_tag_is_a_row_error_for_sales_access(
+        self, user_client, user_profile, org_a, url
+    ):
+        user_profile.has_sales_access = True
+        user_profile.save(update_fields=["has_sales_access"])
+        Tags.objects.create(name="VIP", slug="vip", org=org_a, is_active=False)
+        response = _post(user_client, url, self._tagged("a@example.com", "VIP"))
+        error = response.json()["errors"][0]
+        assert (error["row"], error["field"]) == (1, "tags")
+        assert "archived" in error["message"]
+        assert not Tags.objects.get(org=org_a).is_active
+        assert not Lead.objects.exists()