def test_UserActivityWorker(self) -> None:
        fake_client = self.FakeClient()

        user = self.example_user('hamlet')
        UserActivity.objects.filter(
            user_profile = user.id,
            client = get_client('ios')
        ).delete()

        data = dict(
            user_profile_id = user.id,
            client = 'ios',
            time = time.time(),
            query = 'send_message'
        )
        fake_client.queue.append(('user_activity', data))

        with simulated_queue_client(lambda: fake_client):
            worker = queue_processors.UserActivityWorker()
            worker.setup()
            worker.start()
            activity_records = UserActivity.objects.filter(
                user_profile = user.id,
                client = get_client('ios')
            )
            self.assertTrue(len(activity_records), 1)
            self.assertTrue(activity_records[0].count, 1)

    def handle(self, *args, **options):
        # type: (*Any, **Any) -> None
        Realm.objects.create(string_id=settings.INTERNAL_BOT_DOMAIN.split('.')[0],
                             domain=settings.INTERNAL_BOT_DOMAIN)

        names = [(settings.FEEDBACK_BOT_NAME, settings.FEEDBACK_BOT)]
        create_users(names, bot_type=UserProfile.DEFAULT_BOT)

        get_client("website")
        get_client("API")

        internal_bots = [(bot['name'], bot['email_template'] % (settings.INTERNAL_BOT_DOMAIN,))
                         for bot in settings.INTERNAL_BOTS]
        create_users(internal_bots, bot_type=UserProfile.DEFAULT_BOT)
        # Set the owners for these bots to the bots themselves
        bots = UserProfile.objects.filter(email__in=[bot_info[1] for bot_info in internal_bots])
        for bot in bots:
            bot.bot_owner = bot
            bot.save()

        # Initialize the email gateway bot as an API Super User
        email_gateway_bot = UserProfile.objects.get(email__iexact=settings.EMAIL_GATEWAY_BOT)
        email_gateway_bot.is_api_super_user = True
        email_gateway_bot.save()

        self.stdout.write("Successfully populated database with initial data.\n")
        self.stdout.write("Please run ./manage.py generate_realm_creation_link to generate link for creating organization")

    def handle(self, *args, **options):
        # type: (*Any, **Any) -> None
        Realm.objects.create(domain="zulip.com")

        names = [(settings.FEEDBACK_BOT_NAME, settings.FEEDBACK_BOT)]
        create_users(names, bot_type=UserProfile.DEFAULT_BOT)

        get_client("website")
        get_client("API")

        internal_bots = [
            (bot["name"], bot["email_template"] % (settings.INTERNAL_BOT_DOMAIN,)) for bot in settings.INTERNAL_BOTS
        ]
        create_users(internal_bots, bot_type=UserProfile.DEFAULT_BOT)
        # Set the owners for these bots to the bots themselves
        bots = UserProfile.objects.filter(email__in=[bot_info[1] for bot_info in internal_bots])
        for bot in bots:
            bot.bot_owner = bot
            bot.save()

        # Initialize the email gateway bot as an API Super User
        email_gateway_bot = UserProfile.objects.get(email__iexact=settings.EMAIL_GATEWAY_BOT)
        email_gateway_bot.is_api_super_user = True
        email_gateway_bot.save()

        (admin_realm, _) = do_create_realm(settings.ADMIN_DOMAIN, settings.ADMIN_DOMAIN, True)

        set_default_streams(admin_realm, settings.DEFAULT_NEW_REALM_STREAMS)

        self.stdout.write("Successfully populated database with initial data.\n")

