Add endpoint to fetch presence data for a single user.

This is an experimental API subject to its data format being changed.

Fixes #3638.

zerver/models.py

@@ -1307,6 +1307,26 @@ class UserPresence(models.Model):
         elif status == UserPresence.IDLE:
             return 'idle'
 
+    @staticmethod
+    def get_status_dict_by_user(user_profile):
+        # type: (int) -> defaultdict[Any, Dict[Any, Any]]
+        query = UserPresence.objects.filter(user_profile=user_profile).values(
+            'client__name',
+            'status',
+            'timestamp',
+            'user_profile__email',
+            'user_profile__id',
+            'user_profile__enable_offline_push_notifications',
+            'user_profile__is_mirror_dummy',
+        )
+
+        if PushDeviceToken.objects.filter(user=user_profile).exists():
+            mobile_user_ids = [user_profile.id]
+        else:
+            mobile_user_ids = []
+
+        return UserPresence.get_status_dicts_for_query(query, mobile_user_ids)
+
     @staticmethod
     def get_status_dict_by_realm(realm_id):
         # type: (int) -> defaultdict[Any, Dict[Any, Any]]

zerver/tests/test_presence.py

@@ -6,6 +6,7 @@ from django.http import HttpResponse
 from django.utils import timezone
 
 from typing import Any, Dict
+from zerver.lib.actions import do_deactivate_user
 from zerver.lib.test_helpers import (
     get_user_profile_by_email,
     make_client,
@@ -161,3 +162,43 @@ class UserPresenceTests(ZulipTestCase):
         # We only want @zulip.com emails
         for email in json['presences'].keys():
             self.assertEqual(email_to_domain(email), 'zulip.com')
+
+class SingleUserPresenceTests(ZulipTestCase):
+    def test_single_user_get(self):
+        # type: () -> None
+
+        # First, we setup the test with some data
+        email = "othello@zulip.com"
+        self.login("othello@zulip.com")
+        result = self.client_post("/json/users/me/presence", {'status': 'active'})
+        result = self.client_post("/json/users/me/presence", {'status': 'active'},
+                                  HTTP_USER_AGENT="ZulipDesktop/1.0")
+        result = self.client_post("/api/v1/users/me/presence", {'status': 'idle'},
+                                  HTTP_USER_AGENT="ZulipAndroid/1.0",
+                                  **self.api_auth(email))
+        self.assert_json_success(result)
+
+        # Check some error conditions
+        result = self.client_get("/json/users/nonexistence@zulip.com/presence")
+        self.assert_json_error(result, "No such user")
+
+        result = self.client_get("/json/users/cordelia@zulip.com/presence")
+        self.assert_json_error(result, "No presence data for cordelia@zulip.com")
+
+        do_deactivate_user(get_user_profile_by_email("cordelia@zulip.com"))
+        result = self.client_get("/json/users/cordelia@zulip.com/presence")
+        self.assert_json_error(result, "No such user")
+
+        result = self.client_get("/json/users/new-user-bot@zulip.com/presence")
+        self.assert_json_error(result, "No presence for bot users")
+
+        self.login("sipbtest@mit.edu")
+        result = self.client_get("/json/users/othello@zulip.com/presence")
+        self.assert_json_error(result, "No such user")
+
+        # Then, we check everything works
+        self.login("hamlet@zulip.com")
+        result = self.client_get("/json/users/othello@zulip.com/presence")
+        result_dict = ujson.loads(result.content)
+        self.assertEqual(set(result_dict['presence'].keys()), {"ZulipAndroid", "website"})
+        self.assertEqual(set(result_dict['presence']['website'].keys()), {"status", "timestamp"})

zerver/views/presence.py

@@ -13,13 +13,38 @@ from zerver.lib.actions import get_status_dict, update_user_presence
 from zerver.lib.request import has_request_variables, REQ, JsonableError
 from zerver.lib.response import json_success, json_error
 from zerver.lib.validator import check_bool
-from zerver.models import UserActivity, UserPresence, UserProfile
+from zerver.models import UserActivity, UserPresence, UserProfile, \
+    get_user_profile_by_email
 
 def get_status_list(requesting_user_profile):
     # type: (UserProfile) -> Dict[str, Any]
     return {'presences': get_status_dict(requesting_user_profile),
             'server_timestamp': time.time()}
 
+def get_presence_backend(request, user_profile, email):
+    # type: (HttpRequest, UserProfile, Text) -> HttpResponse
+    try:
+        target = get_user_profile_by_email(email)
+    except UserProfile.DoesNotExist:
+        return json_error(_('No such user'))
+    if target.realm != user_profile.realm:
+        return json_error(_('No such user'))
+    if not target.is_active:
+        return json_error(_('No such user'))
+    if target.is_bot:
+        return json_error(_('No presence 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])
+    for val in result['presence'].values():
+        del val['client']
+        del val['pushable']
+    return json_success(result)
+
 @has_request_variables
 def update_active_status_backend(request, user_profile, status=REQ(),
                                  new_user_input=REQ(validator=check_bool, default=False)):

zproject/urls.py

@@ -193,6 +193,8 @@ v1_api_and_json_patterns = [
          'POST': 'zerver.views.users.create_user_backend'}),
     url(r'^users/(?P<email>(?!me)[^/]*)/reactivate$', rest_dispatch,
         {'POST': 'zerver.views.users.reactivate_user_backend'}),
+    url(r'^users/(?P<email>(?!me)[^/]*)/presence$', rest_dispatch,
+        {'GET': 'zerver.views.presence.get_presence_backend'}),
     url(r'^users/(?P<email>(?!me)[^/]*)$', rest_dispatch,
         {'PATCH': 'zerver.views.users.update_user_backend',
          'DELETE': 'zerver.views.users.deactivate_user_backend'}),