def test_change_delivery_email_end_to_end_with_admins_visibility(self) -> None:
        user_profile = self.example_user('hamlet')
        do_set_realm_property(user_profile.realm, 'email_address_visibility',
                              Realm.EMAIL_ADDRESS_VISIBILITY_ADMINS)

        old_email = user_profile.email
        new_email = '[email protected]'
        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_profile_by_id(user_profile.id)
        self.assertEqual(user_profile.delivery_email, new_email)
        self.assertEqual(user_profile.email, "[email protected]")
        obj.refresh_from_db()
        self.assertEqual(obj.status, 1)
        with self.assertRaises(UserProfile.DoesNotExist):
            get_user(old_email, user_profile.realm)
        with self.assertRaises(UserProfile.DoesNotExist):
            get_user_by_delivery_email(old_email, user_profile.realm)
        self.assertEqual(get_user_by_delivery_email(new_email, user_profile.realm), user_profile)

def create_user_backend(request: HttpRequest, user_profile: UserProfile,
                        email: Text=REQ(), password: Text=REQ(), full_name_raw: Text=REQ("full_name"),
                        short_name: Text=REQ()) -> HttpResponse:
    full_name = check_full_name(full_name_raw)
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        return json_error(_('Bad name or username'))

    # Check that the new user's email address belongs to the admin's realm
    # (Since this is an admin API, we don't require the user to have been
    # invited first.)
    realm = user_profile.realm
    try:
        email_allowed_for_realm(email, user_profile.realm)
    except DomainNotAllowedForRealmError:
        return json_error(_("Email '%(email)s' not allowed in this organization") %
                          {'email': email})
    except DisposableEmailError:
        return json_error(_("Disposable email addresses are not allowed in this organization"))

    try:
        get_user(email, user_profile.realm)
        return json_error(_("Email '%s' already in use") % (email,))
    except UserProfile.DoesNotExist:
        pass

    do_create_user(email, password, realm, full_name, short_name)
    return json_success()