def api_teamcity_webhook(request, user_profile, payload=REQ(argument_type='body'),
                         stream=REQ(default='teamcity')):
    message = payload['build']

    build_name = message['buildFullName']
    build_url = message['buildStatusUrl']
    changes_url = build_url + '&tab=buildChangesDiv'
    build_number = message['buildNumber']
    build_result = message['buildResult']
    build_result_delta = message['buildResultDelta']
    build_status = message['buildStatus']

    if build_result == 'success':
        if build_result_delta == 'fixed':
            status = 'has been fixed! :thumbsup:'
        else:
            status = 'was successful! :thumbsup:'
    elif build_result == 'failure':
        if build_result_delta == 'broken':
            status = 'is broken with status %s! :thumbsdown:' % (build_status)
        else:
            status = 'is still broken with status %s! :thumbsdown:' % (build_status)
    elif build_result == 'running':
        status = 'has started.'
    else:
        status = '(has no message specified for status %s)' % (build_status)

    template = (
        u'%s build %s %s\n'
        u'Details: [changes](%s), [build log](%s)')

    body = template % (build_name, build_number, status, changes_url, build_url)
    topic = build_name

    # Check if this is a personal build, and if so try to private message the user who triggered it.
    if get_teamcity_property_value(message['teamcityProperties'], 'env.BUILD_IS_PERSONAL') == 'true':
        # The triggeredBy field gives us the teamcity user full name, and the "teamcity.build.triggeredBy.username"
        # property gives us the teamcity username. Let's try finding the user email from both.
        teamcity_fullname = message['triggeredBy'].split(';')[0]
        teamcity_user = guess_zulip_user_from_teamcity(teamcity_fullname, user_profile.realm)

        if teamcity_user is None:
            teamcity_shortname = get_teamcity_property_value(message['teamcityProperties'], 'teamcity.build.triggeredBy.username')
            if teamcity_shortname is not None:
                teamcity_user = guess_zulip_user_from_teamcity(teamcity_shortname, user_profile.realm)

        if teamcity_user is None:
            # We can't figure out who started this build - there's nothing we can do here.
            logging.info("Teamcity webhook couldn't find a matching Zulip user for Teamcity user '%s' or '%s'" % (teamcity_fullname, teamcity_shortname))
            return json_success()

        body = "Your personal build of " + body
        check_send_message(user_profile, get_client('ZulipTeamcityWebhook'), 'private', [teamcity_user.email], topic, body)
        return json_success()

    check_send_message(user_profile, get_client('ZulipTeamcityWebhook'), 'stream', [stream], topic, body)
    return json_success()

    def test_alert_words(self):
        user_profile = get_user_profile_by_email("[email protected]")
        do_set_alert_words(user_profile, ["ALERTWORD", "scaryword"])
        msg = Message(sender=user_profile, sending_client=get_client("test"))

        content = "We have an ALERTWORD day today!"
        self.assertEqual(msg.render_markdown(content), "<p>We have an ALERTWORD day today!</p>")
        self.assertEqual(msg.user_ids_with_alert_words, set([user_profile.id]))

        msg = Message(sender=user_profile, sending_client=get_client("test"))
        content = "We have a NOTHINGWORD day today!"
        self.assertEqual(msg.render_markdown(content), "<p>We have a NOTHINGWORD day today!</p>")
        self.assertEqual(msg.user_ids_with_alert_words, set())

def send_response_message(bot_id: str, message: Dict[str, Any], response_message_content: Text) -> None:
    recipient_type_name = message['type']
    bot_user = get_user_profile_by_id(bot_id)
    realm = bot_user.realm

    if recipient_type_name == 'stream':
        recipients = [message['display_recipient']]
        check_send_message(bot_user, get_client("OutgoingWebhookResponse"), recipient_type_name, recipients,
                           message['subject'], response_message_content, realm)
    elif recipient_type_name == 'private':
        recipients = [recipient['email'] for recipient in message['display_recipient']]
        check_send_message(bot_user, get_client("OutgoingWebhookResponse"), recipient_type_name, recipients,
                           None, response_message_content, realm)
    else:
        raise JsonableError(_("Invalid message type"))

 def consume(self, event: Mapping[str, Any]) -> None:
     logging.debug("Received presence event: %s" % (event),)
     user_profile = get_user_profile_by_id(event["user_profile_id"])
     client = get_client(event["client"])
     log_time = timestamp_to_datetime(event["time"])
     status = event["status"]
     do_update_user_presence(user_profile, client, log_time, status)

