Manage emails

dev/environment

@@ -34,3 +34,4 @@ MAIL_SSL=false
 STATUSPAGE_URL=https://2p66nmmycsj3.statuspage.io
 
 TOKEN_PASSWORD_SECRET="an insecure password reset secret key"
+TOKEN_EMAIL_SECRET="an insecure email verification secret key"

tests/common/db/accounts.py

@@ -46,3 +46,4 @@ class Meta:
     user = factory.SubFactory(UserFactory)
     email = FuzzyEmail()
     verified = True
+    primary = True

tests/unit/accounts/test_services.py

@@ -162,16 +162,6 @@ def test_update_user(self, user_service):
         assert password != user_from_db.password
         assert user_service.hasher.verify(password, user_from_db.password)
 
-    def test_verify_email(self, user_service):
-        user = UserFactory.create()
-        EmailFactory.create(user=user, primary=True,
-                            verified=False)
-        EmailFactory.create(user=user, primary=False,
-                            verified=False)
-        user_service.verify_email(user.id, user.emails[0].email)
-        assert user.emails[0].verified
-        assert not user.emails[1].verified
-
     def test_find_by_email(self, user_service):
         user = UserFactory.create()
         EmailFactory.create(user=user, primary=True, verified=False)

tests/unit/accounts/test_views.py

@@ -18,14 +18,15 @@
 import pytest
 
 from pyramid.httpexceptions import HTTPMovedPermanently, HTTPSeeOther
+from sqlalchemy.orm.exc import NoResultFound
 
 from warehouse.accounts import views
 from warehouse.accounts.interfaces import (
     IUserService, ITokenService, TokenExpired, TokenInvalid, TokenMissing,
     TooManyFailedLogins
 )
 
-from ...common.db.accounts import UserFactory
+from ...common.db.accounts import EmailFactory, UserFactory
 
 
 class TestFailedLoginView:
@@ -695,6 +696,166 @@ def test_reset_password_password_date_changed(self, pyramid_request):
         ]
 
 
+class TestVerifyEmail:
+
+    def test_verify_email(self, db_request, user_service, token_service):
+        user = UserFactory(is_active=False)
+        email = EmailFactory(user=user, verified=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),
+            }
+        )
+        db_request.find_service = pretend.call_recorder(
+            lambda *a, **kwargs: token_service,
+        )
+        db_request.session.flash = pretend.call_recorder(
+            lambda *a, **kw: None
+        )
+
+        result = views.verify_email(db_request)
+
+        db_request.db.flush()
+        assert email.verified
+        assert user.is_active
+        assert isinstance(result, HTTPSeeOther)
+        assert result.headers["Location"] == "/"
+        assert db_request.route_path.calls == [pretend.call('manage.profile')]
+        assert token_service.loads.calls == [pretend.call('RANDOM_KEY')]
+        assert db_request.session.flash.calls == [
+            pretend.call(
+                f"Email address {email.email} verified. " +
+                "You can now set this email as your primary address.",
+                queue="success"
+            ),
+        ]
+        assert db_request.find_service.calls == [
+            pretend.call(ITokenService, name="email"),
+        ]
+
+    @pytest.mark.parametrize(
+        ("exception", "message"),
+        [
+            (
+                TokenInvalid,
+                "Invalid token - Request a new verification link",
+            ), (
+                TokenExpired,
+                "Expired token - Request a new verification link",
+            ), (
+                TokenMissing,
+                "Invalid token - No token supplied"
+            ),
+        ],
+    )
+    def test_verify_email_loads_failure(
+            self, pyramid_request, exception, message):
+
+        def loads(token):
+            raise exception
+
+        pyramid_request.find_service = (
+            lambda *a, **kw: pretend.stub(loads=loads)
+        )
+        pyramid_request.params = {"token": "RANDOM_KEY"}
+        pyramid_request.route_path = pretend.call_recorder(lambda name: "/")
+        pyramid_request.session.flash = pretend.call_recorder(
+            lambda *a, **kw: None
+        )
+
+        views.verify_email(pyramid_request)
+
+        assert pyramid_request.route_path.calls == [
+            pretend.call('manage.profile'),
+        ]
+        assert pyramid_request.session.flash.calls == [
+            pretend.call(message, queue='error'),
+        ]
+
+    def test_verify_email_invalid_action(self, pyramid_request):
+        data = {
+            'action': 'invalid-action',
+        }
+        pyramid_request.find_service = (
+            lambda *a, **kw: pretend.stub(loads=lambda a: data)
+        )
+        pyramid_request.params = {"token": "RANDOM_KEY"}
+        pyramid_request.route_path = pretend.call_recorder(lambda name: "/")
+        pyramid_request.session.flash = pretend.call_recorder(
+            lambda *a, **kw: None
+        )
+
+        views.verify_email(pyramid_request)
+
+        assert pyramid_request.route_path.calls == [
+            pretend.call('manage.profile'),
+        ]
+        assert pyramid_request.session.flash.calls == [
+            pretend.call(
+                "Invalid token - Not an email verification token",
+                queue='error',
+            ),
+        ]
+
+    def test_verify_email_not_found(self, pyramid_request):
+        data = {
+            'action': 'email-verify',
+            'email.id': 'invalid',
+        }
+        pyramid_request.find_service = (
+            lambda *a, **kw: pretend.stub(loads=lambda a: data)
+        )
+        pyramid_request.params = {"token": "RANDOM_KEY"}
+        pyramid_request.route_path = pretend.call_recorder(lambda name: "/")
+        pyramid_request.session.flash = pretend.call_recorder(
+            lambda *a, **kw: None
+        )
+
+        def raise_no_result(*a):
+            raise NoResultFound
+
+        pyramid_request.db = pretend.stub(query=raise_no_result)
+
+        views.verify_email(pyramid_request)
+
+        assert pyramid_request.route_path.calls == [
+            pretend.call('manage.profile'),
+        ]
+        assert pyramid_request.session.flash.calls == [
+            pretend.call('Email not found', queue='error')
+        ]
+
+    def test_verify_email_already_verified(self, db_request):
+        user = UserFactory()
+        email = EmailFactory(user=user, verified=True)
+        data = {
+            'action': 'email-verify',
+            'email.id': email.id,
+        }
+        db_request.user = user
+        db_request.find_service = (
+            lambda *a, **kw: pretend.stub(loads=lambda a: data)
+        )
+        db_request.params = {"token": "RANDOM_KEY"}
+        db_request.route_path = pretend.call_recorder(lambda name: "/")
+        db_request.session.flash = pretend.call_recorder(
+            lambda *a, **kw: None
+        )
+
+        views.verify_email(db_request)
+
+        assert db_request.route_path.calls == [
+            pretend.call('manage.profile'),
+        ]
+        assert db_request.session.flash.calls == [
+            pretend.call('Email already verified', queue='error')
+        ]
+
+
 class TestProfileCallout:
 
     def test_profile_callout_returns_user(self):

tests/unit/manage/test_forms.py

@@ -69,3 +69,12 @@ def test_validate_role_name_fails(self, value, expected):
 
         assert not form.validate()
         assert form.role_name.errors == [expected]
+
+
+class TestAddEmailForm:
+
+    def test_creation(self):
+        user_service = pretend.stub()
+        form = forms.AddEmailForm(user_service=user_service)
+
+        assert form.user_service is user_service