def add_bot_backend(request, user_profile, full_name_raw=REQ("full_name"), short_name=REQ(),
                    default_sending_stream_name=REQ('default_sending_stream', default=None),
                    default_events_register_stream_name=REQ('default_events_register_stream', default=None),
                    default_all_public_streams=REQ(validator=check_bool, default=None)):
    # type: (HttpRequest, UserProfile, Text, Text, Optional[Text], Optional[Text], Optional[bool]) -> HttpResponse
    short_name += "-bot"
    full_name = check_full_name(full_name_raw)
    email = '%[email protected]%s' % (short_name, user_profile.realm.get_bot_domain())
    form = CreateUserForm({'full_name': full_name, 'email': email})
    if not form.is_valid():
        # We validate client-side as well
        return json_error(_('Bad name or username'))

    try:
        get_user(email, user_profile.realm)
        return json_error(_("Username already in use"))
    except UserProfile.DoesNotExist:
        pass

    if len(request.FILES) == 0:
        avatar_source = UserProfile.AVATAR_FROM_GRAVATAR
    elif len(request.FILES) != 1:
        return json_error(_("You may only upload one file at a time"))
    else:
        avatar_source = UserProfile.AVATAR_FROM_USER

    default_sending_stream = None
    if default_sending_stream_name is not None:
        (default_sending_stream, ignored_rec, ignored_sub) = access_stream_by_name(
            user_profile, default_sending_stream_name)

    default_events_register_stream = None
    if default_events_register_stream_name is not None:
        (default_events_register_stream, ignored_rec, ignored_sub) = access_stream_by_name(
            user_profile, default_events_register_stream_name)

    bot_profile = do_create_user(email=email, password='',
                                 realm=user_profile.realm, full_name=full_name,
                                 short_name=short_name, active=True,
                                 bot_type=UserProfile.DEFAULT_BOT,
                                 bot_owner=user_profile,
                                 avatar_source=avatar_source,
                                 default_sending_stream=default_sending_stream,
                                 default_events_register_stream=default_events_register_stream,
                                 default_all_public_streams=default_all_public_streams)
    if len(request.FILES) == 1:
        user_file = list(request.FILES.values())[0]
        upload_avatar_image(user_file, user_profile, bot_profile)
    json_result = dict(
        api_key=bot_profile.api_key,
        avatar_url=avatar_url(bot_profile),
        default_sending_stream=get_stream_name(bot_profile.default_sending_stream),
        default_events_register_stream=get_stream_name(bot_profile.default_events_register_stream),
        default_all_public_streams=bot_profile.default_all_public_streams,
    )
    return json_success(json_result)

 def _login(self, user_email, user_realm, password=None):
     # type: (Text, Realm, str) -> None
     if password:
         user_profile = get_user(user_email, user_realm)
         user_profile.set_password(password)
         user_profile.save()
     self.login(user_email, password)

    def test_get_accounts_for_email(self) -> None:
        def check_account_present_in_accounts(user: UserProfile, accounts: List[Dict[str, Optional[str]]]) -> None:
            for account in accounts:
                realm = user.realm
                if account["avatar"] == avatar_url(user) and account["full_name"] == user.full_name \
                        and account["realm_name"] == realm.name and account["string_id"] == realm.string_id:
                    return
            raise AssertionError("Account not found")

        lear_realm = get_realm("lear")
        cordelia_in_zulip = self.example_user("cordelia")
        cordelia_in_lear = get_user("[email protected]", lear_realm)

        email = "[email protected]"
        accounts = get_accounts_for_email(email)
        self.assert_length(accounts, 2)
        check_account_present_in_accounts(cordelia_in_zulip, accounts)
        check_account_present_in_accounts(cordelia_in_lear, accounts)

        email = "[email protected]"
        accounts = get_accounts_for_email(email)
        self.assert_length(accounts, 2)
        check_account_present_in_accounts(cordelia_in_zulip, accounts)
        check_account_present_in_accounts(cordelia_in_lear, accounts)

        email = "[email protected]"
        accounts = get_accounts_for_email(email)
        self.assert_length(accounts, 1)
        check_account_present_in_accounts(self.example_user("iago"), accounts)

    def test_file_download_authorization_public(self):
        # type: () -> None
        subscribed_users = [self.example_email("hamlet"), self.example_email("iago")]
        unsubscribed_users = [self.example_email("othello"), self.example_email("prospero")]
        realm = get_realm("zulip")
        for email in subscribed_users:
            self.subscribe(get_user(email, realm), "test-subscribe")

        self.login(self.example_email("hamlet"))
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        result = self.client_post("/json/user_uploads", {'file': fp})
        uri = result.json()['uri']
        fp_path_id = re.sub('/user_uploads/', '', uri)
        body = "First message ...[zulip.txt](http://localhost:9991/user_uploads/" + fp_path_id + ")"
        self.send_stream_message(self.example_email("hamlet"), "test-subscribe", body, "test")
        self.logout()

        # Now all users should be able to access the files
        for user in subscribed_users + unsubscribed_users:
            self.login(user)
            response = self.client_get(uri)
            data = b"".join(response.streaming_content)
            self.assertEqual(b"zulip!", data)
            self.logout()

