fix(headless): delete pending email address

Author: pennersr

ChangeLog.rst

@@ -15,6 +15,11 @@ Note worthy changes
   To support this, a new field was added to the ``Client`` model. Therefore, you
   need to run migrations for the ``allauth.idp.oidc`` app.
 
+- Headless: an email address that was in the process of being verified by code
+  was presented in the list of email address, but could not be deleted. Now, a
+  ``DELETE`` will actually abort the process, effectively removing the pending
+  email address from the list.
+
 
 65.11.2 (2025-09-09)
 ********************

allauth/account/adapter.py

@@ -119,6 +119,8 @@ def can_delete_email(self, email_address) -> bool:
         """
         from allauth.account.models import EmailAddress
 
+        if not email_address.pk:
+            return True
         has_other = (
             EmailAddress.objects.filter(user_id=email_address.user_id)
             .exclude(pk=email_address.pk)

allauth/account/internal/flows/email_verification_by_code.py

@@ -87,6 +87,8 @@ def initiate(cls, *, request, user, email: str) -> "EmailVerificationProcess":
 
     @classmethod
     def resume(cls, request) -> Optional["EmailVerificationProcess"]:
+        if not app_settings.EMAIL_VERIFICATION_BY_CODE_ENABLED:
+            return None
         state = request.session.get(EMAIL_VERIFICATION_CODE_SESSION_KEY)
         if not state:
             return None

allauth/account/internal/flows/manage_email.py

@@ -68,7 +68,9 @@ def add_email(request: HttpRequest, form):
         )
 
 
-def can_mark_as_primary(email_address: EmailAddress):
+def can_mark_as_primary(email_address: EmailAddress) -> bool:
+    if not email_address.pk:
+        return False
     return (
         email_address.verified
         or not EmailAddress.objects.filter(

allauth/headless/account/inputs.py

@@ -200,12 +200,29 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def clean_email(self):
+        self.process = None
         email = self.cleaned_data["email"]
         validate_email(email)
+
+        # Select a database backed email...
         try:
             return EmailAddress.objects.get_for_user(user=self.user, email=email)
         except EmailAddress.DoesNotExist:
-            raise get_adapter().validation_error("unknown_email")
+            pass
+
+        # Or, if email verification by code is active, try the pending email
+        request = context.request
+        self.process = flows.email_verification_by_code.EmailVerificationProcess.resume(
+            request
+        )
+        if self.process:
+            email_address = self.process.email_address
+            if email_address.email.lower() == email.lower() and (
+                request.user.is_anonymous or email_address.user_id == request.user.pk
+            ):
+                return email_address
+
+        raise get_adapter().validation_error("unknown_email")
 
 
 class DeleteEmailInput(SelectEmailInput):
@@ -227,18 +244,7 @@ def clean_email(self):
 
 
 class ResendEmailVerificationInput(SelectEmailInput):
-    def clean_email(self):
-        if not account_settings.EMAIL_VERIFICATION_BY_CODE_ENABLED:
-            self.process = None
-            return super().clean_email()
-        email = self.cleaned_data["email"]
-        validate_email(email)
-        self.process = flows.email_verification_by_code.EmailVerificationProcess.resume(
-            context.request
-        )
-        if not self.process:
-            raise get_adapter().validation_error("unknown_email")
-        return self.process.email_address
+    pass
 
 
 class ReauthenticateInput(ReauthenticateForm, inputs.Input):

allauth/headless/account/views.py

@@ -420,7 +420,10 @@ def post(self, request, *args, **kwargs):
 
     def delete(self, request, *args, **kwargs):
         addr = self.input.cleaned_data["email"]
-        flows.manage_email.delete_email(request, addr)
+        if addr.pk:
+            flows.manage_email.delete_email(request, addr)
+        else:
+            self.input.process.abort()
         return self._respond_email_list()
 
     def patch(self, request, *args, **kwargs):

tests/apps/headless/account/test_email_verification_by_code.py

@@ -36,7 +36,7 @@ def test_email_verification_rate_limits_login(
             content_type="application/json",
         )
         if attempt == 0:
-            assert resp.status_code == 401
+            assert resp.status_code == HTTPStatus.UNAUTHORIZED
             flow = [
                 flow for flow in resp.json()["data"]["flows"] if flow.get("is_pending")
             ][0]
@@ -80,7 +80,7 @@ def test_email_verification_rate_limits_submitting_codes(
         },
         content_type="application/json",
     )
-    assert resp.status_code == 401
+    assert resp.status_code == HTTPStatus.UNAUTHORIZED
     flow = [flow for flow in resp.json()["data"]["flows"] if flow.get("is_pending")][0]
     assert flow["id"] == Flow.VERIFY_EMAIL
 
@@ -134,7 +134,7 @@ def test_add_email(
         data={"email": new_email},
         content_type="application/json",
     )
-    assert resp.status_code == 200
+    assert resp.status_code == HTTPStatus.OK
 
     # It's in the response, albeit unverified.
     assert len(resp.json()["data"]) == 2
@@ -162,7 +162,7 @@ def test_add_email(
         data={"key": code},
         content_type="application/json",
     )
-    assert resp.status_code == 200
+    assert resp.status_code == HTTPStatus.OK
     assert resp.json()["data"]["user"]["email"] == new_email
 
     # ACCOUNT_CHANGE_EMAIL = True, so the other one is gone.
@@ -204,7 +204,7 @@ def test_signup_with_email_verification(
         },
         content_type="application/json",
     )
-    assert resp.status_code == 401
+    assert resp.status_code == HTTPStatus.UNAUTHORIZED
     assert User.objects.filter(email=email).exists()
     data = resp.json()
     flow = next((f for f in data["data"]["flows"] if f.get("is_pending")))
@@ -215,14 +215,14 @@ def test_signup_with_email_verification(
         headless_reverse("headless:account:verify_email"),
         HTTP_X_EMAIL_VERIFICATION_KEY=code,
     )
-    assert resp.status_code == 200
+    assert resp.status_code == HTTPStatus.OK
     assert resp.json() == {
         "data": {
             "email": email,
             "user": ANY,
         },
         "meta": {"is_authenticating": True},
-        "status": 200,
+        "status": HTTPStatus.OK,
     }
     resp = client.post(
         headless_reverse("headless:account:verify_email"),
@@ -232,7 +232,7 @@ def test_signup_with_email_verification(
     addr = EmailAddress.objects.get(email=email)
     assert addr.verified
 
-    assert resp.status_code == 200
+    assert resp.status_code == HTTPStatus.OK
     data = resp.json()
     assert data["meta"]["is_authenticated"]
 
@@ -267,7 +267,7 @@ def test_resend_at_signup(
         },
         content_type="application/json",
     )
-    assert resp.status_code == 401
+    assert resp.status_code == HTTPStatus.UNAUTHORIZED
     assert User.objects.filter(email=email).exists()
     data = resp.json()
     flow = next((f for f in data["data"]["flows"] if f.get("is_pending")))
@@ -278,7 +278,7 @@ def test_resend_at_signup(
         headless_reverse("headless:account:verify_email"),
         HTTP_X_EMAIL_VERIFICATION_KEY=code,
     )
-    assert resp.status_code == 200
+    assert resp.status_code == HTTPStatus.OK
     assert resp.json() == {
         "data": {
             "email": email,
@@ -339,3 +339,41 @@ def test_add_resend_verify_email(
             content_type="application/json",
         )
         assert EmailAddress.objects.filter(email=new_email, verified=True).exists()
+
+
+def test_remove_unverified_email(
+    auth_client,
+    user,
+    email_factory,
+    headless_reverse,
+    settings,
+    get_last_email_verification_code,
+    mailoutbox,
+):
+    settings.ACCOUNT_AUTHENTICATION_METHOD = "email"
+    settings.ACCOUNT_EMAIL_VERIFICATION_BY_CODE_ENABLED = True
+    settings.ACCOUNT_CHANGE_EMAIL = True
+    new_email = email_factory()
+
+    # Let's add an email...
+    resp = auth_client.post(
+        headless_reverse("headless:account:manage_email"),
+        data={"email": new_email},
+        content_type="application/json",
+    )
+    assert resp.status_code == HTTPStatus.OK
+
+    # It's in the response, albeit unverified.
+    assert len(resp.json()["data"]) == 2
+    email_map = {addr["email"]: addr for addr in resp.json()["data"]}
+    assert not email_map[new_email]["verified"]
+
+    # Delete the pending email.
+    resp = auth_client.delete(
+        headless_reverse("headless:account:manage_email"),
+        data={"email": new_email},
+        content_type="application/json",
+    )
+    assert resp.status_code == HTTPStatus.OK
+    assert len(resp.json()["data"]) == 1
+    assert new_email not in {addr["email"] for addr in resp.json()["data"]}