# -*- coding: utf-8 -*- from __future__ import absolute_import from __future__ import print_function from django.http import HttpResponse from typing import Any, Dict from zerver.lib.test_helpers import ( get_user_profile_by_email, make_client, queries_captured, ZulipTestCase, ) from zerver.models import ( split_email_to_domain, Client, UserActivity, UserProfile, ) import datetime import ujson class ActivityTest(ZulipTestCase): def test_activity(self): # type: () -> None self.login("hamlet@zulip.com") client, _ = Client.objects.get_or_create(name='website') query = '/json/users/me/pointer' last_visit = datetime.datetime.now() count=150 for user_profile in UserProfile.objects.all(): UserActivity.objects.get_or_create( user_profile=user_profile, client=client, query=query, count=count, last_visit=last_visit ) with queries_captured() as queries: self.client_get('/activity') self.assert_length(queries, 13) class UserPresenceTests(ZulipTestCase): def test_get_empty(self): # type: () -> None self.login("hamlet@zulip.com") result = self.client_post("/json/get_active_statuses") self.assert_json_success(result) json = ujson.loads(result.content) for email, presence in json['presences'].items(): self.assertEqual(presence, {}) def test_invalid_presence(self): # type: () -> None email = "hamlet@zulip.com" self.login(email) result = self.client_post("/json/users/me/presence", {'status': 'foo'}) self.assert_json_error(result, 'Invalid presence status: foo') def test_set_idle(self): # type: () -> None email = "hamlet@zulip.com" self.login(email) client = 'website' def test_result(result): # type: (HttpResponse) -> datetime.datetime self.assert_json_success(result) json = ujson.loads(result.content) self.assertEqual(json['presences'][email][client]['status'], 'idle') self.assertIn('timestamp', json['presences'][email][client]) self.assertIsInstance(json['presences'][email][client]['timestamp'], int) self.assertEqual(list(json['presences'].keys()), ['hamlet@zulip.com']) return json['presences'][email][client]['timestamp'] result = self.client_post("/json/users/me/presence", {'status': 'idle'}) test_result(result) result = self.client_post("/json/get_active_statuses", {}) timestamp = test_result(result) email = "othello@zulip.com" self.login(email) self.client_post("/json/users/me/presence", {'status': 'idle'}) result = self.client_post("/json/get_active_statuses", {}) self.assert_json_success(result) json = ujson.loads(result.content) self.assertEqual(json['presences'][email][client]['status'], 'idle') self.assertEqual(json['presences']['hamlet@zulip.com'][client]['status'], 'idle') self.assertEqual(sorted(json['presences'].keys()), ['hamlet@zulip.com', 'othello@zulip.com']) newer_timestamp = json['presences'][email][client]['timestamp'] self.assertGreaterEqual(newer_timestamp, timestamp) def test_set_active(self): # type: () -> None self.login("hamlet@zulip.com") client = 'website' self.client_post("/json/users/me/presence", {'status': 'idle'}) result = self.client_post("/json/get_active_statuses", {}) self.assert_json_success(result) json = ujson.loads(result.content) self.assertEqual(json['presences']["hamlet@zulip.com"][client]['status'], 'idle') email = "othello@zulip.com" self.login("othello@zulip.com") self.client_post("/json/users/me/presence", {'status': 'idle'}) result = self.client_post("/json/get_active_statuses", {}) self.assert_json_success(result) json = ujson.loads(result.content) self.assertEqual(json['presences'][email][client]['status'], 'idle') self.assertEqual(json['presences']['hamlet@zulip.com'][client]['status'], 'idle') self.client_post("/json/users/me/presence", {'status': 'active'}) result = self.client_post("/json/get_active_statuses", {}) self.assert_json_success(result) json = ujson.loads(result.content) self.assertEqual(json['presences'][email][client]['status'], 'active') self.assertEqual(json['presences']['hamlet@zulip.com'][client]['status'], 'idle') def test_no_mit(self): # type: () -> None """Zephyr mirror realms such as MIT never get a list of users""" self.login("espuser@mit.edu") result = self.client_post("/json/users/me/presence", {'status': 'idle'}) self.assert_json_success(result) json = ujson.loads(result.content) self.assertEqual(json['presences'], {}) def test_mirror_presence(self): # type: () -> None """Zephyr mirror realms find out the status of their mirror bot""" email = 'espuser@mit.edu' user_profile = get_user_profile_by_email(email) self.login(email) def post_presence(): # type: () -> Dict[str, Any] result = self.client_post("/json/users/me/presence", {'status': 'idle'}) self.assert_json_success(result) json = ujson.loads(result.content) return json json = post_presence() self.assertEqual(json['zephyr_mirror_active'], False) self._simulate_mirror_activity_for_user(user_profile) json = post_presence() self.assertEqual(json['zephyr_mirror_active'], True) def _simulate_mirror_activity_for_user(self, user_profile): # type: (UserProfile) -> None last_visit = datetime.datetime.now() client = make_client('zephyr_mirror') UserActivity.objects.get_or_create( user_profile=user_profile, client=client, query='get_events_backend', count=2, last_visit=last_visit ) def test_same_realm(self): # type: () -> None self.login("espuser@mit.edu") self.client_post("/json/users/me/presence", {'status': 'idle'}) result = self.client_post("/accounts/logout/") # Ensure we don't see hamlet@zulip.com information leakage self.login("hamlet@zulip.com") result = self.client_post("/json/users/me/presence", {'status': 'idle'}) self.assert_json_success(result) json = ujson.loads(result.content) self.assertEqual(json['presences']["hamlet@zulip.com"]["website"]['status'], 'idle') # We only want @zulip.com emails for email in json['presences'].keys(): self.assertEqual(split_email_to_domain(email), 'zulip.com')