def common_get_active_user(email: str, realm: Realm,
                           return_data: Dict[str, Any]=None) -> Optional[UserProfile]:
    try:
        user_profile = get_user(email, realm)
    except UserProfile.DoesNotExist:
        # If the user doesn't have an account in the target realm, we
        # check whether they might have an account in another realm,
        # and if so, provide a helpful error message via
        # `invalid_subdomain`.
        if not UserProfile.objects.filter(email__iexact=email).exists():
            return None
        if return_data is not None:
            return_data['invalid_subdomain'] = True
        return None
    if not user_profile.is_active:
        if return_data is not None:
            if user_profile.is_mirror_dummy:
                # Record whether it's a mirror dummy account
                return_data['is_mirror_dummy'] = True
            return_data['inactive_user'] = True
        return None
    if user_profile.realm.deactivated:
        if return_data is not None:
            return_data['inactive_realm'] = True
        return None
    return user_profile

    def test_confirm_email_change(self) -> None:
        user_profile = self.example_user('hamlet')
        old_email = user_profile.email
        new_email = '[email protected]'
        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 get_presence_backend(request: HttpRequest, user_profile: UserProfile,
                         email: Text) -> HttpResponse:
    try:
        target = get_user(email, user_profile.realm)
    except UserProfile.DoesNotExist:
        return json_error(_('No such user'))
    if not target.is_active:
        return json_error(_('No such user'))
    if target.is_bot:
        return json_error(_('Presence is not supported for bot users.'))

    presence_dict = UserPresence.get_status_dict_by_user(target)
    if len(presence_dict) == 0:
        return json_error(_('No presence data for %s' % (target.email,)))

    # For initial version, we just include the status and timestamp keys
    result = dict(presence=presence_dict[target.email])
    aggregated_info = result['presence']['aggregated']
    aggr_status_duration = datetime_to_timestamp(timezone_now()) - aggregated_info['timestamp']
    if aggr_status_duration > settings.OFFLINE_THRESHOLD_SECS:
        aggregated_info['status'] = 'offline'
    for val in result['presence'].values():
        val.pop('client', None)
        val.pop('pushable', None)
    return json_success(result)

    def test_create_outgoing_webhook_bot(self, **extras: Any) -> None:
        self.login(self.example_email('hamlet'))
        bot_info = {
            'full_name': 'Outgoing Webhook test bot',
            'short_name': 'outgoingservicebot',
            'bot_type': UserProfile.OUTGOING_WEBHOOK_BOT,
            'payload_url': ujson.dumps('http://127.0.0.1:5002/bots/followup'),
        }
        bot_info.update(extras)
        result = self.client_post("/json/bots", bot_info)
        self.assert_json_success(result)

        bot_email = "[email protected]"
        bot_realm = get_realm('zulip')
        bot = get_user(bot_email, bot_realm)
        services = get_bot_services(bot.id)
        service = services[0]

        self.assertEqual(len(services), 1)
        self.assertEqual(service.name, "outgoingservicebot")
        self.assertEqual(service.base_url, "http://127.0.0.1:5002/bots/followup")
        self.assertEqual(service.user_profile, bot)

        # invalid URL test case.
        bot_info['payload_url'] = ujson.dumps('http://127.0.0.:5002/bots/followup')
        result = self.client_post("/json/bots", bot_info)
        self.assert_json_error(result, "Enter a valid URL.")

    def test_notify_bot_owner_on_invalid_json(self) -> None:
        @api_key_only_webhook_view('ClientName', notify_bot_owner_on_invalid_json=False)
        def my_webhook_no_notify(request: HttpRequest, user_profile: UserProfile) -> None:
            raise InvalidJSONError("Malformed JSON")

        @api_key_only_webhook_view('ClientName', notify_bot_owner_on_invalid_json=True)
        def my_webhook_notify(request: HttpRequest, user_profile: UserProfile) -> None:
            raise InvalidJSONError("Malformed JSON")

        webhook_bot_email = '[email protected]'
        webhook_bot_realm = get_realm('zulip')
        webhook_bot = get_user(webhook_bot_email, webhook_bot_realm)
        webhook_bot_api_key = get_api_key(webhook_bot)
        request = HostRequestMock()
        request.POST['api_key'] = webhook_bot_api_key
        request.host = "zulip.testserver"
        expected_msg = INVALID_JSON_MESSAGE.format(webhook_name='ClientName')

        last_message_id = self.get_last_message().id
        with self.assertRaisesRegex(JsonableError, "Malformed JSON"):
            my_webhook_no_notify(request)  # type: ignore # mypy doesn't seem to apply the decorator

        # First verify that without the setting, it doesn't send a PM to bot owner.
        msg = self.get_last_message()
        self.assertEqual(msg.id, last_message_id)
        self.assertNotEqual(msg.content, expected_msg.strip())

        # Then verify that with the setting, it does send such a message.
        with self.assertRaisesRegex(JsonableError, "Malformed JSON"):
            my_webhook_notify(request)  # type: ignore # mypy doesn't seem to apply the decorator
        msg = self.get_last_message()
        self.assertNotEqual(msg.id, last_message_id)
        self.assertEqual(msg.sender.email, self.notification_bot().email)
        self.assertEqual(msg.content, expected_msg.strip())

def get_assignee_mention(assignee_email: str, realm: Realm) -> str:
    if assignee_email != '':
        try:
            assignee_name = get_user(assignee_email, realm).full_name
        except UserProfile.DoesNotExist:
            assignee_name = assignee_email
        return u"**{}**".format(assignee_name)
    return ''

def deactivate_bot_backend(request: HttpRequest, user_profile: UserProfile,
                           email: Text) -> HttpResponse:
    try:
        target = get_user(email, user_profile.realm)
    except UserProfile.DoesNotExist:
        return json_error(_('No such bot'))
    if not target.is_bot:
        return json_error(_('No such bot'))
    return _deactivate_user_profile_backend(request, user_profile, target)