def api_freshdesk_webhook(request: HttpRequest, user_profile: UserProfile,
                          payload: Dict[str, Any]=REQ(argument_type='body'),
                          stream: Text=REQ(default='freshdesk')) -> HttpResponse:
    ticket_data = payload["freshdesk_webhook"]

    required_keys = [
        "triggered_event", "ticket_id", "ticket_url", "ticket_type",
        "ticket_subject", "ticket_description", "ticket_status",
        "ticket_priority", "requester_name", "requester_email",
    ]

    for key in required_keys:
        if ticket_data.get(key) is None:
            logging.warning("Freshdesk webhook error. Payload was:")
            logging.warning(request.body)
            return json_error(_("Missing key %s in JSON") % (key,))

    ticket = TicketDict(ticket_data)

    subject = "#%s: %s" % (ticket.id, ticket.subject)
    event_info = parse_freshdesk_event(ticket.triggered_event)

    if event_info[1] == "created":
        content = format_freshdesk_ticket_creation_message(ticket)
    elif event_info[0] == "note_type":
        content = format_freshdesk_note_message(ticket, event_info)
    elif event_info[0] in ("status", "priority"):
        content = format_freshdesk_property_change_message(ticket, event_info)
    else:
        # Not an event we know handle; do nothing.
        return json_success()

    check_send_stream_message(user_profile, get_client("ZulipFreshdeskWebhook"),
                              stream, subject, content)
    return json_success()

def api_newrelic_webhook(request, user_profile, alert=REQ(validator=check_dict([]), default=None),
                             deployment=REQ(validator=check_dict([]), default=None)):
    try:
        stream = request.GET['stream']
    except (AttributeError, KeyError):
        return json_error("Missing stream parameter.")

    if alert:
        # Use the message as the subject because it stays the same for
        # "opened", "acknowledged", and "closed" messages that should be
        # grouped.
        subject = alert['message']
        content = "%(long_description)s\n[View alert](%(alert_url)s)" % (alert)
    elif deployment:
        subject = "%s deploy" % (deployment['application_name'])
        content = """`%(revision)s` deployed by **%(deployed_by)s**
%(description)s

%(changelog)s""" % (deployment)
    else:
        return json_error("Unknown webhook request")

    check_send_message(user_profile, get_client("ZulipNewRelicWebhook"), "stream",
                       [stream], subject, content)
    return json_success()

def api_beanstalk_webhook(request, user_profile,
                          payload=REQ(validator=check_dict([]))):
    # Beanstalk supports both SVN and git repositories
    # We distinguish between the two by checking for a
    # 'uri' key that is only present for git repos
    git_repo = 'uri' in payload
    if git_repo:
        # To get a linkable url,
        subject, content = build_message_from_gitlog(user_profile, payload['repository']['name'],
                                                     payload['ref'], payload['commits'],
                                                     payload['before'], payload['after'],
                                                     payload['repository']['url'],
                                                     payload['pusher_name'])
    else:
        author = payload.get('author_full_name')
        url = payload.get('changeset_url')
        revision = payload.get('revision')
        (short_commit_msg, _, _) = payload.get('message').partition("\n")

        subject = "svn r%s" % (revision,)
        content = "%s pushed [revision %s](%s):\n\n> %s" % (author, revision, url, short_commit_msg)

    check_send_message(user_profile, get_client("ZulipBeanstalkWebhook"), "stream",
                       ["commits"], subject, content)
    return json_success()

def api_travis_webhook(request, user_profile, stream=REQ(default='travis'), topic=REQ(default=None)):
    message = ujson.loads(request.POST['payload'])

    author = message['author_name']
    message_type = message['status_message']
    changes = message['compare_url']

    good_status = ['Passed', 'Fixed']
    bad_status  = ['Failed', 'Broken', 'Still Failing']
    emoji = ''
    if message_type in good_status:
        emoji = ':thumbsup:'
    elif message_type in bad_status:
        emoji = ':thumbsdown:'
    else:
        emoji = "(No emoji specified for status '%s'.)" % (message_type,)

    build_url = message['build_url']

    template = (
        u'Author: %s\n'
        u'Build status: %s %s\n'
        u'Details: [changes](%s), [build log](%s)')

    body = template % (author, message_type, emoji, changes, build_url)

    check_send_message(user_profile, get_client('ZulipTravisWebhook'), 'stream', [stream], topic, body)
    return json_success()

