def deactivate_user_own_backend(request, user_profile):
    # type: (HttpRequest, UserProfile) -> HttpResponse

    if user_profile.is_realm_admin and check_last_admin(user_profile):
        return json_error(_('Cannot deactivate the only organization administrator'))
    do_deactivate_user(user_profile)
    return json_success()

def _deactivate_user_profile_backend(request, user_profile, target):
    # type: (HttpRequest, UserProfile, UserProfile) -> HttpResponse
    if not user_profile.can_admin_user(target):
        return json_error(_('Insufficient permission'))

    do_deactivate_user(target)
    return json_success()

    def handle(self, *args, **options):
        # type: (*Any, **Any) -> None
        realm = self.get_realm(options)
        user_profile = self.get_user(options['email'], realm)

        print("Deactivating %s (%s) - %s" % (user_profile.full_name,
                                             user_profile.email,
                                             user_profile.realm.string_id))
        print("%s has the following active sessions:" % (user_profile.email,))
        for session in user_sessions(user_profile):
            print(session.expire_date, session.get_decoded())
        print("")
        print("%s has %s active bots that will also be deactivated." % (
            user_profile.email,
            UserProfile.objects.filter(
                is_bot=True, is_active=True, bot_owner=user_profile
            ).count()
        ))

        if not options["for_real"]:
            print("This was a dry run. Pass -f to actually deactivate.")
            exit(1)

        do_deactivate_user(user_profile)
        print("Sessions deleted, user deactivated.")

 def test_inactive_user(self):
     # type: () -> None
     do_deactivate_user(self.user_profile)
     result = self.client_post("/api/v1/fetch_api_key",
                               dict(username=self.email,
                                    password=initial_password(self.email)))
     self.assert_json_error_contains(result, "Your account has been disabled", 403)

    def test_access_user_by_id(self) -> None:
        iago = self.example_user("iago")

        # Must be a valid user ID in the realm
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, 1234)
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, self.mit_user("sipbtest").id)

        # Can only access bot users if allow_deactivated is passed
        bot = self.example_user("welcome_bot")
        access_user_by_id(iago, bot.id, allow_bots=True)
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, bot.id)

        # Can only access deactivated users if allow_deactivated is passed
        hamlet = self.example_user("hamlet")
        do_deactivate_user(hamlet)
        with self.assertRaises(JsonableError):
            access_user_by_id(iago, hamlet.id)
        access_user_by_id(iago, hamlet.id, allow_deactivated=True)

        # Non-admin user can't admin another user
        with self.assertRaises(JsonableError):
            access_user_by_id(self.example_user("cordelia"), self.example_user("aaron").id)

    def test_send_deactivated_user(self):
        """
        rest_dispatch rejects requests from deactivated users, both /json and api

        """
        email = "[email protected]"
        user_profile = get_user_profile_by_email(email)
        self.login(email)
        do_deactivate_user(user_profile)

        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
        do_reactivate_user(user_profile)
        self.login(email)
        user_profile.is_active = False
        user_profile.save()

        result = self.client_post("/json/messages", {"type": "private",
                                                     "content": "Test message",
                                                     "client": "test suite",
                                                     "to": "[email protected]"})
        self.assert_json_error_contains(result, "Account not active", 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, "Account not active", status_code=401)

 def test_clear_scheduled_jobs(self) -> None:
     user = self.example_user('hamlet')
     send_future_email('zerver/emails/followup_day1', user.realm,
                       to_user_ids=[user.id], delay=datetime.timedelta(hours=1))
     self.assertEqual(ScheduledEmail.objects.count(), 1)
     do_deactivate_user(user)
     self.assertEqual(ScheduledEmail.objects.count(), 0)

