mirror of
				https://github.com/zulip/zulip.git
				synced 2025-11-04 05:53:43 +00:00 
			
		
		
		
	
		
			
				
	
	
		
			206 lines
		
	
	
		
			8.6 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			206 lines
		
	
	
		
			8.6 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
# -*- coding: utf-8 -*-
 | 
						|
from __future__ import absolute_import
 | 
						|
 | 
						|
import datetime
 | 
						|
from typing import Any
 | 
						|
 | 
						|
import django
 | 
						|
import mock
 | 
						|
from django.conf import settings
 | 
						|
from django.core import mail
 | 
						|
from django.http import HttpResponse
 | 
						|
from django.urls import reverse
 | 
						|
from django.utils.timezone import now
 | 
						|
 | 
						|
from confirmation.models import Confirmation, generate_key, confirmation_url
 | 
						|
from zerver.lib.actions import do_start_email_change_process, do_set_realm_property
 | 
						|
from zerver.lib.test_classes import (
 | 
						|
    ZulipTestCase,
 | 
						|
)
 | 
						|
from zerver.lib.send_email import FromAddress
 | 
						|
from zerver.models import get_user, EmailChangeStatus, Realm, get_realm
 | 
						|
 | 
						|
 | 
						|
class EmailChangeTestCase(ZulipTestCase):
 | 
						|
    def test_confirm_email_change_with_non_existent_key(self):
 | 
						|
        # type: () -> None
 | 
						|
        self.login(self.example_email("hamlet"))
 | 
						|
        key = generate_key()
 | 
						|
        url = confirmation_url(key, 'testserver', Confirmation.EMAIL_CHANGE)
 | 
						|
        response = self.client_get(url)
 | 
						|
        self.assert_in_success_response(["Whoops. We couldn't find your confirmation link in the system."], response)
 | 
						|
 | 
						|
    def test_confirm_email_change_with_invalid_key(self):
 | 
						|
        # type: () -> None
 | 
						|
        self.login(self.example_email("hamlet"))
 | 
						|
        key = 'invalid key'
 | 
						|
        url = confirmation_url(key, 'testserver', Confirmation.EMAIL_CHANGE)
 | 
						|
        response = self.client_get(url)
 | 
						|
        self.assert_in_success_response(["Whoops. The confirmation link is malformed."], response)
 | 
						|
 | 
						|
    def test_email_change_when_not_logging_in(self):
 | 
						|
        # type: () -> None
 | 
						|
        key = generate_key()
 | 
						|
        url = confirmation_url(key, 'testserver', Confirmation.EMAIL_CHANGE)
 | 
						|
        response = self.client_get(url)
 | 
						|
        self.assertEqual(response.status_code, 302)
 | 
						|
 | 
						|
    def test_confirm_email_change_when_time_exceeded(self):
 | 
						|
        # type: () -> None
 | 
						|
        user_profile = self.example_user('hamlet')
 | 
						|
        old_email = user_profile.email
 | 
						|
        new_email = 'hamlet-new@zulip.com'
 | 
						|
        self.login(self.example_email("hamlet"))
 | 
						|
        obj = EmailChangeStatus.objects.create(new_email=new_email,
 | 
						|
                                               old_email=old_email,
 | 
						|
                                               user_profile=user_profile,
 | 
						|
                                               realm=user_profile.realm)
 | 
						|
        key = generate_key()
 | 
						|
        date_sent = now() - datetime.timedelta(days=2)
 | 
						|
        Confirmation.objects.create(content_object=obj,
 | 
						|
                                    date_sent=date_sent,
 | 
						|
                                    confirmation_key=key,
 | 
						|
                                    type=Confirmation.EMAIL_CHANGE)
 | 
						|
        url = confirmation_url(key, user_profile.realm.host, Confirmation.EMAIL_CHANGE)
 | 
						|
        response = self.client_get(url)
 | 
						|
        self.assert_in_success_response(["Whoops. The confirmation link has expired."], response)
 | 
						|
 | 
						|
    def test_confirm_email_change(self):
 | 
						|
        # type: () -> None
 | 
						|
        user_profile = self.example_user('hamlet')
 | 
						|
        old_email = user_profile.email
 | 
						|
        new_email = 'hamlet-new@zulip.com'
 | 
						|
        new_realm = get_realm('zulip')
 | 
						|
        self.login(self.example_email('hamlet'))
 | 
						|
        obj = EmailChangeStatus.objects.create(new_email=new_email,
 | 
						|
                                               old_email=old_email,
 | 
						|
                                               user_profile=user_profile,
 | 
						|
                                               realm=user_profile.realm)
 | 
						|
        key = generate_key()
 | 
						|
        Confirmation.objects.create(content_object=obj,
 | 
						|
                                    date_sent=now(),
 | 
						|
                                    confirmation_key=key,
 | 
						|
                                    type=Confirmation.EMAIL_CHANGE)
 | 
						|
        url = confirmation_url(key, user_profile.realm.host, Confirmation.EMAIL_CHANGE)
 | 
						|
        response = self.client_get(url)
 | 
						|
 | 
						|
        self.assertEqual(response.status_code, 200)
 | 
						|
        self.assert_in_success_response(["This confirms that the email address for your Zulip"],
 | 
						|
                                        response)
 | 
						|
        user_profile = get_user(new_email, new_realm)
 | 
						|
        self.assertTrue(bool(user_profile))
 | 
						|
        obj.refresh_from_db()
 | 
						|
        self.assertEqual(obj.status, 1)
 | 
						|
 | 
						|
    def test_start_email_change_process(self):
 | 
						|
        # type: () -> None
 | 
						|
        user_profile = self.example_user('hamlet')
 | 
						|
        do_start_email_change_process(user_profile, 'hamlet-new@zulip.com')
 | 
						|
        self.assertEqual(EmailChangeStatus.objects.count(), 1)
 | 
						|
 | 
						|
    def test_end_to_end_flow(self):
 | 
						|
        # type: () -> None
 | 
						|
        data = {'email': 'hamlet-new@zulip.com'}
 | 
						|
        email = self.example_email("hamlet")
 | 
						|
        self.login(email)
 | 
						|
        url = '/json/settings'
 | 
						|
        self.assertEqual(len(mail.outbox), 0)
 | 
						|
        result = self.client_patch(url, data)
 | 
						|
        self.assertEqual(len(mail.outbox), 1)
 | 
						|
        self.assert_in_success_response(['Check your email for a confirmation link.'], result)
 | 
						|
        email_message = mail.outbox[0]
 | 
						|
        self.assertEqual(
 | 
						|
            email_message.subject,
 | 
						|
            'Verify your new email address'
 | 
						|
        )
 | 
						|
        body = email_message.body
 | 
						|
        from_email = email_message.from_email
 | 
						|
        self.assertIn('We received a request to change the email', body)
 | 
						|
        self.assertIn('Zulip Account Security', from_email)
 | 
						|
        self.assertIn(FromAddress.NOREPLY, email_message.from_email)
 | 
						|
 | 
						|
        activation_url = [s for s in body.split('\n') if s][4]
 | 
						|
        response = self.client_get(activation_url)
 | 
						|
 | 
						|
        self.assert_in_success_response(["This confirms that the email address"],
 | 
						|
                                        response)
 | 
						|
 | 
						|
        # Now confirm trying to change your email back doesn't throw an immediate error
 | 
						|
        result = self.client_patch(url, {"email": "hamlet@zulip.com"})
 | 
						|
        self.assert_in_success_response(['Check your email for a confirmation link.'], result)
 | 
						|
 | 
						|
    def test_unauthorized_email_change(self):
 | 
						|
        # type: () -> None
 | 
						|
        data = {'email': 'hamlet-new@zulip.com'}
 | 
						|
        user_profile = self.example_user('hamlet')
 | 
						|
        email = user_profile.email
 | 
						|
        self.login(email)
 | 
						|
        do_set_realm_property(user_profile.realm, 'email_changes_disabled', True)
 | 
						|
        url = '/json/settings'
 | 
						|
        result = self.client_patch(url, data)
 | 
						|
        self.assertEqual(len(mail.outbox), 0)
 | 
						|
        self.assertEqual(result.status_code, 400)
 | 
						|
        self.assert_in_response("Email address changes are disabled in this organization.",
 | 
						|
                                result)
 | 
						|
 | 
						|
    def test_email_change_already_taken(self):
 | 
						|
        # type: () -> None
 | 
						|
        data = {'email': 'cordelia@zulip.com'}
 | 
						|
        user_profile = self.example_user('hamlet')
 | 
						|
        email = user_profile.email
 | 
						|
        self.login(email)
 | 
						|
 | 
						|
        url = '/json/settings'
 | 
						|
        result = self.client_patch(url, data)
 | 
						|
        self.assertEqual(len(mail.outbox), 0)
 | 
						|
        self.assertEqual(result.status_code, 400)
 | 
						|
        self.assert_in_response("Already has an account",
 | 
						|
                                result)
 | 
						|
 | 
						|
    def test_unauthorized_email_change_from_email_confirmation_link(self):
 | 
						|
        # type: () -> None
 | 
						|
        data = {'email': 'hamlet-new@zulip.com'}
 | 
						|
        user_profile = self.example_user('hamlet')
 | 
						|
        email = user_profile.email
 | 
						|
        self.login(email)
 | 
						|
        url = '/json/settings'
 | 
						|
        self.assertEqual(len(mail.outbox), 0)
 | 
						|
        result = self.client_patch(url, data)
 | 
						|
        self.assertEqual(len(mail.outbox), 1)
 | 
						|
        self.assert_in_success_response(['Check your email for a confirmation link.'], result)
 | 
						|
        email_message = mail.outbox[0]
 | 
						|
        self.assertEqual(
 | 
						|
            email_message.subject,
 | 
						|
            'Verify your new email address'
 | 
						|
        )
 | 
						|
        body = email_message.body
 | 
						|
        self.assertIn('We received a request to change the email', body)
 | 
						|
 | 
						|
        do_set_realm_property(user_profile.realm, 'email_changes_disabled', True)
 | 
						|
 | 
						|
        activation_url = [s for s in body.split('\n') if s][4]
 | 
						|
        response = self.client_get(activation_url)
 | 
						|
 | 
						|
        self.assertEqual(response.status_code, 400)
 | 
						|
        self.assert_in_response("Email address changes are disabled in this organization.",
 | 
						|
                                response)
 | 
						|
 | 
						|
    def test_post_invalid_email(self):
 | 
						|
        # type: () -> None
 | 
						|
        data = {'email': 'hamlet-new'}
 | 
						|
        email = self.example_email("hamlet")
 | 
						|
        self.login(email)
 | 
						|
        url = '/json/settings'
 | 
						|
        result = self.client_patch(url, data)
 | 
						|
        self.assert_in_response('Invalid address', result)
 | 
						|
 | 
						|
    def test_post_same_email(self):
 | 
						|
        # type: () -> None
 | 
						|
        data = {'email': self.example_email("hamlet")}
 | 
						|
        email = self.example_email("hamlet")
 | 
						|
        self.login(email)
 | 
						|
        url = '/json/settings'
 | 
						|
        result = self.client_patch(url, data)
 | 
						|
        self.assertEqual('success', result.json()['result'])
 | 
						|
        self.assertEqual('', result.json()['msg'])
 |