def principal_to_user_profile(agent: UserProfile, principal: Text) -> UserProfile:
    try:
        return get_user(principal, agent.realm)
    except UserProfile.DoesNotExist:
        # We have to make sure we don't leak information about which users
        # are registered for Zulip in a different realm.  We could do
        # something a little more clever and check the domain part of the
        # principal to maybe give a better error message
        raise PrincipalError(principal)

 def get_streams(self, email: str, realm: Realm) -> List[str]:
     """
     Helper function to get the stream names for a user
     """
     user_profile = get_user(email, realm)
     subs = get_stream_subscriptions_for_user(user_profile).filter(
         active=True,
     )
     return [cast(str, get_display_recipient(sub.recipient)) for sub in subs]

def api_yo_app_webhook(request: HttpRequest, user_profile: UserProfile,
                       email: str = REQ(default=""),
                       username: str = REQ(default='Yo Bot'),
                       topic: Optional[str] = REQ(default=None),
                       user_ip: Optional[str] = REQ(default=None)) -> HttpResponse:
    body = ('Yo from %s') % (username,)
    receiving_user = get_user(email, user_profile.realm)
    check_send_private_message(user_profile, request.client, receiving_user, body)
    return json_success()

    def test_webhook_http_header_header_exists(self) -> None:
        webhook_bot = get_user('[email protected]', get_realm('zulip'))
        request = HostRequestMock()
        request.META['HTTP_X_CUSTOM_HEADER'] = 'custom_value'
        request.user = webhook_bot

        header_value = validate_extract_webhook_http_header(request, 'X_CUSTOM_HEADER',
                                                            'test_webhook')

        self.assertEqual(header_value, 'custom_value')

    def test_create_embedded_bot(self, **extras: Any) -> None:
        self.login(self.example_email('hamlet'))

        # Test to create embedded bot with correct service_name
        bot_config_info = {'key': 'value'}
        bot_info = {
            'full_name': 'Embedded test bot',
            'short_name': 'embeddedservicebot',
            'bot_type': UserProfile.EMBEDDED_BOT,
            'service_name': 'followup',
            'config_data': ujson.dumps(bot_config_info),
        }
        bot_info.update(extras)
        with self.settings(EMBEDDED_BOTS_ENABLED=False):
            result = self.client_post("/json/bots", bot_info)
            self.assert_json_error(result, 'Embedded bots are not enabled.')

        result = self.client_post("/json/bots", bot_info)
        self.assert_json_success(result)

        bot_email = "[email protected]"
        bot_realm = get_realm('zulip')
        bot = get_user(bot_email, bot_realm)
        services = get_bot_services(bot.id)
        service = services[0]
        bot_config = get_bot_config(bot)
        self.assertEqual(bot_config, bot_config_info)
        self.assertEqual(len(services), 1)
        self.assertEqual(service.name, "followup")
        self.assertEqual(service.user_profile, bot)

        # Test to create embedded bot with incorrect service_name
        bot_info = {
            'full_name': 'Embedded test bot',
            'short_name': 'embeddedservicebot',
            'bot_type': UserProfile.EMBEDDED_BOT,
            'service_name': 'not_existing_service',
        }
        bot_info.update(extras)
        result = self.client_post("/json/bots", bot_info)
        self.assert_json_error(result, 'Invalid embedded bot name.')

        # Test to create embedded bot with an invalid config value
        malformatted_bot_config_info = {'foo': ['bar', 'baz']}
        bot_info = {
            'full_name': 'Embedded test bot',
            'short_name': 'embeddedservicebot2',
            'bot_type': UserProfile.EMBEDDED_BOT,
            'service_name': 'followup',
            'config_data': ujson.dumps(malformatted_bot_config_info)
        }
        bot_info.update(extras)
        result = self.client_post("/json/bots", bot_info)
        self.assert_json_error(result, 'config_data contains a value that is not a string')

    def send_personal_message(self, from_email: str, to_email: str, content: str="test content",
                              sender_realm: str="zulip") -> int:
        sender = get_user(from_email, get_realm(sender_realm))

        recipient_list = [to_email]
        (sending_client, _) = Client.objects.get_or_create(name="test suite")

        return check_send_message(
            sender, sending_client, 'private', recipient_list, None,
            content
        )

    def test_add_bot_with_default_all_public_streams(self) -> None:
        self.login(self.example_email('hamlet'))
        self.assert_num_bots_equal(0)
        result = self.create_bot(default_all_public_streams=ujson.dumps(True))
        self.assert_num_bots_equal(1)
        self.assertTrue(result['default_all_public_streams'])

        bot_email = '[email protected]'
        bot_realm = get_realm('zulip')
        profile = get_user(bot_email, bot_realm)
        self.assertEqual(profile.default_all_public_streams, True)