def api_bitbucket_webhook(request, user_profile, payload=REQ(validator=check_dict([])),
                          stream=REQ(default='commits')):
    # type: (HttpRequest, UserProfile, Dict[str, Any], str) -> None
    repository = payload['repository']
    commits = [{'id': commit['raw_node'], 'message': commit['message'],
                'url': '%s%scommits/%s' % (payload['canon_url'],
                                           repository['absolute_url'],
                                           commit['raw_node'])}
               for commit in payload['commits']]

    subject = repository['name']
    if len(commits) == 0:
        # Bitbucket doesn't give us enough information to really give
        # a useful message :/
        content = ("%s [force pushed](%s)"
                   % (payload['user'],
                      payload['canon_url'] + repository['absolute_url']))
    else:
        branch = payload['commits'][-1]['branch']
        content = build_commit_list_content(commits, branch, None, payload['user'])
        subject += '/%s' % (branch,)

    check_send_message(user_profile, get_client("ZulipBitBucketWebhook"), "stream",
                       [stream], subject, content)
    return json_success()

def api_stash_webhook(request, user_profile, stream=REQ(default='')):
    try:
        payload = ujson.loads(request.body)
    except ValueError:
        return json_error("Malformed JSON input")

    # We don't get who did the push, or we'd try to report that.
    try:
        repo_name = payload["repository"]["name"]
        project_name = payload["repository"]["project"]["name"]
        branch_name = payload["refChanges"][0]["refId"].split("/")[-1]
        commit_entries = payload["changesets"]["values"]
        commits = [(entry["toCommit"]["displayId"],
                    entry["toCommit"]["message"].split("\n")[0]) for \
                       entry in commit_entries]
        head_ref = commit_entries[-1]["toCommit"]["displayId"]
    except KeyError as e:
        return json_error("Missing key %s in JSON" % (e.message,))

    try:
        stream = request.GET['stream']
    except (AttributeError, KeyError):
        stream = 'commits'

    subject = "%s/%s: %s" % (project_name, repo_name, branch_name)

    content = "`%s` was pushed to **%s** in **%s/%s** with:\n\n" % (
        head_ref, branch_name, project_name, repo_name)
    content += "\n".join("* `%s`: %s" % (
            commit[0], commit[1]) for commit in commits)

    check_send_message(user_profile, get_client("ZulipStashWebhook"), "stream",
                       [stream], subject, content)
    return json_success()

def api_circleci_webhook(request, user_profile, payload=REQ(argument_type='body'), stream=REQ(default='circleci')):
    payload = payload['payload']
    subject = get_subject(payload)
    body = get_body(payload)

    check_send_message(user_profile, get_client('ZulipCircleCIWebhook'), 'stream', [stream], subject, body)
    return json_success()

def api_stash_webhook(request, user_profile, payload=REQ(argument_type='body'),
                      stream=REQ(default='commits')):
    # type: (HttpRequest, UserProfile, Dict[str, Any], text_type) -> HttpResponse
    # We don't get who did the push, or we'd try to report that.
    try:
        repo_name = payload["repository"]["name"]
        project_name = payload["repository"]["project"]["name"]
        branch_name = payload["refChanges"][0]["refId"].split("/")[-1]
        commit_entries = payload["changesets"]["values"]
        commits = [(entry["toCommit"]["displayId"],
                    entry["toCommit"]["message"].split("\n")[0]) for \
                       entry in commit_entries]
        head_ref = commit_entries[-1]["toCommit"]["displayId"]
    except KeyError as e:
        return json_error(_("Missing key %s in JSON") % (e.message,))

    subject = "%s/%s: %s" % (project_name, repo_name, branch_name)

    content = "`%s` was pushed to **%s** in **%s/%s** with:\n\n" % (
        head_ref, branch_name, project_name, repo_name)
    content += "\n".join("* `%s`: %s" % (
            commit[0], commit[1]) for commit in commits)

    check_send_message(user_profile, get_client("ZulipStashWebhook"), "stream",
                       [stream], subject, content)
    return json_success()

