def test_change_signup_notifications_stream(self) -> None:
        # We need an admin user.
        email = '[email protected]'
        self.login(email)

        disabled_signup_notifications_stream_id = -1
        req = dict(signup_notifications_stream_id = ujson.dumps(disabled_signup_notifications_stream_id))
        result = self.client_patch('/json/realm', req)
        self.assert_json_success(result)
        realm = get_realm('zulip')
        self.assertEqual(realm.signup_notifications_stream, None)

        new_signup_notifications_stream_id = 4
        req = dict(signup_notifications_stream_id = ujson.dumps(new_signup_notifications_stream_id))

        result = self.client_patch('/json/realm', req)
        self.assert_json_success(result)
        realm = get_realm('zulip')
        self.assertEqual(realm.signup_notifications_stream.id, new_signup_notifications_stream_id)

        invalid_signup_notifications_stream_id = 1234
        req = dict(signup_notifications_stream_id = ujson.dumps(invalid_signup_notifications_stream_id))
        result = self.client_patch('/json/realm', req)
        self.assert_json_error(result, 'Invalid stream id')
        realm = get_realm('zulip')
        self.assertNotEqual(realm.signup_notifications_stream.id, invalid_signup_notifications_stream_id)

    def test_valid_user_id(self) -> None:
        realm = get_realm("zulip")
        hamlet = self.example_user('hamlet')
        othello = self.example_user('othello')
        bot = self.example_user("welcome_bot")

        # Invalid user ID
        invalid_uid = 1000
        self.assertEqual(check_valid_user_id(realm.id, invalid_uid),
                         "Invalid user ID: %d" % (invalid_uid))
        self.assertEqual(check_valid_user_id(realm.id, "abc"),
                         "User id is not an integer")
        self.assertEqual(check_valid_user_id(realm.id, str(othello.id)),
                         "User id is not an integer")

        # User is in different realm
        self.assertEqual(check_valid_user_id(get_realm("zephyr").id, hamlet.id),
                         "Invalid user ID: %d" % (hamlet.id))

        # User is not active
        hamlet.is_active = False
        hamlet.save()
        self.assertEqual(check_valid_user_id(realm.id, hamlet.id),
                         "User is deactivated")
        self.assertEqual(check_valid_user_id(realm.id, hamlet.id, allow_deactivated=True),
                         None)

        # User is bot
        self.assertEqual(check_valid_user_id(realm.id, bot.id),
                         "User with id %d is bot" % (bot.id))

        # Succesfully get non-bot, active user belong to your realm
        self.assertEqual(check_valid_user_id(realm.id, othello.id), None)

    def test_send_deactivated_realm(self):
        """
        rest_dispatch rejects requests in a deactivated realm, both /json and api

        """
        realm = get_realm("zulip.com")
        do_deactivate_realm(get_realm("zulip.com"))

        result = self.client_post("/json/messages", {"type": "private",
                                                     "content": "Test message",
                                                     "client": "test suite",
                                                     "to": "[email protected]"})
        self.assert_json_error_contains(result, "Not logged in", status_code=401)

        # Even if a logged-in session was leaked, it still wouldn't work
        realm.deactivated = False
        realm.save()
        self.login("[email protected]")
        realm.deactivated = True
        realm.save()

        result = self.client_post("/json/messages", {"type": "private",
                                                     "content": "Test message",
                                                     "client": "test suite",
                                                     "to": "[email protected]"})
        self.assert_json_error_contains(result, "has been deactivated", status_code=400)

        result = self.client_post("/api/v1/messages", {"type": "private",
                                                       "content": "Test message",
                                                       "client": "test suite",
                                                       "to": "[email protected]"},
                                  **self.api_auth("[email protected]"))
        self.assert_json_error_contains(result, "has been deactivated", status_code=401)

 def test_delete_all_user_sessions(self) -> None:
     self.do_test_session(self.example_email("hamlet"),
                          lambda: delete_all_user_sessions(),
                          get_realm("zulip"), True)
     self.do_test_session(self.mit_email("sipbtest"),
                          lambda: delete_all_user_sessions(),
                          get_realm("zephyr"), True)

    def test_change_video_chat_provider(self) -> None:
        self.assertEqual(get_realm('zulip').video_chat_provider, "Jitsi")
        email = self.example_email("iago")
        self.login(email)

        req = {"video_chat_provider": ujson.dumps("Google Hangouts")}
        result = self.client_patch('/json/realm', req)
        self.assert_json_error(result, "Invalid domain: Domain can't be empty.")

        req = {
            "video_chat_provider": ujson.dumps("Google Hangouts"),
            "google_hangouts_domain": ujson.dumps("invaliddomain"),
        }
        result = self.client_patch('/json/realm', req)
        self.assert_json_error(result, "Invalid domain: Domain must have at least one dot (.)")

        req = {
            "video_chat_provider": ujson.dumps("Google Hangouts"),
            "google_hangouts_domain": ujson.dumps("zulip.com"),
        }
        result = self.client_patch('/json/realm', req)
        self.assert_json_success(result)
        self.assertEqual(get_realm('zulip').video_chat_provider, "Google Hangouts")

        req = {"video_chat_provider": ujson.dumps("Jitsi")}
        result = self.client_patch('/json/realm', req)
        self.assert_json_success(result)
        self.assertEqual(get_realm('zulip').video_chat_provider, "Jitsi")

    def test_create_realm_domain(self) -> None:
        self.login(self.example_email("iago"))
        data = {'domain': ujson.dumps(''),
                'allow_subdomains': ujson.dumps(True)}
        result = self.client_post("/json/realm/domains", info=data)
        self.assert_json_error(result, 'Invalid domain: Domain can\'t be empty.')

        data['domain'] = ujson.dumps('acme.com')
        result = self.client_post("/json/realm/domains", info=data)
        self.assert_json_success(result)
        realm = get_realm('zulip')
        self.assertTrue(RealmDomain.objects.filter(realm=realm, domain='acme.com',
                                                   allow_subdomains=True).exists())

        result = self.client_post("/json/realm/domains", info=data)
        self.assert_json_error(result, 'The domain acme.com is already a part of your organization.')

        mit_user_profile = self.mit_user("sipbtest")
        self.login(mit_user_profile.email, realm=get_realm("zephyr"))

        do_change_is_admin(mit_user_profile, True)

        result = self.client_post("/json/realm/domains", info=data,
                                  HTTP_HOST=mit_user_profile.realm.host)
        self.assert_json_success(result)

    def test_realm_message_content_allowed_in_email_notifications(self) -> None:
        user = self.example_user("hamlet")

        # When message content is allowed at realm level
        realm = get_realm("zulip")
        realm.message_content_allowed_in_email_notifications = True
        realm.save(update_fields=['message_content_allowed_in_email_notifications'])

        # Emails have missed message content when message content is enabled by the user
        do_change_notification_settings(user, "message_content_in_email_notifications", True)
        mail.outbox = []
        self._extra_context_in_personal_missed_stream_messages(False, show_message_content=True)

        # Emails don't have missed message content when message content is disabled by the user
        do_change_notification_settings(user, "message_content_in_email_notifications", False)
        mail.outbox = []
        self._extra_context_in_personal_missed_stream_messages(False, show_message_content=False)

        # When message content is not allowed at realm level
        # Emails don't have missed message irrespective of message content setting of the user
        realm = get_realm("zulip")
        realm.message_content_allowed_in_email_notifications = False
        realm.save(update_fields=['message_content_allowed_in_email_notifications'])

        do_change_notification_settings(user, "message_content_in_email_notifications", True)
        mail.outbox = []
        self._extra_context_in_personal_missed_stream_messages(False, show_message_content=False)

        do_change_notification_settings(user, "message_content_in_email_notifications", False)
        mail.outbox = []
        self._extra_context_in_personal_missed_stream_messages(False, show_message_content=False)

    def test_get_user_by_id_in_realm_including_cross_realm(self) -> None:
        realm = get_realm('zulip')
        hamlet = self.example_user('hamlet')
        othello = self.example_user('othello')
        bot = self.example_user('welcome_bot')

        # Pass in the ID of a cross-realm bot and a valid realm
        cross_realm_bot = get_user_by_id_in_realm_including_cross_realm(
            bot.id, realm)
        self.assertEqual(cross_realm_bot.email, bot.email)
        self.assertEqual(cross_realm_bot.id, bot.id)

        # Pass in the ID of a cross-realm bot but with a invalid realm,
        # note that the realm should be irrelevant here
        cross_realm_bot = get_user_by_id_in_realm_including_cross_realm(
            bot.id, get_realm('invalid'))
        self.assertEqual(cross_realm_bot.email, bot.email)
        self.assertEqual(cross_realm_bot.id, bot.id)

        # Pass in the ID of a non-cross-realm user with a realm
        user_profile = get_user_by_id_in_realm_including_cross_realm(
            othello.id, realm)
        self.assertEqual(user_profile.email, othello.email)
        self.assertEqual(user_profile.id, othello.id)

        # If the realm doesn't match, or if the ID is not that of a
        # cross-realm bot, UserProfile.DoesNotExist is raised
        with self.assertRaises(UserProfile.DoesNotExist):
            get_user_by_id_in_realm_including_cross_realm(
                hamlet.id, get_realm('invalid'))

 def test_delete_realm_user_sessions(self) -> None:
     realm = get_realm('zulip')
     self.do_test_session(self.example_email("hamlet"),
                          lambda: delete_realm_user_sessions(realm),
                          get_realm("zulip"), True)
     self.do_test_session(self.mit_email("sipbtest"),
                          lambda: delete_realm_user_sessions(realm),
                          get_realm("zephyr"), False)

 def test_delete_user_sessions(self) -> None:
     user_profile = self.example_user('hamlet')
     email = user_profile.email
     self.do_test_session(str(email), lambda: delete_user_sessions(user_profile),
                          get_realm("zulip"), True)
     self.do_test_session(str(self.example_email("othello")),
                          lambda: delete_user_sessions(user_profile),
                          get_realm("zulip"), False)

    def test_upgrade_where_subscription_save_fails_at_first(
            self, mock5: Mock, mock4: Mock, mock3: Mock, mock2: Mock, mock1: Mock) -> None:
        user = self.example_user("hamlet")
        self.login(user.email)
        # From https://stripe.com/docs/testing#cards: Attaching this card to
        # a Customer object succeeds, but attempts to charge the customer fail.
        self.client_post("/upgrade/", {'stripeToken': stripe_create_token('4000000000000341').id,
                                       'signed_seat_count': self.signed_seat_count,
                                       'salt': self.salt,
                                       'plan': Plan.CLOUD_ANNUAL})
        # Check that we created a Customer object with has_billing_relationship False
        customer = Customer.objects.get(realm=get_realm('zulip'))
        self.assertFalse(customer.has_billing_relationship)
        original_stripe_customer_id = customer.stripe_customer_id
        # Check that we created a customer in stripe, with no subscription
        stripe_customer = stripe_get_customer(customer.stripe_customer_id)
        self.assertFalse(extract_current_subscription(stripe_customer))
        # Check that we correctly populated RealmAuditLog
        audit_log_entries = list(RealmAuditLog.objects.filter(acting_user=user)
                                 .values_list('event_type', flat=True).order_by('id'))
        self.assertEqual(audit_log_entries, [RealmAuditLog.STRIPE_CUSTOMER_CREATED,
                                             RealmAuditLog.STRIPE_CARD_CHANGED])
        # Check that we did not update Realm
        realm = get_realm("zulip")
        self.assertFalse(realm.has_seat_based_plan)
        # Check that we still get redirected to /upgrade
        response = self.client_get("/billing/")
        self.assertEqual(response.status_code, 302)
        self.assertEqual('/upgrade/', response.url)

        # Try again, with a valid card
        self.client_post("/upgrade/", {'stripeToken': stripe_create_token().id,
                                       'signed_seat_count': self.signed_seat_count,
                                       'salt': self.salt,
                                       'plan': Plan.CLOUD_ANNUAL})
        customer = Customer.objects.get(realm=get_realm('zulip'))
        # Impossible to create two Customers, but check that we didn't
        # change stripe_customer_id and that we updated has_billing_relationship
        self.assertEqual(customer.stripe_customer_id, original_stripe_customer_id)
        self.assertTrue(customer.has_billing_relationship)
        # Check that we successfully added a subscription
        stripe_customer = stripe_get_customer(customer.stripe_customer_id)
        self.assertTrue(extract_current_subscription(stripe_customer))
        # Check that we correctly populated RealmAuditLog
        audit_log_entries = list(RealmAuditLog.objects.filter(acting_user=user)
                                 .values_list('event_type', flat=True).order_by('id'))
        self.assertEqual(audit_log_entries, [RealmAuditLog.STRIPE_CUSTOMER_CREATED,
                                             RealmAuditLog.STRIPE_CARD_CHANGED,
                                             RealmAuditLog.STRIPE_CARD_CHANGED,
                                             RealmAuditLog.STRIPE_PLAN_CHANGED,
                                             RealmAuditLog.REALM_PLAN_TYPE_CHANGED])
        # Check that we correctly updated Realm
        realm = get_realm("zulip")
        self.assertTrue(realm.has_seat_based_plan)
        # Check that we can no longer access /upgrade
        response = self.client_get("/upgrade/")
        self.assertEqual(response.status_code, 302)
        self.assertEqual('/billing/', response.url)

 def test_realm_icon_version(self) -> None:
     self.login(self.example_email("iago"))
     realm = get_realm('zulip')
     icon_version = realm.icon_version
     self.assertEqual(icon_version, 1)
     with get_test_image_file(self.correct_files[0][0]) as fp:
         self.client_post("/json/realm/icon", {'file': fp})
     realm = get_realm('zulip')
     self.assertEqual(realm.icon_version, icon_version + 1)

 def test_realm_reactivation_link(self) -> None:
     realm = get_realm('zulip')
     do_deactivate_realm(realm)
     self.assertTrue(realm.deactivated)
     confirmation_url = create_confirmation_link(realm, realm.host, Confirmation.REALM_REACTIVATION)
     response = self.client_get(confirmation_url)
     self.assert_in_success_response(['Your organization has been successfully reactivated'], response)
     realm = get_realm('zulip')
     self.assertFalse(realm.deactivated)

    def test_upgrade_where_subscription_save_fails_at_first(self, create_customer: mock.Mock) -> None:
        user = self.example_user("hamlet")
        self.login(user.email)
        with mock.patch('stripe.Subscription.create',
                        side_effect=stripe.error.CardError('message', 'param', 'code', json_body={})):
            self.client_post("/upgrade/", {'stripeToken': self.token,
                                           'signed_seat_count': self.signed_seat_count,
                                           'salt': self.salt,
                                           'plan': Plan.CLOUD_ANNUAL})
        # Check that we created a customer in stripe
        create_customer.assert_called()
        create_customer.reset_mock()
        # Check that we created a Customer with has_billing_relationship=False
        self.assertTrue(Customer.objects.filter(
            stripe_customer_id=self.stripe_customer_id, has_billing_relationship=False).exists())
        # Check that we correctly populated RealmAuditLog
        audit_log_entries = list(RealmAuditLog.objects.filter(acting_user=user)
                                 .values_list('event_type', flat=True).order_by('id'))
        self.assertEqual(audit_log_entries, [RealmAuditLog.STRIPE_CUSTOMER_CREATED,
                                             RealmAuditLog.STRIPE_CARD_ADDED])
        # Check that we did not update Realm
        realm = get_realm("zulip")
        self.assertFalse(realm.has_seat_based_plan)
        # Check that we still get redirected to /upgrade
        response = self.client_get("/billing/")
        self.assertEqual(response.status_code, 302)
        self.assertEqual('/upgrade/', response.url)

        # mock_create_customer just returns a customer with no subscription object
        with mock.patch("stripe.Subscription.create", side_effect=mock_customer_with_subscription):
            with mock.patch("stripe.Customer.retrieve", side_effect=mock_create_customer):
                with mock.patch("stripe.Customer.save", side_effect=mock_create_customer):
                    self.client_post("/upgrade/", {'stripeToken': self.token,
                                                   'signed_seat_count': self.signed_seat_count,
                                                   'salt': self.salt,
                                                   'plan': Plan.CLOUD_ANNUAL})
        # Check that we do not create a new customer in stripe
        create_customer.assert_not_called()
        # Impossible to create two Customers, but check that we updated has_billing_relationship
        self.assertTrue(Customer.objects.filter(
            stripe_customer_id=self.stripe_customer_id, has_billing_relationship=True).exists())
        # Check that we correctly populated RealmAuditLog
        audit_log_entries = list(RealmAuditLog.objects.filter(acting_user=user)
                                 .values_list('event_type', flat=True).order_by('id'))
        self.assertEqual(audit_log_entries, [RealmAuditLog.STRIPE_CUSTOMER_CREATED,
                                             RealmAuditLog.STRIPE_CARD_ADDED,
                                             RealmAuditLog.STRIPE_CARD_ADDED,
                                             RealmAuditLog.STRIPE_PLAN_CHANGED,
                                             RealmAuditLog.REALM_PLAN_TYPE_CHANGED])
        # Check that we correctly updated Realm
        realm = get_realm("zulip")
        self.assertTrue(realm.has_seat_based_plan)
        # Check that we can no longer access /upgrade
        response = self.client_get("/upgrade/")
        self.assertEqual(response.status_code, 302)
        self.assertEqual('/billing/', response.url)

    def test_initial_plan_type(self) -> None:
        with self.settings(BILLING_ENABLED=True):
            self.assertEqual(do_create_realm('hosted', 'hosted').plan_type, Realm.LIMITED)
            self.assertEqual(get_realm("hosted").max_invites, settings.INVITES_DEFAULT_REALM_DAILY_MAX)
            self.assertEqual(get_realm("hosted").message_visibility_limit, Realm.MESSAGE_VISIBILITY_LIMITED)

        with self.settings(BILLING_ENABLED=False):
            self.assertEqual(do_create_realm('onpremise', 'onpremise').plan_type, Realm.SELF_HOSTED)
            self.assertEqual(get_realm('onpremise').max_invites, settings.INVITES_DEFAULT_REALM_DAILY_MAX)
            self.assertEqual(get_realm('onpremise').message_visibility_limit, None)

    def test_deactivate_realm_by_non_admin(self) -> None:
        email = self.example_email('hamlet')
        self.login(email)
        realm = get_realm('zulip')
        self.assertFalse(realm.deactivated)

        result = self.client_post('/json/realm/deactivate')
        self.assert_json_error(result, "Must be an organization administrator")
        realm = get_realm('zulip')
        self.assertFalse(realm.deactivated)

    def test_deactivate_realm_by_admin(self) -> None:
        email = self.example_email('iago')
        self.login(email)
        realm = get_realm('zulip')
        self.assertFalse(realm.deactivated)

        result = self.client_post('/json/realm/deactivate')
        self.assert_json_success(result)
        realm = get_realm('zulip')
        self.assertTrue(realm.deactivated)

 def test_do_set_realm_name_caching(self) -> None:
     """The main complicated thing about setting realm names is fighting the
     cache, and we start by populating the cache for Hamlet, and we end
     by checking the cache to ensure that the new value is there."""
     self.example_user('hamlet')
     realm = get_realm('zulip')
     new_name = u'Zed You Elle Eye Pea'
     do_set_realm_property(realm, 'name', new_name)
     self.assertEqual(get_realm(realm.string_id).name, new_name)
     self.assert_user_profile_cache_gets_new_name(self.example_user('hamlet'), new_name)

    def test_user_ids_to_users(self) -> None:
        real_user_ids = [
            self.example_user('hamlet').id,
            self.example_user('cordelia').id,
        ]

        user_ids_to_users(real_user_ids, get_realm("zulip"))
        with self.assertRaises(JsonableError):
            user_ids_to_users([1234], get_realm("zephyr"))
        with self.assertRaises(JsonableError):
            user_ids_to_users(real_user_ids, get_realm("zephyr"))

    def test_user_ids_to_users(self) -> None:
        real_user_ids = [
            self.example_user('hamlet').id,
            self.example_user('cordelia').id,
        ]

        self.assertEqual(user_ids_to_users([], get_realm("zulip")), [])
        self.assertEqual(set([user_profile.id for user_profile in user_ids_to_users(real_user_ids, get_realm("zulip"))]),
                         set(real_user_ids))
        with self.assertRaises(JsonableError):
            user_ids_to_users([1234], get_realm("zephyr"))
        with self.assertRaises(JsonableError):
            user_ids_to_users(real_user_ids, get_realm("zephyr"))