fix: update domain status during verification (#18357)

Author: miketheman

tests/conftest.py

@@ -58,6 +58,7 @@
 from warehouse.organizations.interfaces import IOrganizationService
 from warehouse.packaging import services as packaging_services
 from warehouse.packaging.interfaces import IProjectService
+from warehouse.rate_limiting import DummyRateLimiter, IRateLimiter
 from warehouse.search import services as search_services
 from warehouse.search.interfaces import ISearchService
 from warehouse.subscriptions import services as subscription_services
@@ -164,6 +165,7 @@ def pyramid_services(
     query_results_cache_service,
     search_service,
     domain_status_service,
+    ratelimit_service,
 ):
     services = _Services()
 
@@ -190,6 +192,8 @@ def pyramid_services(
     services.register_service(query_results_cache_service, IQueryResultsCache)
     services.register_service(search_service, ISearchService)
     services.register_service(domain_status_service, IDomainStatusService)
+    services.register_service(ratelimit_service, IRateLimiter, name="email.add")
+    services.register_service(ratelimit_service, IRateLimiter, name="email.verify")
 
     return services
 
@@ -550,6 +554,13 @@ def domain_status_service(mocker):
     return service
 
 
+@pytest.fixture
+def ratelimit_service(mocker):
+    service = DummyRateLimiter()
+    mocker.spy(service, "clear")
+    return service
+
+
 class QueryRecorder:
     def __init__(self):
         self.queries = []

tests/unit/accounts/test_views.py

@@ -24,6 +24,7 @@
 
 from warehouse.accounts import views
 from warehouse.accounts.interfaces import (
+    IDomainStatusService,
     IPasswordBreachedService,
     ITokenService,
     IUserService,
@@ -2575,52 +2576,47 @@ class TestVerifyEmail:
         ],
     )
     def test_verify_email(
-        self, db_request, user_service, token_service, is_primary, confirm_message
+        self, mocker, db_request, ratelimit_service, is_primary, confirm_message
     ):
         user = UserFactory(is_active=False, totp_secret=None)
         email = EmailFactory(user=user, verified=False, primary=is_primary)
         db_request.user = user
         db_request.GET.update({"token": "RANDOM_KEY"})
-        db_request.route_path = pretend.call_recorder(lambda name: "/")
-        token_service.loads = pretend.call_recorder(
-            lambda token: {"action": "email-verify", "email.id": str(email.id)}
-        )
-        email_limiter = pretend.stub(clear=pretend.call_recorder(lambda a: None))
-        verify_limiter = pretend.stub(clear=pretend.call_recorder(lambda a: None))
-        services = {
-            "email": token_service,
-            "email.add": email_limiter,
-            "email.verify": verify_limiter,
-        }
-        db_request.find_service = pretend.call_recorder(
-            lambda a, name, **kwargs: services[name]
+        db_request.route_path = mocker.Mock(return_value="/")
+        mock_token_service_loads = mocker.patch(
+            "warehouse.accounts.services.TokenService.loads",
+            return_value={"action": "email-verify", "email.id": str(email.id)},
         )
-        db_request.session.flash = pretend.call_recorder(lambda *a, **kw: None)
+        spy_find_service = mocker.spy(db_request, "find_service")
+        spy_session_flash = mocker.spy(db_request.session, "flash")
 
         result = views.verify_email(db_request)
 
         db_request.db.flush()
         assert email.verified
+        assert email.domain_last_status == ["active"]
         assert user.is_active
         assert isinstance(result, HTTPSeeOther)
         assert result.headers["Location"] == "/"
-        assert db_request.route_path.calls == [
-            pretend.call("manage.account.two-factor")
-        ]
-        assert token_service.loads.calls == [pretend.call("RANDOM_KEY")]
-        assert email_limiter.clear.calls == [pretend.call(db_request.remote_addr)]
-        assert verify_limiter.clear.calls == [pretend.call(user.id)]
-        assert db_request.session.flash.calls == [
-            pretend.call(
-                f"Email address {email.email} verified. " + confirm_message,
-                queue="success",
-            )
-        ]
-        assert db_request.find_service.calls == [
-            pretend.call(ITokenService, name="email"),
-            pretend.call(IRateLimiter, name="email.add"),
-            pretend.call(IRateLimiter, name="email.verify"),
-        ]
+        db_request.route_path.assert_called_once_with("manage.account.two-factor")
+        mock_token_service_loads.assert_called_once_with("RANDOM_KEY")
+        ratelimit_service.clear.assert_has_calls(
+            [
+                mocker.call(db_request.remote_addr),
+                mocker.call(user.id),
+            ]
+        )
+        spy_session_flash.assert_called_once_with(
+            f"Email address {email.email} verified. " + confirm_message, queue="success"
+        )
+        spy_find_service.assert_has_calls(
+            [
+                mocker.call(ITokenService, name="email"),
+                mocker.call(IRateLimiter, name="email.add"),
+                mocker.call(IRateLimiter, name="email.verify"),
+                mocker.call(IDomainStatusService),
+            ]
+        )
 
     @pytest.mark.parametrize(
         ("exception", "message"),
@@ -2702,34 +2698,24 @@ def test_verify_email_already_verified(self, db_request):
             pretend.call("Email already verified", queue="error")
         ]
 
-    def test_verify_email_with_existing_2fa(self, db_request, token_service):
+    def test_verify_email_with_existing_2fa(self, mocker, db_request):
         user = UserFactory(is_active=False, totp_secret=b"secret")
         email = EmailFactory(user=user, verified=False, primary=False)
         db_request.user = user
         db_request.GET.update({"token": "RANDOM_KEY"})
-        db_request.route_path = pretend.call_recorder(lambda name: "/")
-        token_service.loads = pretend.call_recorder(
-            lambda token: {"action": "email-verify", "email.id": str(email.id)}
-        )
-        email_limiter = pretend.stub(clear=pretend.call_recorder(lambda a: None))
-        verify_limiter = pretend.stub(clear=pretend.call_recorder(lambda a: None))
-        services = {
-            "email": token_service,
-            "email.add": email_limiter,
-            "email.verify": verify_limiter,
-        }
-        db_request.find_service = pretend.call_recorder(
-            lambda a, name, **kwargs: services[name]
+        db_request.route_path = mocker.Mock(return_value="/")
+        mocker.patch(
+            "warehouse.accounts.services.TokenService.loads",
+            return_value={"action": "email-verify", "email.id": str(email.id)},
         )
-        db_request.session.flash = pretend.call_recorder(lambda *a, **kw: None)
 
         assert db_request.user.has_two_factor
 
         result = views.verify_email(db_request)
 
         assert isinstance(result, HTTPSeeOther)
         assert result.headers["Location"] == "/"
-        assert db_request.route_path.calls == [pretend.call("manage.account")]
+        db_request.route_path.assert_called_once_with("manage.account")
         assert db_request.user.is_active
 
 

warehouse/accounts/views.py

@@ -50,6 +50,7 @@
     TooManyPasswordResetRequests,
 )
 from warehouse.accounts.models import Email, TermsOfServiceEngagement, User
+from warehouse.accounts.utils import update_email_domain_status
 from warehouse.admin.flags import AdminFlagValue
 from warehouse.authnz import Permissions
 from warehouse.cache.origin import origin_cache
@@ -1006,6 +1007,8 @@ def _error(message):
 
     if email.verified:
         return _error(request._("Email already verified"))
+    # Run the domain status update now that the end user has verified it.
+    update_email_domain_status(email, request)
 
     email.verified = True
     email.unverify_reason = None