def api_yo_app_webhook(request, user_profile, email=REQ(default=None),
                       username=REQ(default='Yo Bot'), topic=REQ(default='None'),
                       user_ip=REQ(default='None')):

    body = ('Yo from %s') % (username,)
    check_send_message(user_profile, get_client('ZulipYoWebhook'), 'private', [email], topic, body)
    return json_success()

def api_wordpress_webhook(request, user_profile,
                          stream=REQ(default="wordpress"),
                          topic=REQ(default="WordPress Notification"),
                          hook=REQ(default="WordPress Action"),
                          post_title=REQ(default="New WordPress Post"),
                          post_type=REQ(default="post"),
                          post_url=REQ(default="WordPress Post URL"),
                          display_name=REQ(default="New User Name"),
                          user_email=REQ(default="New User Email"),
                          user_login=REQ(default="Logged in User")):
    # type: (HttpRequest, UserProfile, text_type, text_type, text_type, text_type, text_type, text_type, text_type, text_type, text_type) -> HttpResponse

    # remove trailing whitespace (issue for some test fixtures)
    hook = hook.rstrip()

    if hook == 'publish_post' or hook == 'publish_page':
        data = PUBLISH_POST_OR_PAGE_TEMPLATE.format(type=post_type, title=post_title, url=post_url)

    elif hook == 'user_register':
        data = USER_REGISTER_TEMPLATE.format(name=display_name, email=user_email)

    elif hook == 'wp_login':
        data = WP_LOGIN_TEMPLATE.format(name=user_login)

    else:
        return json_error(_("Unknown WordPress webhook action: " + hook))

    check_send_message(user_profile, get_client("ZulipWordPressWebhook"), "stream",
                       [stream], topic, data)
    return json_success()

 def consume(self, event):
     logging.info("Received event: %s" % (event),)
     user_profile = get_user_profile_by_id(event["user_profile_id"])
     client = get_client(event["client"])
     log_time = timestamp_to_datetime(event["time"])
     status = event["status"]
     do_update_user_presence(user_profile, client, log_time, status)

def api_beanstalk_webhook(request, user_profile,
                          payload=REQ(validator=check_dict([])),
                          branches=REQ(default=None)):
    # type: (HttpRequest, UserProfile, Dict[str, Any], Optional[Text]) -> HttpResponse
    # Beanstalk supports both SVN and git repositories
    # We distinguish between the two by checking for a
    # 'uri' key that is only present for git repos
    git_repo = 'uri' in payload
    if git_repo:
        if branches is not None and branches.find(payload['branch']) == -1:
            return json_success()
        # To get a linkable url,
        for commit in payload['commits']:
            commit['author'] = {'username': commit['author']['name']}

        subject, content = build_message_from_gitlog(user_profile, payload['repository']['name'],
                                                     payload['ref'], payload['commits'],
                                                     payload['before'], payload['after'],
                                                     payload['repository']['url'],
                                                     payload['pusher_name'])
    else:
        author = payload.get('author_full_name')
        url = payload.get('changeset_url')
        revision = payload.get('revision')
        (short_commit_msg, _, _) = payload.get('message').partition("\n")

        subject = "svn r%s" % (revision,)
        content = "%s pushed [revision %s](%s):\n\n> %s" % (author, revision, url, short_commit_msg)

    check_send_message(user_profile, get_client("ZulipBeanstalkWebhook"), "stream",
                       ["commits"], subject, content)
    return json_success()

 def consume(self, event):
     # type: (Mapping[str, Any]) -> None
     user_profile = get_user_profile_by_id(event["user_profile_id"])
     client = get_client(event["client"])
     log_time = timestamp_to_datetime(event["time"])
     query = event["query"]
     do_update_user_activity(user_profile, client, query, log_time)