def deactivate_user_own_backend(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
    if UserProfile.objects.filter(realm=user_profile.realm, is_active=True).count() == 1:
        raise CannotDeactivateLastUserError(is_last_admin=False)
    if user_profile.is_realm_admin and check_last_admin(user_profile):
        raise CannotDeactivateLastUserError(is_last_admin=True)

    do_deactivate_user(user_profile, acting_user=user_profile)
    return json_success()

def deactivate_user_own_backend(request, user_profile):
    # type: (HttpRequest, UserProfile) -> HttpResponse
    admins = set(user_profile.realm.get_admin_users())

    if user_profile.is_realm_admin and len(admins) == 1:
        return json_error(_('Cannot deactivate the only admin'))
    do_deactivate_user(user_profile)
    return json_success()

def sync_user_from_ldap(user_profile: UserProfile) -> bool:
    backend = ZulipLDAPUserPopulator()
    updated_user = backend.populate_user(backend.django_to_ldap_username(user_profile.email))
    if not updated_user:
        if settings.LDAP_DEACTIVATE_NON_MATCHING_USERS:
            do_deactivate_user(user_profile)
        return False
    return True

    def test_login_deactivated_user(self):
        """
        logging in fails with an inactive user

        """
        email = "[email protected]"
        user_profile = get_user_profile_by_email(email)
        do_deactivate_user(user_profile)

        result = self.login_with_return("[email protected]")
        self.assert_in_response("Please enter a correct email and password", result)

    def test_login_deactivated_user(self):
        # type: () -> None
        """
        logging in fails with an inactive user

        """
        user_profile = self.example_user('hamlet')
        do_deactivate_user(user_profile)

        result = self.login_with_return(self.example_email("hamlet"))
        self.assert_in_response(
            "Sorry for the trouble, but your account has been deactivated",
            result)

 def test_user_activation(self) -> None:
     realm = get_realm('zulip')
     now = timezone_now()
     user = do_create_user('email', 'password', realm, 'full_name', 'short_name')
     do_deactivate_user(user)
     do_activate_user(user)
     do_deactivate_user(user)
     do_reactivate_user(user)
     self.assertEqual(RealmAuditLog.objects.filter(event_time__gte=now).count(), 5)
     event_types = list(RealmAuditLog.objects.filter(
         realm=realm, acting_user=None, modified_user=user, modified_stream=None,
         event_time__gte=now, event_time__lte=now+timedelta(minutes=60))
         .order_by('event_time').values_list('event_type', flat=True))
     self.assertEqual(event_types, [RealmAuditLog.USER_CREATED, RealmAuditLog.USER_DEACTIVATED, RealmAuditLog.USER_ACTIVATED,
                                    RealmAuditLog.USER_DEACTIVATED, RealmAuditLog.USER_REACTIVATED])

    def test_webhook_deactivated_user(self):
        """
        Deactivated users can't use webhooks

        """
        email = "[email protected]"
        user_profile = get_user_profile_by_email(email)
        do_deactivate_user(user_profile)

        api_key = self.get_api_key(email)
        url = "/api/v1/external/jira?api_key=%s&stream=jira_custom" % (api_key,)
        data = self.fixture_data('jira', "created")
        result = self.client_post(url, data,
                                  content_type="application/json")
        self.assert_json_error_contains(result, "Account not active", status_code=400)

    def test_webhook_deactivated_user(self):
        # type: () -> None
        """
        Deactivated users can't use webhooks

        """
        user_profile = self.example_user('hamlet')
        do_deactivate_user(user_profile)

        url = "/api/v1/external/jira?api_key=%s&stream=jira_custom" % (
            user_profile.api_key,)
        data = self.fixture_data('jira', "created_v2")
        result = self.client_post(url, data,
                                  content_type="application/json")
        self.assert_json_error_contains(result, "Account not active", status_code=400)

    def test_get_seat_count(self) -> None:
        realm = get_realm("zulip")
        initial_count = get_seat_count(realm)
        user1 = UserProfile.objects.create(realm=realm, email='[email protected]', pointer=-1)
        user2 = UserProfile.objects.create(realm=realm, email='[email protected]', pointer=-1)
        self.assertEqual(get_seat_count(realm), initial_count + 2)

        # Test that bots aren't counted
        user1.is_bot = True
        user1.save(update_fields=['is_bot'])
        self.assertEqual(get_seat_count(realm), initial_count + 1)

        # Test that inactive users aren't counted
        do_deactivate_user(user2)
        self.assertEqual(get_seat_count(realm), initial_count)

 def get_or_build_user(self, username: str,
                       ldap_user: _LDAPUser) -> Tuple[UserProfile, bool]:  # nocoverage
     (user, built) = super().get_or_build_user(username, ldap_user)
     self.sync_avatar_from_ldap(user, ldap_user)
     if 'userAccountControl' in settings.AUTH_LDAP_USER_ATTR_MAP:
         user_disabled_in_ldap = self.is_account_control_disabled_user(ldap_user)
         if user_disabled_in_ldap and user.is_active:
             logging.info("Deactivating user %s because they are disabled in LDAP." %
                          (user.email,))
             do_deactivate_user(user)
             return (user, built)
         if not user_disabled_in_ldap and not user.is_active:
             logging.info("Reactivating user %s because they are not disabled in LDAP." %
                          (user.email,))
             do_reactivate_user(user)
     return (user, built)

    def test_requires_billing_update_for_is_active_changes(self) -> None:
        count = RealmAuditLog.objects.count()
        realm = get_realm("zulip")
        user1 = do_create_user('[email protected]', 'password', realm, 'full name', 'short name')
        do_deactivate_user(user1)
        do_reactivate_user(user1)
        # Not a proper use of do_activate_user, but it's fine to call it like this for this test
        do_activate_user(user1)
        self.assertEqual(count + 4,
                         RealmAuditLog.objects.filter(requires_billing_update=False).count())

        realm.has_seat_based_plan = True
        realm.save(update_fields=['has_seat_based_plan'])
        user2 = do_create_user('[email protected]', 'password', realm, 'full name', 'short name')
        do_deactivate_user(user2)
        do_reactivate_user(user2)
        do_activate_user(user2)
        self.assertEqual(4, RealmAuditLog.objects.filter(requires_billing_update=True).count())

 def test_active_users_log_by_is_bot(self):
     # type: () -> None
     property = 'active_users_log:is_bot:day'
     user = do_create_user('email', 'password', self.default_realm, 'full_name', 'short_name')
     self.assertEqual(1, RealmCount.objects.filter(property=property, subgroup=False)
                      .aggregate(Sum('value'))['value__sum'])
     do_deactivate_user(user)
     self.assertEqual(0, RealmCount.objects.filter(property=property, subgroup=False)
                      .aggregate(Sum('value'))['value__sum'])
     do_activate_user(user)
     self.assertEqual(1, RealmCount.objects.filter(property=property, subgroup=False)
                      .aggregate(Sum('value'))['value__sum'])
     do_deactivate_user(user)
     self.assertEqual(0, RealmCount.objects.filter(property=property, subgroup=False)
                      .aggregate(Sum('value'))['value__sum'])
     do_reactivate_user(user)
     self.assertEqual(1, RealmCount.objects.filter(property=property, subgroup=False)
                      .aggregate(Sum('value'))['value__sum'])

    def test_billing_quantity_changes_end_to_end(
            self, mock_customer_with_subscription: Mock, mock_create_subscription: Mock,
            mock_create_customer: Mock) -> None:
        self.login(self.example_email("hamlet"))
        processor = BillingProcessor.objects.create(
            log_row=RealmAuditLog.objects.order_by('id').first(), state=BillingProcessor.DONE)

        def check_billing_processor_update(event_type: str, quantity: int) -> None:
            def check_subscription_save(subscription: stripe.Subscription, idempotency_key: str) -> None:
                self.assertEqual(subscription.quantity, quantity)
                log_row = RealmAuditLog.objects.filter(
                    event_type=event_type, requires_billing_update=True).order_by('-id').first()
                self.assertEqual(idempotency_key, 'process_billing_log_entry:%s' % (log_row.id,))
                self.assertEqual(subscription.proration_date, datetime_to_timestamp(log_row.event_time))
            with patch.object(stripe.Subscription, 'save', autospec=True,
                              side_effect=check_subscription_save):
                run_billing_processor_one_step(processor)

        # Test STRIPE_PLAN_QUANTITY_RESET
        new_seat_count = 123
        # change the seat count while the user is going through the upgrade flow
        with patch('corporate.lib.stripe.get_seat_count', return_value=new_seat_count):
            self.client_post("/upgrade/", {'stripeToken': self.token,
                                           'signed_seat_count': self.signed_seat_count,
                                           'salt': self.salt,
                                           'plan': Plan.CLOUD_ANNUAL})
        check_billing_processor_update(RealmAuditLog.STRIPE_PLAN_QUANTITY_RESET, new_seat_count)

        # Test USER_CREATED
        user = do_create_user('[email protected]', 'password', get_realm('zulip'), 'full name', 'short name')
        check_billing_processor_update(RealmAuditLog.USER_CREATED, self.quantity + 1)

        # Test USER_DEACTIVATED
        do_deactivate_user(user)
        check_billing_processor_update(RealmAuditLog.USER_DEACTIVATED, self.quantity - 1)

        # Test USER_REACTIVATED
        do_reactivate_user(user)
        check_billing_processor_update(RealmAuditLog.USER_REACTIVATED, self.quantity + 1)

        # Test USER_ACTIVATED
        # Not a proper use of do_activate_user, but it's fine to call it like this for this test
        do_activate_user(user)
        check_billing_processor_update(RealmAuditLog.USER_ACTIVATED, self.quantity + 1)