mirror of https://github.com/zulip/zulip.git
1912 lines
79 KiB
Python
1912 lines
79 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import absolute_import
|
|
|
|
from typing import Any, Dict, List, Mapping, Optional, Sequence
|
|
|
|
from zerver.lib import cache
|
|
|
|
from zerver.lib.test_helpers import (
|
|
AuthedTestCase, queries_captured, stub, tornado_redirected_to_list
|
|
)
|
|
|
|
from zerver.decorator import (
|
|
JsonableError
|
|
)
|
|
|
|
from zerver.lib.test_runner import (
|
|
slow
|
|
)
|
|
|
|
from zerver.models import (
|
|
get_display_recipient, Message, Realm, Recipient, Stream, Subscription,
|
|
UserProfile,
|
|
)
|
|
|
|
from zerver.lib.actions import (
|
|
create_stream_if_needed, do_add_default_stream, do_add_subscription, do_change_is_admin,
|
|
do_create_realm, do_remove_default_stream, do_set_realm_create_stream_by_admins_only,
|
|
gather_subscriptions_helper,
|
|
gather_subscriptions, get_default_streams_for_realm, get_realm, get_stream,
|
|
get_user_profile_by_email, set_default_streams, get_subscription
|
|
)
|
|
|
|
from django.http import HttpResponse
|
|
import random
|
|
import ujson
|
|
import six
|
|
from six import text_type
|
|
from six.moves import range, urllib
|
|
|
|
|
|
class StreamAdminTest(AuthedTestCase):
|
|
def test_make_stream_public(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'private_stream', invite_only=True)
|
|
|
|
do_change_is_admin(user_profile, True)
|
|
params = {
|
|
'stream_name': 'private_stream'
|
|
}
|
|
result = self.client.post("/json/make_stream_public", params)
|
|
self.assert_json_error(result, 'You are not invited to this stream.')
|
|
|
|
do_add_subscription(user_profile, stream)
|
|
|
|
do_change_is_admin(user_profile, True)
|
|
params = {
|
|
'stream_name': 'private_stream'
|
|
}
|
|
result = self.client.post("/json/make_stream_public", params)
|
|
self.assert_json_success(result)
|
|
stream = Stream.objects.get(name='private_stream', realm=realm)
|
|
self.assertFalse(stream.invite_only)
|
|
|
|
def test_make_stream_private(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'public_stream')
|
|
|
|
do_change_is_admin(user_profile, True)
|
|
params = {
|
|
'stream_name': 'public_stream'
|
|
}
|
|
result = self.client.post("/json/make_stream_private", params)
|
|
self.assert_json_success(result)
|
|
stream = Stream.objects.get(name='public_stream', realm=realm)
|
|
self.assertTrue(stream.invite_only)
|
|
|
|
def test_deactivate_stream_backend(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'new_stream')
|
|
do_add_subscription(user_profile, stream, no_log=True)
|
|
do_change_is_admin(user_profile, True)
|
|
|
|
result = self.client.delete('/json/streams/new_stream')
|
|
self.assert_json_success(result)
|
|
subscription_exists = Subscription.objects.filter(
|
|
user_profile=user_profile,
|
|
recipient__type_id=stream.id,
|
|
recipient__type=Recipient.STREAM,
|
|
active=True,
|
|
).exists()
|
|
self.assertFalse(subscription_exists)
|
|
|
|
def test_deactivate_stream_backend_requires_existing_stream(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'new_stream')
|
|
do_change_is_admin(user_profile, True)
|
|
|
|
result = self.client.delete('/json/streams/unknown_stream')
|
|
self.assert_json_error(result, 'No such stream name')
|
|
|
|
def test_deactivate_stream_backend_requires_realm_admin(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'new_stream')
|
|
do_add_subscription(user_profile, stream, no_log=True)
|
|
|
|
result = self.client.delete('/json/streams/new_stream')
|
|
self.assert_json_error(result, 'Must be a realm administrator')
|
|
|
|
def test_rename_stream(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'stream_name1')
|
|
do_add_subscription(user_profile, stream, no_log=True)
|
|
do_change_is_admin(user_profile, True)
|
|
|
|
events = [] # type: List[Dict[str, Any]]
|
|
with tornado_redirected_to_list(events):
|
|
result = self.client.post('/json/rename_stream?old_name=stream_name1&new_name=stream_name2')
|
|
self.assert_json_success(result)
|
|
|
|
event = events[1]['event']
|
|
self.assertEqual(event, dict(
|
|
op='update',
|
|
type='stream',
|
|
property='name',
|
|
value='stream_name2',
|
|
name='stream_name1'
|
|
))
|
|
users = events[1]['users']
|
|
self.assertEqual(users, [user_profile.id])
|
|
|
|
stream_name1_exists = Stream.objects.filter(
|
|
name='stream_name1',
|
|
realm=realm,
|
|
).exists()
|
|
self.assertFalse(stream_name1_exists)
|
|
stream_name2_exists = Stream.objects.filter(
|
|
name='stream_name2',
|
|
realm=realm,
|
|
).exists()
|
|
self.assertTrue(stream_name2_exists)
|
|
|
|
def test_rename_stream_requires_realm_admin(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'stream_name1')
|
|
|
|
result = self.client.post('/json/rename_stream?old_name=stream_name1&new_name=stream_name2')
|
|
self.assert_json_error(result, 'Must be a realm administrator')
|
|
|
|
def test_change_stream_description(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'stream_name1')
|
|
do_add_subscription(user_profile, stream, no_log=True)
|
|
do_change_is_admin(user_profile, True)
|
|
|
|
events = [] # type: List[Dict[str, Any]]
|
|
with tornado_redirected_to_list(events):
|
|
result = self.client_patch('/json/streams/stream_name1',
|
|
{'description': ujson.dumps('Test description')})
|
|
self.assert_json_success(result)
|
|
|
|
event = events[0]['event']
|
|
self.assertEqual(event, dict(
|
|
op='update',
|
|
type='stream',
|
|
property='description',
|
|
value='Test description',
|
|
name='stream_name1'
|
|
))
|
|
users = events[0]['users']
|
|
self.assertEqual(users, [user_profile.id])
|
|
|
|
stream = Stream.objects.get(
|
|
name='stream_name1',
|
|
realm=realm,
|
|
)
|
|
self.assertEqual('Test description', stream.description)
|
|
|
|
def test_change_stream_description_requires_realm_admin(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
|
|
realm = user_profile.realm
|
|
stream, _ = create_stream_if_needed(realm, 'stream_name1')
|
|
do_add_subscription(user_profile, stream, no_log=True)
|
|
do_change_is_admin(user_profile, False)
|
|
|
|
result = self.client_patch('/json/streams/stream_name1',
|
|
{'description': ujson.dumps('Test description')})
|
|
self.assert_json_error(result, 'Must be a realm administrator')
|
|
|
|
def set_up_stream_for_deletion(self, stream_name, invite_only=False,
|
|
subscribed=True):
|
|
# type: (str, bool, bool) -> Stream
|
|
"""
|
|
Create a stream for deletion by an administrator.
|
|
"""
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
stream, _ = create_stream_if_needed(user_profile.realm, stream_name,
|
|
invite_only=invite_only)
|
|
|
|
# For testing deleting streams you aren't on.
|
|
if subscribed:
|
|
do_add_subscription(user_profile, stream, no_log=True)
|
|
|
|
do_change_is_admin(user_profile, True)
|
|
|
|
return stream
|
|
|
|
def delete_stream(self, stream, subscribed=True):
|
|
# type: (Stream, bool) -> None
|
|
"""
|
|
Delete the stream and assess the result.
|
|
"""
|
|
active_name = stream.name
|
|
|
|
events = [] # type: List[Dict[str, Any]]
|
|
with tornado_redirected_to_list(events):
|
|
result = self.client.delete('/json/streams/' + active_name)
|
|
self.assert_json_success(result)
|
|
|
|
deletion_events = [e['event'] for e in events if e['event']['type'] == 'subscription']
|
|
if subscribed:
|
|
self.assertEqual(deletion_events[0], dict(
|
|
op='remove',
|
|
type='subscription',
|
|
subscriptions=[{'name': active_name, 'stream_id': stream.id}]
|
|
))
|
|
else:
|
|
# You could delete the stream, but you weren't on it so you don't
|
|
# receive an unsubscription event.
|
|
self.assertEqual(deletion_events, [])
|
|
|
|
with self.assertRaises(Stream.DoesNotExist):
|
|
Stream.objects.get(realm=get_realm("zulip.com"), name=active_name)
|
|
|
|
# A deleted stream's name is changed, is deactivated, is invite-only,
|
|
# and has no subscribers.
|
|
deactivated_stream_name = "!DEACTIVATED:" + active_name
|
|
deactivated_stream = Stream.objects.get(name=deactivated_stream_name)
|
|
self.assertTrue(deactivated_stream.deactivated)
|
|
self.assertTrue(deactivated_stream.invite_only)
|
|
self.assertEqual(deactivated_stream.name, deactivated_stream_name)
|
|
subscribers = self.users_subscribed_to_stream(
|
|
deactivated_stream_name, "zulip.com")
|
|
self.assertEqual(subscribers, [])
|
|
|
|
# It doesn't show up in the list of public streams anymore.
|
|
result = self.client.get("/json/streams?include_subscribed=false")
|
|
public_streams = [s["name"] for s in ujson.loads(result.content)["streams"]]
|
|
self.assertNotIn(active_name, public_streams)
|
|
self.assertNotIn(deactivated_stream_name, public_streams)
|
|
|
|
# Even if you could guess the new name, you can't subscribe to it.
|
|
result = self.client.post(
|
|
"/json/users/me/subscriptions",
|
|
{"subscriptions": ujson.dumps([{"name": deactivated_stream_name}])})
|
|
self.assert_json_error(
|
|
result, "Unable to access stream (%s)." % (deactivated_stream_name,))
|
|
|
|
def test_delete_public_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
When an administrator deletes a public stream, that stream is not
|
|
visible to users at all anymore.
|
|
"""
|
|
stream = self.set_up_stream_for_deletion("newstream")
|
|
self.delete_stream(stream)
|
|
|
|
def test_delete_private_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
Administrators can delete private streams they are on.
|
|
"""
|
|
stream = self.set_up_stream_for_deletion("newstream", invite_only=True)
|
|
self.delete_stream(stream)
|
|
|
|
def test_delete_streams_youre_not_on(self):
|
|
# type: () -> None
|
|
"""
|
|
Administrators can delete public streams they aren't on, but cannot
|
|
delete private streams they aren't on.
|
|
"""
|
|
pub_stream = self.set_up_stream_for_deletion(
|
|
"pubstream", subscribed=False)
|
|
self.delete_stream(pub_stream, subscribed=False)
|
|
|
|
priv_stream = self.set_up_stream_for_deletion(
|
|
"privstream", subscribed=False, invite_only=True)
|
|
|
|
result = self.client.delete('/json/streams/' + priv_stream.name)
|
|
self.assert_json_error(
|
|
result, "Cannot administer invite-only streams this way")
|
|
|
|
def attempt_unsubscribe_of_principal(self, is_admin=False, is_subbed=True,
|
|
invite_only=False, other_user_subbed=True):
|
|
# type: (bool, bool, bool, bool) -> HttpResponse
|
|
|
|
# Set up the main user, who is in most cases an admin.
|
|
email = "hamlet@zulip.com"
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
realm = user_profile.realm
|
|
if is_admin:
|
|
do_change_is_admin(user_profile, True)
|
|
|
|
# Set up the stream.
|
|
stream_name = u"hümbüǵ"
|
|
stream, _ = create_stream_if_needed(realm, stream_name,
|
|
invite_only=invite_only)
|
|
|
|
# Set up the principal to be unsubscribed.
|
|
other_email = "cordelia@zulip.com"
|
|
other_user_profile = get_user_profile_by_email(other_email)
|
|
|
|
# Subscribe the admin and/or principal as specified in the flags.
|
|
if is_subbed:
|
|
do_add_subscription(user_profile, stream, no_log=True)
|
|
if other_user_subbed:
|
|
do_add_subscription(other_user_profile, stream, no_log=True)
|
|
|
|
result = self.client.post(
|
|
"/json/subscriptions/remove",
|
|
{"subscriptions": ujson.dumps([stream.name]),
|
|
"principals": ujson.dumps([other_email])})
|
|
|
|
# If the removal succeeded, then assert that Cordelia is no longer subscribed.
|
|
if result.status_code not in [400]:
|
|
subbed_users = self.users_subscribed_to_stream(stream_name, other_user_profile.realm.domain)
|
|
self.assertNotIn(other_user_profile, subbed_users)
|
|
|
|
return result
|
|
|
|
def test_cant_remove_others_from_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
If you're not an admin, you can't remove other people from streams.
|
|
"""
|
|
result = self.attempt_unsubscribe_of_principal(
|
|
is_admin=False, is_subbed=True, invite_only=False,
|
|
other_user_subbed=True)
|
|
self.assert_json_error(
|
|
result, "This action requires administrative rights")
|
|
|
|
def test_admin_remove_others_from_public_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
If you're an admin, you can remove people from public streams, even
|
|
those you aren't on.
|
|
"""
|
|
result = self.attempt_unsubscribe_of_principal(
|
|
is_admin=True, is_subbed=True, invite_only=False,
|
|
other_user_subbed=True)
|
|
json = self.assert_json_success(result)
|
|
self.assertEqual(len(json["removed"]), 1)
|
|
self.assertEqual(len(json["not_subscribed"]), 0)
|
|
|
|
def test_admin_remove_others_from_subbed_private_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
If you're an admin, you can remove other people from private streams you
|
|
are on.
|
|
"""
|
|
result = self.attempt_unsubscribe_of_principal(
|
|
is_admin=True, is_subbed=True, invite_only=True,
|
|
other_user_subbed=True)
|
|
json = self.assert_json_success(result)
|
|
self.assertEqual(len(json["removed"]), 1)
|
|
self.assertEqual(len(json["not_subscribed"]), 0)
|
|
|
|
def test_admin_remove_others_from_unsubbed_private_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
Even if you're an admin, you can't remove people from private
|
|
streams you aren't on.
|
|
"""
|
|
result = self.attempt_unsubscribe_of_principal(
|
|
is_admin=True, is_subbed=False, invite_only=True,
|
|
other_user_subbed=True)
|
|
self.assert_json_error(
|
|
result, "Cannot administer invite-only streams this way")
|
|
|
|
def test_create_stream_by_admins_only_setting(self):
|
|
# type: () -> None
|
|
"""
|
|
When realm.create_stream_by_admins_only setting is active,
|
|
non admin users shouldn't be able to create new streams.
|
|
"""
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
user_profile = get_user_profile_by_email(email)
|
|
do_change_is_admin(user_profile, False)
|
|
|
|
do_set_realm_create_stream_by_admins_only(user_profile.realm, True)
|
|
stream_name = ['adminsonlysetting']
|
|
result = self.common_subscribe_to_streams(
|
|
email,
|
|
stream_name
|
|
)
|
|
self.assert_json_error(result, 'User cannot create streams.')
|
|
|
|
# Change setting back to default
|
|
do_set_realm_create_stream_by_admins_only(user_profile.realm, False)
|
|
|
|
def test_remove_already_not_subbed(self):
|
|
# type: () -> None
|
|
"""
|
|
Trying to unsubscribe someone who already isn't subscribed to a stream
|
|
fails gracefully.
|
|
"""
|
|
result = self.attempt_unsubscribe_of_principal(
|
|
is_admin=True, is_subbed=False, invite_only=False,
|
|
other_user_subbed=False)
|
|
json = self.assert_json_success(result)
|
|
self.assertEqual(len(json["removed"]), 0)
|
|
self.assertEqual(len(json["not_subscribed"]), 1)
|
|
|
|
def test_remove_invalid_user(self):
|
|
# type: () -> None
|
|
"""
|
|
Trying to unsubscribe an invalid user from a stream fails gracefully.
|
|
"""
|
|
admin_email = "hamlet@zulip.com"
|
|
self.login(admin_email)
|
|
user_profile = get_user_profile_by_email(admin_email)
|
|
do_change_is_admin(user_profile, True)
|
|
realm = user_profile.realm
|
|
|
|
stream_name = u"hümbüǵ"
|
|
stream, _ = create_stream_if_needed(realm, stream_name)
|
|
|
|
result = self.client.post("/json/subscriptions/remove",
|
|
{"subscriptions": ujson.dumps([stream.name]),
|
|
"principals": ujson.dumps(["baduser@zulip.com"])})
|
|
self.assert_json_error(
|
|
result,
|
|
"User not authorized to execute queries on behalf of 'baduser@zulip.com'",
|
|
status_code=403)
|
|
|
|
class DefaultStreamTest(AuthedTestCase):
|
|
def get_default_stream_names(self, realm):
|
|
# type: (Realm) -> Set[text_type]
|
|
streams = get_default_streams_for_realm(realm)
|
|
stream_names = [s.name for s in streams]
|
|
return set(stream_names)
|
|
|
|
def test_set_default_streams(self):
|
|
# type: () -> None
|
|
(realm, _) = do_create_realm("testrealm.com", "Test Realm")
|
|
stream_names = ['apple', 'banana', 'Carrot Cake']
|
|
expected_names = stream_names + ['announce']
|
|
set_default_streams(realm, stream_names)
|
|
stream_names_set = self.get_default_stream_names(realm)
|
|
self.assertEqual(stream_names_set, set(expected_names))
|
|
|
|
def test_set_default_streams_no_notifications_stream(self):
|
|
# type: () -> None
|
|
(realm, _) = do_create_realm("testrealm.com", "Test Realm")
|
|
realm.notifications_stream = None
|
|
realm.save(update_fields=["notifications_stream"])
|
|
stream_names = ['apple', 'banana', 'Carrot Cake']
|
|
expected_names = stream_names
|
|
set_default_streams(realm, stream_names)
|
|
stream_names_set = self.get_default_stream_names(realm)
|
|
self.assertEqual(stream_names_set, set(expected_names))
|
|
|
|
def test_add_and_remove_default_stream(self):
|
|
# type: () -> None
|
|
realm = get_realm("zulip.com")
|
|
orig_stream_names = self.get_default_stream_names(realm)
|
|
do_add_default_stream(realm, 'Added Stream')
|
|
new_stream_names = self.get_default_stream_names(realm)
|
|
added_stream_names = new_stream_names - orig_stream_names
|
|
self.assertEqual(added_stream_names, set(['Added Stream']))
|
|
# idempotentcy--2nd call to add_default_stream should be a noop
|
|
do_add_default_stream(realm, 'Added Stream')
|
|
self.assertEqual(self.get_default_stream_names(realm), new_stream_names)
|
|
|
|
# start removing
|
|
do_remove_default_stream(realm, 'Added Stream')
|
|
self.assertEqual(self.get_default_stream_names(realm), orig_stream_names)
|
|
# idempotentcy--2nd call to remove_default_stream should be a noop
|
|
do_remove_default_stream(realm, 'Added Stream')
|
|
self.assertEqual(self.get_default_stream_names(realm), orig_stream_names)
|
|
|
|
def test_api_calls(self):
|
|
# type: () -> None
|
|
self.login("hamlet@zulip.com")
|
|
user_profile = get_user_profile_by_email('hamlet@zulip.com')
|
|
do_change_is_admin(user_profile, True)
|
|
stream_name = 'stream ADDED via api'
|
|
result = self.client_put('/json/default_streams', dict(stream_name=stream_name))
|
|
self.assert_json_success(result)
|
|
self.assertTrue(stream_name in self.get_default_stream_names(user_profile.realm))
|
|
|
|
# and remove it
|
|
result = self.client_delete('/json/default_streams', dict(stream_name=stream_name))
|
|
self.assert_json_success(result)
|
|
self.assertFalse(stream_name in self.get_default_stream_names(user_profile.realm))
|
|
|
|
class SubscriptionPropertiesTest(AuthedTestCase):
|
|
def test_set_stream_color(self):
|
|
# type: () -> None
|
|
"""
|
|
A POST request to /json/subscriptions/property with stream_name and
|
|
color data sets the stream color, and for that stream only.
|
|
"""
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
|
|
old_subs, _ = gather_subscriptions(get_user_profile_by_email(test_email))
|
|
sub = old_subs[0]
|
|
stream_name = sub['name']
|
|
new_color = "#ffffff" # TODO: ensure that this is different from old_color
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": "color",
|
|
"stream": stream_name,
|
|
"value": "#ffffff"}])})
|
|
|
|
self.assert_json_success(result)
|
|
|
|
new_subs = gather_subscriptions(get_user_profile_by_email(test_email))[0]
|
|
found_sub = None
|
|
for sub in new_subs:
|
|
if sub['name'] == stream_name:
|
|
found_sub = sub
|
|
break
|
|
|
|
self.assertIsNotNone(found_sub)
|
|
self.assertEqual(found_sub['color'], new_color)
|
|
|
|
new_subs.remove(found_sub)
|
|
for sub in old_subs:
|
|
if sub['name'] == stream_name:
|
|
found_sub = sub
|
|
break
|
|
old_subs.remove(found_sub)
|
|
self.assertEqual(old_subs, new_subs)
|
|
|
|
def test_set_color_missing_stream_name(self):
|
|
# type: () -> None
|
|
"""
|
|
Updating the color property requires a `stream` key.
|
|
"""
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": "color",
|
|
"value": "#ffffff"}])})
|
|
|
|
self.assert_json_error(
|
|
result, "stream key is missing from subscription_data[0]")
|
|
|
|
def test_json_subscription_property_invalid_verb(self):
|
|
# type: () -> None
|
|
"""
|
|
Called by invalid request method. No other request method other than
|
|
'post' is allowed in this case.
|
|
"""
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
subs = gather_subscriptions(get_user_profile_by_email(test_email))[0]
|
|
|
|
result = self.client.get(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": "in_home_view",
|
|
"stream": subs[0]["name"],
|
|
"value": False}])})
|
|
self.assert_json_error(result, "Invalid verb")
|
|
|
|
def test_set_color_missing_color(self):
|
|
# type: () -> None
|
|
"""
|
|
Updating the color property requires a color.
|
|
"""
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
subs = gather_subscriptions(get_user_profile_by_email(test_email))[0]
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": "color",
|
|
"stream": subs[0]["name"]}])})
|
|
|
|
self.assert_json_error(
|
|
result, "value key is missing from subscription_data[0]")
|
|
|
|
def test_set_pin_to_top(self):
|
|
# type: () -> None
|
|
"""
|
|
A POST request to /json/subscriptions/property with stream_name and
|
|
pin_to_top data pins the stream.
|
|
"""
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
|
|
user_profile = get_user_profile_by_email(test_email)
|
|
old_subs, _ = gather_subscriptions(user_profile)
|
|
sub = old_subs[0]
|
|
stream_name = sub['name']
|
|
new_pin_to_top = not sub['pin_to_top']
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": "pin_to_top",
|
|
"stream": stream_name,
|
|
"value": new_pin_to_top}])})
|
|
|
|
self.assert_json_success(result)
|
|
|
|
updated_sub = get_subscription(stream_name, user_profile)
|
|
|
|
self.assertIsNotNone(updated_sub)
|
|
self.assertEqual(updated_sub.pin_to_top, new_pin_to_top)
|
|
|
|
def test_set_subscription_property_incorrect(self):
|
|
# type: () -> None
|
|
"""
|
|
Trying to set a property incorrectly returns a JSON error.
|
|
"""
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
subs = gather_subscriptions(get_user_profile_by_email(test_email))[0]
|
|
|
|
property_name = "in_home_view"
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": property_name,
|
|
"value": "bad",
|
|
"stream": subs[0]["name"]}])})
|
|
|
|
self.assert_json_error(result,
|
|
'%s is not a boolean' % (property_name,))
|
|
|
|
property_name = "desktop_notifications"
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": property_name,
|
|
"value": "bad",
|
|
"stream": subs[0]["name"]}])})
|
|
|
|
self.assert_json_error(result,
|
|
'%s is not a boolean' % (property_name,))
|
|
|
|
property_name = "audible_notifications"
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": property_name,
|
|
"value": "bad",
|
|
"stream": subs[0]["name"]}])})
|
|
|
|
self.assert_json_error(result,
|
|
'%s is not a boolean' % (property_name,))
|
|
|
|
property_name = "color"
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": property_name,
|
|
"value": False,
|
|
"stream": subs[0]["name"]}])})
|
|
|
|
self.assert_json_error(result,
|
|
'%s is not a string' % (property_name,))
|
|
|
|
def test_json_subscription_property_invalid_stream(self):
|
|
# type: () -> None
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
|
|
stream_name = "invalid_stream"
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": "in_home_view",
|
|
"stream": stream_name,
|
|
"value": False}])})
|
|
|
|
self.assert_json_error(result, "Invalid stream %s" % (stream_name,))
|
|
|
|
def test_set_invalid_property(self):
|
|
# type: () -> None
|
|
"""
|
|
Trying to set an invalid property returns a JSON error.
|
|
"""
|
|
test_email = "hamlet@zulip.com"
|
|
self.login(test_email)
|
|
subs = gather_subscriptions(get_user_profile_by_email(test_email))[0]
|
|
result = self.client.post(
|
|
"/json/subscriptions/property",
|
|
{"subscription_data": ujson.dumps([{"property": "bad",
|
|
"value": "bad",
|
|
"stream": subs[0]["name"]}])})
|
|
|
|
self.assert_json_error(result,
|
|
"Unknown subscription property: bad")
|
|
|
|
class SubscriptionRestApiTest(AuthedTestCase):
|
|
def test_basic_add_delete(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
# add
|
|
request = {
|
|
'add': ujson.dumps([{'name': 'my_test_stream_1'}])
|
|
}
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
request,
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_success(result)
|
|
streams = self.get_streams(email)
|
|
self.assertTrue('my_test_stream_1' in streams)
|
|
|
|
# now delete the same stream
|
|
request = {
|
|
'delete': ujson.dumps(['my_test_stream_1'])
|
|
}
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
request,
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_success(result)
|
|
streams = self.get_streams(email)
|
|
self.assertTrue('my_test_stream_1' not in streams)
|
|
|
|
def test_bad_add_parameters(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
def check_for_error(val, expected_message):
|
|
# type: (Any, str) -> None
|
|
request = {
|
|
'add': ujson.dumps(val)
|
|
}
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
request,
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_error(result, expected_message)
|
|
|
|
check_for_error(['foo'], 'add[0] is not a dict')
|
|
check_for_error([{'bogus': 'foo'}], 'name key is missing from add[0]')
|
|
check_for_error([{'name': {}}], 'add[0]["name"] is not a string')
|
|
|
|
def test_bad_principals(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
request = {
|
|
'add': ujson.dumps([{'name': 'my_new_stream'}]),
|
|
'principals': ujson.dumps([{}]),
|
|
}
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
request,
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_error(result, 'principals[0] is not a string')
|
|
|
|
def test_bad_delete_parameters(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
request = {
|
|
'delete': ujson.dumps([{'name': 'my_test_stream_1'}])
|
|
}
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
request,
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_error(result, "delete[0] is not a string")
|
|
|
|
def test_add_or_delete_not_specified(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
{},
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_error(result,
|
|
'Nothing to do. Specify at least one of "add" or "delete".')
|
|
|
|
def test_patch_enforces_valid_stream_name_check(self):
|
|
# type: () -> None
|
|
"""
|
|
Only way to force an error is with a empty string.
|
|
"""
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
invalid_stream_name = ""
|
|
request = {
|
|
'delete': ujson.dumps([invalid_stream_name])
|
|
}
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
request,
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_error(result,
|
|
"Invalid stream name (%s)." % (invalid_stream_name,))
|
|
|
|
def test_stream_name_too_long(self):
|
|
# type: () -> None
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
long_stream_name = "a" * 61
|
|
request = {
|
|
'delete': ujson.dumps([long_stream_name])
|
|
}
|
|
result = self.client_patch(
|
|
"/api/v1/users/me/subscriptions",
|
|
request,
|
|
**self.api_auth(email)
|
|
)
|
|
self.assert_json_error(result,
|
|
"Stream name (%s) too long." % (long_stream_name,))
|
|
|
|
class SubscriptionAPITest(AuthedTestCase):
|
|
|
|
def setUp(self):
|
|
# type: () -> None
|
|
"""
|
|
All tests will be logged in as hamlet. Also save various useful values
|
|
as attributes that tests can access.
|
|
"""
|
|
self.test_email = "hamlet@zulip.com"
|
|
self.login(self.test_email)
|
|
self.user_profile = get_user_profile_by_email(self.test_email)
|
|
self.realm = self.user_profile.realm
|
|
self.streams = self.get_streams(self.test_email)
|
|
|
|
def make_random_stream_names(self, existing_stream_names):
|
|
# type: (List[text_type]) -> List[text_type]
|
|
"""
|
|
Helper function to make up random stream names. It takes
|
|
existing_stream_names and randomly appends a digit to the end of each,
|
|
but avoids names that appear in the list names_to_avoid.
|
|
"""
|
|
random_streams = []
|
|
all_stream_names = [stream.name for stream in Stream.objects.filter(realm=self.realm)]
|
|
for stream in existing_stream_names:
|
|
random_stream = stream + str(random.randint(0, 9))
|
|
if random_stream not in all_stream_names:
|
|
random_streams.append(random_stream)
|
|
return random_streams
|
|
|
|
def test_successful_subscriptions_list(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling /api/v1/users/me/subscriptions should successfully return your subscriptions.
|
|
"""
|
|
email = self.test_email
|
|
result = self.client.get("/api/v1/users/me/subscriptions", **self.api_auth(email))
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
self.assertIn("subscriptions", json)
|
|
for stream in json['subscriptions']:
|
|
self.assertIsInstance(stream['name'], six.string_types)
|
|
self.assertIsInstance(stream['color'], six.string_types)
|
|
self.assertIsInstance(stream['invite_only'], bool)
|
|
# check that the stream name corresponds to an actual stream
|
|
try:
|
|
Stream.objects.get(name__iexact=stream['name'], realm=self.realm)
|
|
except Stream.DoesNotExist:
|
|
self.fail("stream does not exist")
|
|
list_streams = [stream['name'] for stream in json["subscriptions"]]
|
|
# also check that this matches the list of your subscriptions
|
|
self.assertEqual(sorted(list_streams), sorted(self.streams))
|
|
|
|
def helper_check_subs_before_and_after_add(self, subscriptions, other_params,
|
|
subscribed, already_subscribed,
|
|
email, new_subs, invite_only=False):
|
|
# type: (List[text_type], Dict[str, Any], List[text_type], List[text_type], text_type, List[text_type], bool) -> None
|
|
"""
|
|
Check result of adding subscriptions.
|
|
|
|
You can add subscriptions for yourself or possibly many
|
|
principals, which is why e-mails map to subscriptions in the
|
|
result.
|
|
|
|
The result json is of the form
|
|
|
|
{"msg": "",
|
|
"result": "success",
|
|
"already_subscribed": {"iago@zulip.com": ["Venice", "Verona"]},
|
|
"subscribed": {"iago@zulip.com": ["Venice8"]}}
|
|
"""
|
|
result = self.common_subscribe_to_streams(self.test_email, subscriptions,
|
|
other_params, invite_only=invite_only)
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
self.assertEqual(sorted(subscribed), sorted(json["subscribed"][email]))
|
|
self.assertEqual(sorted(already_subscribed), sorted(json["already_subscribed"][email]))
|
|
new_streams = self.get_streams(email)
|
|
self.assertEqual(sorted(new_streams), sorted(new_subs))
|
|
|
|
def test_successful_subscriptions_add(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions should successfully add
|
|
streams, and should determine which are new subscriptions vs
|
|
which were already subscribed. We add 2 new streams to the
|
|
list of subscriptions and confirm the right number of events
|
|
are generated.
|
|
"""
|
|
self.assertNotEqual(len(self.streams), 0) # necessary for full test coverage
|
|
add_streams = [u"Verona2", u"Denmark5"]
|
|
self.assertNotEqual(len(add_streams), 0) # necessary for full test coverage
|
|
events = [] # type: List[Dict[str, Any]]
|
|
with tornado_redirected_to_list(events):
|
|
self.helper_check_subs_before_and_after_add(self.streams + add_streams, {},
|
|
add_streams, self.streams, self.test_email, self.streams + add_streams)
|
|
self.assert_length(events, 6, True)
|
|
|
|
def test_successful_subscriptions_add_with_announce(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions should successfully add
|
|
streams, and should determine which are new subscriptions vs
|
|
which were already subscribed. We add 2 new streams to the
|
|
list of subscriptions and confirm the right number of events
|
|
are generated.
|
|
"""
|
|
self.assertNotEqual(len(self.streams), 0)
|
|
add_streams = [u"Verona2", u"Denmark5"]
|
|
self.assertNotEqual(len(add_streams), 0)
|
|
events = [] # type: List[Dict[str, Any]]
|
|
other_params = {
|
|
'announce': 'true',
|
|
}
|
|
notifications_stream = Stream.objects.get(name=self.streams[0], realm=self.realm)
|
|
self.realm.notifications_stream = notifications_stream
|
|
self.realm.save()
|
|
|
|
# Delete the UserProfile from the cache so the realm change will be
|
|
# picked up
|
|
cache.cache_delete(cache.user_profile_by_email_cache_key(self.test_email))
|
|
with tornado_redirected_to_list(events):
|
|
self.helper_check_subs_before_and_after_add(self.streams + add_streams, other_params,
|
|
add_streams, self.streams, self.test_email, self.streams + add_streams)
|
|
self.assertEqual(len(events), 7)
|
|
|
|
def test_successful_subscriptions_notifies_pm(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions should notify when a new stream is created.
|
|
"""
|
|
invitee = "iago@zulip.com"
|
|
invitee_full_name = 'Iago'
|
|
|
|
current_stream = self.get_streams(invitee)[0]
|
|
invite_streams = self.make_random_stream_names([current_stream])[:1]
|
|
result = self.common_subscribe_to_streams(
|
|
invitee,
|
|
invite_streams,
|
|
extra_post_data={
|
|
'announce': 'true',
|
|
'principals': '["%s"]' % (self.user_profile.email,)
|
|
},
|
|
)
|
|
self.assert_json_success(result)
|
|
|
|
msg = self.get_last_message()
|
|
self.assertEqual(msg.recipient.type, Recipient.PERSONAL)
|
|
self.assertEqual(msg.sender_id,
|
|
get_user_profile_by_email('notification-bot@zulip.com').id)
|
|
expected_msg = "Hi there! %s just created a new stream '%s'. " \
|
|
"!_stream_subscribe_button(%s)" % (invitee_full_name,
|
|
invite_streams[0],
|
|
invite_streams[0])
|
|
self.assertEqual(msg.content, expected_msg)
|
|
|
|
def test_successful_subscriptions_notifies_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions should notify when a new stream is created.
|
|
"""
|
|
invitee = "iago@zulip.com"
|
|
invitee_full_name = 'Iago'
|
|
|
|
current_stream = self.get_streams(invitee)[0]
|
|
invite_streams = self.make_random_stream_names([current_stream])[:1]
|
|
|
|
notifications_stream = Stream.objects.get(name=current_stream, realm=self.realm)
|
|
self.realm.notifications_stream = notifications_stream
|
|
self.realm.save()
|
|
|
|
# Delete the UserProfile from the cache so the realm change will be
|
|
# picked up
|
|
cache.cache_delete(cache.user_profile_by_email_cache_key(invitee))
|
|
|
|
result = self.common_subscribe_to_streams(
|
|
invitee,
|
|
invite_streams,
|
|
extra_post_data=dict(
|
|
announce='true',
|
|
principals='["%s"]' % (self.user_profile.email,)
|
|
),
|
|
)
|
|
self.assert_json_success(result)
|
|
|
|
msg = self.get_last_message()
|
|
self.assertEqual(msg.recipient.type, Recipient.STREAM)
|
|
self.assertEqual(msg.sender_id,
|
|
get_user_profile_by_email('notification-bot@zulip.com').id)
|
|
expected_msg = "%s just created a new stream `%s`. " \
|
|
"!_stream_subscribe_button(%s)" % (invitee_full_name,
|
|
invite_streams[0],
|
|
invite_streams[0])
|
|
self.assertEqual(msg.content, expected_msg)
|
|
|
|
def test_successful_subscriptions_notifies_with_escaping(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions should notify when a new stream is created.
|
|
"""
|
|
invitee = "iago@zulip.com"
|
|
invitee_full_name = 'Iago'
|
|
|
|
current_stream = self.get_streams(invitee)[0]
|
|
notifications_stream = Stream.objects.get(name=current_stream, realm=self.realm)
|
|
self.realm.notifications_stream = notifications_stream
|
|
self.realm.save()
|
|
|
|
invite_streams = ['strange ) \\ test']
|
|
result = self.common_subscribe_to_streams(
|
|
invitee,
|
|
invite_streams,
|
|
extra_post_data={
|
|
'announce': 'true',
|
|
'principals': '["%s"]' % (self.user_profile.email,)
|
|
},
|
|
)
|
|
self.assert_json_success(result)
|
|
|
|
msg = self.get_last_message()
|
|
self.assertEqual(msg.sender_id,
|
|
get_user_profile_by_email('notification-bot@zulip.com').id)
|
|
expected_msg = "%s just created a new stream `%s`. " \
|
|
"!_stream_subscribe_button(strange \\) \\\\ test)" % (
|
|
invitee_full_name,
|
|
invite_streams[0])
|
|
self.assertEqual(msg.content, expected_msg)
|
|
|
|
def test_non_ascii_stream_subscription(self):
|
|
# type: () -> None
|
|
"""
|
|
Subscribing to a stream name with non-ASCII characters succeeds.
|
|
"""
|
|
self.helper_check_subs_before_and_after_add(self.streams + [u"hümbüǵ"], {},
|
|
[u"hümbüǵ"], self.streams, self.test_email, self.streams + [u"hümbüǵ"])
|
|
|
|
def test_subscriptions_add_too_long(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions on a stream whose name is >60
|
|
characters should return a JSON error.
|
|
"""
|
|
# character limit is 60 characters
|
|
long_stream_name = "a" * 61
|
|
result = self.common_subscribe_to_streams(self.test_email, [long_stream_name])
|
|
self.assert_json_error(result,
|
|
"Stream name (%s) too long." % (long_stream_name,))
|
|
|
|
def test_user_settings_for_adding_streams(self):
|
|
# type: () -> None
|
|
with stub(UserProfile, 'can_create_streams', lambda self: False):
|
|
result = self.common_subscribe_to_streams(self.test_email, ['stream1'])
|
|
self.assert_json_error(result, 'User cannot create streams.')
|
|
|
|
with stub(UserProfile, 'can_create_streams', lambda self: True):
|
|
result = self.common_subscribe_to_streams(self.test_email, ['stream2'])
|
|
self.assert_json_success(result)
|
|
|
|
# User should still be able to subscribe to an existing stream
|
|
with stub(UserProfile, 'can_create_streams', lambda self: False):
|
|
result = self.common_subscribe_to_streams(self.test_email, ['stream2'])
|
|
self.assert_json_success(result)
|
|
|
|
def test_subscriptions_add_invalid_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions on a stream whose name is invalid (as
|
|
defined by valid_stream_name in zerver/views.py) should return a JSON
|
|
error.
|
|
"""
|
|
# currently, the only invalid name is the empty string
|
|
invalid_stream_name = ""
|
|
result = self.common_subscribe_to_streams(self.test_email, [invalid_stream_name])
|
|
self.assert_json_error(result,
|
|
"Invalid stream name (%s)." % (invalid_stream_name,))
|
|
|
|
def assert_adding_subscriptions_for_principal(self, invitee, streams, invite_only=False):
|
|
# type: (text_type, List[text_type], bool) -> None
|
|
"""
|
|
Calling POST /json/users/me/subscriptions on behalf of another principal (for
|
|
whom you have permission to add subscriptions) should successfully add
|
|
those subscriptions and send a message to the subscribee notifying
|
|
them.
|
|
"""
|
|
other_profile = get_user_profile_by_email(invitee)
|
|
current_streams = self.get_streams(invitee)
|
|
self.assertIsInstance(other_profile, UserProfile)
|
|
self.assertNotEqual(len(current_streams), 0) # necessary for full test coverage
|
|
self.assertNotEqual(len(streams), 0) # necessary for full test coverage
|
|
streams_to_sub = streams[:1] # just add one, to make the message easier to check
|
|
streams_to_sub.extend(current_streams)
|
|
self.helper_check_subs_before_and_after_add(streams_to_sub,
|
|
{"principals": ujson.dumps([invitee])}, streams[:1], current_streams,
|
|
invitee, streams_to_sub, invite_only=invite_only)
|
|
# verify that the user was sent a message informing them about the subscription
|
|
msg = self.get_last_message()
|
|
self.assertEqual(msg.recipient.type, msg.recipient.PERSONAL)
|
|
self.assertEqual(msg.sender_id,
|
|
get_user_profile_by_email("notification-bot@zulip.com").id)
|
|
expected_msg = ("Hi there! We thought you'd like to know that %s just "
|
|
"subscribed you to the %sstream [%s](#narrow/stream/%s)."
|
|
% (self.user_profile.full_name,
|
|
'**invite-only** ' if invite_only else '',
|
|
streams[0], urllib.parse.quote(streams[0].encode('utf-8'))))
|
|
|
|
if not Stream.objects.get(name=streams[0]).invite_only:
|
|
expected_msg += ("\nYou can see historical content on a "
|
|
"non-invite-only stream by narrowing to it.")
|
|
self.assertEqual(msg.content, expected_msg)
|
|
recipients = get_display_recipient(msg.recipient)
|
|
self.assertEqual(len(recipients), 1)
|
|
assert isinstance(recipients, Sequence)
|
|
assert isinstance(recipients[0], Mapping)
|
|
# The 2 assert statements above are required to make the mypy check pass.
|
|
# They inform mypy that in the line below, recipients is a Sequence of Mappings.
|
|
self.assertEqual(recipients[0]['email'], invitee)
|
|
|
|
def test_multi_user_subscription(self):
|
|
# type: () -> None
|
|
email1 = 'cordelia@zulip.com'
|
|
email2 = 'iago@zulip.com'
|
|
realm = get_realm("zulip.com")
|
|
streams_to_sub = ['multi_user_stream']
|
|
events = [] # type: List[Dict[str, Any]]
|
|
with tornado_redirected_to_list(events):
|
|
with queries_captured() as queries:
|
|
self.common_subscribe_to_streams(
|
|
self.test_email,
|
|
streams_to_sub,
|
|
dict(principals=ujson.dumps([email1, email2])),
|
|
)
|
|
self.assert_length(queries, 43)
|
|
|
|
self.assert_length(events, 8, exact=True)
|
|
for ev in [x for x in events if x['event']['type'] not in ('message', 'stream')]:
|
|
if isinstance(ev['event']['subscriptions'][0], dict):
|
|
self.assertEqual(ev['event']['op'], 'add')
|
|
self.assertEqual(
|
|
set(ev['event']['subscriptions'][0]['subscribers']),
|
|
set([email1, email2])
|
|
)
|
|
else:
|
|
# Check "peer_add" events for streams users were
|
|
# never subscribed to, in order for the neversubscribed
|
|
# structure to stay up-to-date.
|
|
self.assertEqual(ev['event']['op'], 'peer_add')
|
|
|
|
stream = get_stream('multi_user_stream', realm)
|
|
self.assertEqual(stream.num_subscribers(), 2)
|
|
|
|
# Now add ourselves
|
|
events = []
|
|
with tornado_redirected_to_list(events):
|
|
with queries_captured() as queries:
|
|
self.common_subscribe_to_streams(
|
|
self.test_email,
|
|
streams_to_sub,
|
|
dict(principals=ujson.dumps([self.test_email])),
|
|
)
|
|
self.assert_length(queries, 8)
|
|
|
|
self.assert_length(events, 2, True)
|
|
add_event, add_peer_event = events
|
|
self.assertEqual(add_event['event']['type'], 'subscription')
|
|
self.assertEqual(add_event['event']['op'], 'add')
|
|
self.assertEqual(add_event['users'], [get_user_profile_by_email(self.test_email).id])
|
|
self.assertEqual(
|
|
set(add_event['event']['subscriptions'][0]['subscribers']),
|
|
set([email1, email2, self.test_email])
|
|
)
|
|
|
|
self.assertEqual(len(add_peer_event['users']), 13)
|
|
self.assertEqual(add_peer_event['event']['type'], 'subscription')
|
|
self.assertEqual(add_peer_event['event']['op'], 'peer_add')
|
|
self.assertEqual(add_peer_event['event']['user_email'], self.test_email)
|
|
|
|
stream = get_stream('multi_user_stream', realm)
|
|
self.assertEqual(stream.num_subscribers(), 3)
|
|
|
|
# Finally, add othello, exercising the do_add_subscription() code path.
|
|
events = []
|
|
email3 = 'othello@zulip.com'
|
|
user_profile = get_user_profile_by_email(email3)
|
|
stream = get_stream('multi_user_stream', realm)
|
|
with tornado_redirected_to_list(events):
|
|
do_add_subscription(user_profile, stream)
|
|
|
|
self.assert_length(events, 2, True)
|
|
add_event, add_peer_event = events
|
|
|
|
self.assertEqual(add_event['event']['type'], 'subscription')
|
|
self.assertEqual(add_event['event']['op'], 'add')
|
|
self.assertEqual(add_event['users'], [get_user_profile_by_email(email3).id])
|
|
self.assertEqual(
|
|
set(add_event['event']['subscriptions'][0]['subscribers']),
|
|
set([email1, email2, email3, self.test_email])
|
|
)
|
|
|
|
self.assertEqual(len(add_peer_event['users']), 14)
|
|
self.assertEqual(add_peer_event['event']['type'], 'subscription')
|
|
self.assertEqual(add_peer_event['event']['op'], 'peer_add')
|
|
self.assertEqual(add_peer_event['event']['user_email'], email3)
|
|
|
|
|
|
def test_bulk_subscribe_MIT(self):
|
|
# type: () -> None
|
|
realm = get_realm("mit.edu")
|
|
streams = ["stream_%s" % i for i in range(40)]
|
|
for stream in streams:
|
|
create_stream_if_needed(realm, stream)
|
|
|
|
events = [] # type: List[Dict[str, Any]]
|
|
with tornado_redirected_to_list(events):
|
|
with queries_captured() as queries:
|
|
self.common_subscribe_to_streams(
|
|
'starnine@mit.edu',
|
|
streams,
|
|
dict(principals=ujson.dumps(['starnine@mit.edu'])),
|
|
)
|
|
# Make sure Zephyr mirroring realms such as MIT do not get
|
|
# any tornado subscription events
|
|
self.assert_length(events, 0, True)
|
|
self.assert_length(queries, 7)
|
|
|
|
def test_bulk_subscribe_many(self):
|
|
# type: () -> None
|
|
|
|
# Create a whole bunch of streams
|
|
realm = get_realm("zulip.com")
|
|
streams = ["stream_%s" % i for i in range(20)]
|
|
for stream in streams:
|
|
create_stream_if_needed(realm, stream)
|
|
|
|
with queries_captured() as queries:
|
|
self.common_subscribe_to_streams(
|
|
self.test_email,
|
|
streams,
|
|
dict(principals=ujson.dumps([self.test_email])),
|
|
)
|
|
# Make sure we don't make O(streams) queries
|
|
self.assert_length(queries, 9)
|
|
|
|
@slow(0.15, "common_subscribe_to_streams is slow")
|
|
def test_subscriptions_add_for_principal(self):
|
|
# type: () -> None
|
|
"""
|
|
You can subscribe other people to streams.
|
|
"""
|
|
invitee = "iago@zulip.com"
|
|
current_streams = self.get_streams(invitee)
|
|
invite_streams = self.make_random_stream_names(current_streams)
|
|
self.assert_adding_subscriptions_for_principal(invitee, invite_streams)
|
|
|
|
@slow(0.15, "common_subscribe_to_streams is slow")
|
|
def test_subscriptions_add_for_principal_invite_only(self):
|
|
# type: () -> None
|
|
"""
|
|
You can subscribe other people to invite only streams.
|
|
"""
|
|
invitee = "iago@zulip.com"
|
|
current_streams = self.get_streams(invitee)
|
|
invite_streams = self.make_random_stream_names(current_streams)
|
|
self.assert_adding_subscriptions_for_principal(invitee, invite_streams,
|
|
invite_only=True)
|
|
|
|
@slow(0.15, "common_subscribe_to_streams is slow")
|
|
def test_non_ascii_subscription_for_principal(self):
|
|
# type: () -> None
|
|
"""
|
|
You can subscribe other people to streams even if they containing
|
|
non-ASCII characters.
|
|
"""
|
|
self.assert_adding_subscriptions_for_principal("iago@zulip.com", [u"hümbüǵ"])
|
|
|
|
def test_subscription_add_invalid_principal(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling subscribe on behalf of a principal that does not exist
|
|
should return a JSON error.
|
|
"""
|
|
invalid_principal = "rosencrantz-and-guildenstern@zulip.com"
|
|
# verify that invalid_principal actually doesn't exist
|
|
with self.assertRaises(UserProfile.DoesNotExist):
|
|
get_user_profile_by_email(invalid_principal)
|
|
result = self.common_subscribe_to_streams(self.test_email, self.streams,
|
|
{"principals": ujson.dumps([invalid_principal])})
|
|
self.assert_json_error(result, "User not authorized to execute queries on behalf of '%s'"
|
|
% (invalid_principal,), status_code=403)
|
|
|
|
def test_subscription_add_principal_other_realm(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling subscribe on behalf of a principal in another realm
|
|
should return a JSON error.
|
|
"""
|
|
principal = "starnine@mit.edu"
|
|
profile = get_user_profile_by_email(principal)
|
|
# verify that principal exists (thus, the reason for the error is the cross-realming)
|
|
self.assertIsInstance(profile, UserProfile)
|
|
result = self.common_subscribe_to_streams(self.test_email, self.streams,
|
|
{"principals": ujson.dumps([principal])})
|
|
self.assert_json_error(result, "User not authorized to execute queries on behalf of '%s'"
|
|
% (principal,), status_code=403)
|
|
|
|
def helper_check_subs_before_and_after_remove(self, subscriptions, json_dict,
|
|
email, new_subs):
|
|
# type: (List[text_type], Dict[str, Any], text_type, List[text_type]) -> None
|
|
"""
|
|
Check result of removing subscriptions.
|
|
|
|
Unlike adding subscriptions, you can only remove subscriptions
|
|
for yourself, so the result format is different.
|
|
|
|
{"msg": "",
|
|
"removed": ["Denmark", "Scotland", "Verona"],
|
|
"not_subscribed": ["Rome"], "result": "success"}
|
|
"""
|
|
result = self.client.post("/json/subscriptions/remove",
|
|
{"subscriptions": ujson.dumps(subscriptions)})
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
for key, val in six.iteritems(json_dict):
|
|
self.assertEqual(sorted(val), sorted(json[key])) # we don't care about the order of the items
|
|
new_streams = self.get_streams(email)
|
|
self.assertEqual(sorted(new_streams), sorted(new_subs))
|
|
|
|
def test_successful_subscriptions_remove(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling /json/subscriptions/remove should successfully remove streams,
|
|
and should determine which were removed vs which weren't subscribed to.
|
|
We cannot randomly generate stream names because the remove code
|
|
verifies whether streams exist.
|
|
"""
|
|
if len(self.streams) < 2:
|
|
self.fail() # necesssary for full test coverage
|
|
streams_to_remove = self.streams[1:]
|
|
not_subbed = []
|
|
for stream in Stream.objects.all():
|
|
if stream.name not in self.streams:
|
|
not_subbed.append(stream.name)
|
|
random.shuffle(not_subbed)
|
|
self.assertNotEqual(len(not_subbed), 0) # necessary for full test coverage
|
|
try_to_remove = not_subbed[:3] # attempt to remove up to 3 streams not already subbed to
|
|
streams_to_remove.extend(try_to_remove)
|
|
self.helper_check_subs_before_and_after_remove(streams_to_remove,
|
|
{"removed": self.streams[1:], "not_subscribed": try_to_remove},
|
|
self.test_email, [self.streams[0]])
|
|
|
|
def test_subscriptions_remove_fake_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling /json/subscriptions/remove on a stream that doesn't exist
|
|
should return a JSON error.
|
|
"""
|
|
random_streams = self.make_random_stream_names(self.streams)
|
|
self.assertNotEqual(len(random_streams), 0) # necessary for full test coverage
|
|
streams_to_remove = random_streams[:1] # pick only one fake stream, to make checking the error message easy
|
|
result = self.client.post("/json/subscriptions/remove",
|
|
{"subscriptions": ujson.dumps(streams_to_remove)})
|
|
self.assert_json_error(result, "Stream(s) (%s) do not exist" % (random_streams[0],))
|
|
|
|
def helper_subscriptions_exists(self, stream, exists, subscribed):
|
|
# type: (text_type, bool, bool) -> None
|
|
"""
|
|
A helper function that calls /json/subscriptions/exists on a stream and
|
|
verifies that the returned JSON dictionary has the exists and
|
|
subscribed values passed in as parameters. (If subscribed should not be
|
|
present, pass in None.)
|
|
"""
|
|
result = self.client.post("/json/subscriptions/exists",
|
|
{"stream": stream})
|
|
json = ujson.loads(result.content)
|
|
self.assertIn("exists", json)
|
|
self.assertEqual(json["exists"], exists)
|
|
if exists:
|
|
self.assert_json_success(result)
|
|
else:
|
|
self.assertEquals(result.status_code, 404)
|
|
if subscribed:
|
|
self.assertIn("subscribed", json)
|
|
self.assertEqual(json["subscribed"], subscribed)
|
|
|
|
def test_successful_subscriptions_exists_subbed(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling /json/subscriptions/exist on a stream to which you are subbed
|
|
should return that it exists and that you are subbed.
|
|
"""
|
|
self.assertNotEqual(len(self.streams), 0) # necessary for full test coverage
|
|
self.helper_subscriptions_exists(self.streams[0], True, True)
|
|
|
|
def test_successful_subscriptions_exists_not_subbed(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling /json/subscriptions/exist on a stream to which you are not
|
|
subbed should return that it exists and that you are not subbed.
|
|
"""
|
|
all_stream_names = [stream.name for stream in Stream.objects.filter(realm=self.realm)]
|
|
streams_not_subbed = list(set(all_stream_names) - set(self.streams))
|
|
self.assertNotEqual(len(streams_not_subbed), 0) # necessary for full test coverage
|
|
self.helper_subscriptions_exists(streams_not_subbed[0], True, False)
|
|
|
|
def test_subscriptions_does_not_exist(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling /json/subscriptions/exist on a stream that doesn't exist should
|
|
return that it doesn't exist.
|
|
"""
|
|
random_streams = self.make_random_stream_names(self.streams)
|
|
self.assertNotEqual(len(random_streams), 0) # necessary for full test coverage
|
|
self.helper_subscriptions_exists(random_streams[0], False, False)
|
|
|
|
def test_subscriptions_exist_invalid_name(self):
|
|
# type: () -> None
|
|
"""
|
|
Calling /json/subscriptions/exist on a stream whose name is invalid (as
|
|
defined by valid_stream_name in zerver/views.py) should return a JSON
|
|
error.
|
|
"""
|
|
# currently, the only invalid stream name is the empty string
|
|
invalid_stream_name = ""
|
|
result = self.client.post("/json/subscriptions/exists",
|
|
{"stream": invalid_stream_name})
|
|
self.assert_json_error(result, "Invalid characters in stream name")
|
|
|
|
def test_existing_subscriptions_autosubscription(self):
|
|
# type: () -> None
|
|
"""
|
|
Call /json/subscriptions/exist on an existing stream and autosubscribe to it.
|
|
"""
|
|
stream_name = self.streams[0]
|
|
result = self.client.post("/json/subscriptions/exists",
|
|
{"stream": stream_name, "autosubscribe": True})
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
self.assertIn("exists", json)
|
|
self.assertTrue(json["exists"])
|
|
|
|
def get_subscription(self, user_profile, stream_name):
|
|
# type: (UserProfile, text_type) -> Subscription
|
|
stream = Stream.objects.get(realm=self.realm, name=stream_name)
|
|
return Subscription.objects.get(
|
|
user_profile=user_profile,
|
|
recipient__type=Recipient.STREAM,
|
|
recipient__type_id=stream.id,
|
|
)
|
|
|
|
def test_subscriptions_add_notification_default_true(self):
|
|
# type: () -> None
|
|
"""
|
|
When creating a subscription, the desktop and audible notification
|
|
settings for that stream are derived from the global notification
|
|
settings.
|
|
"""
|
|
invitee = "iago@zulip.com"
|
|
user_profile = get_user_profile_by_email(invitee)
|
|
user_profile.enable_stream_desktop_notifications = True
|
|
user_profile.enable_stream_sounds = True
|
|
user_profile.save()
|
|
current_stream = self.get_streams(invitee)[0]
|
|
invite_streams = self.make_random_stream_names([current_stream])
|
|
self.assert_adding_subscriptions_for_principal(invitee, invite_streams)
|
|
subscription = self.get_subscription(user_profile, invite_streams[0])
|
|
self.assertTrue(subscription.desktop_notifications)
|
|
self.assertTrue(subscription.audible_notifications)
|
|
|
|
def test_subscriptions_add_notification_default_false(self):
|
|
# type: () -> None
|
|
"""
|
|
When creating a subscription, the desktop and audible notification
|
|
settings for that stream are derived from the global notification
|
|
settings.
|
|
"""
|
|
invitee = "iago@zulip.com"
|
|
user_profile = get_user_profile_by_email(invitee)
|
|
user_profile.enable_stream_desktop_notifications = False
|
|
user_profile.enable_stream_sounds = False
|
|
user_profile.save()
|
|
current_stream = self.get_streams(invitee)[0]
|
|
invite_streams = self.make_random_stream_names([current_stream])
|
|
self.assert_adding_subscriptions_for_principal(invitee, invite_streams)
|
|
subscription = self.get_subscription(user_profile, invite_streams[0])
|
|
self.assertFalse(subscription.desktop_notifications)
|
|
self.assertFalse(subscription.audible_notifications)
|
|
|
|
|
|
class GetPublicStreamsTest(AuthedTestCase):
|
|
|
|
def test_public_streams(self):
|
|
# type: () -> None
|
|
"""
|
|
Ensure that streams successfully returns a list of streams
|
|
"""
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
result = self.client.get("/json/streams?include_subscribed=false")
|
|
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
|
|
self.assertIn("streams", json)
|
|
self.assertIsInstance(json["streams"], list)
|
|
|
|
def test_public_streams_api(self):
|
|
# type: () -> None
|
|
"""
|
|
Ensure that the query we use to get public streams successfully returns
|
|
a list of streams
|
|
"""
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
# Check it correctly lists the user's subs with include_public=false
|
|
result = self.client.get("/api/v1/streams?include_public=false", **self.api_auth(email))
|
|
result2 = self.client.get("/api/v1/users/me/subscriptions", **self.api_auth(email))
|
|
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
|
|
self.assertIn("streams", json)
|
|
|
|
self.assertIsInstance(json["streams"], list)
|
|
|
|
self.assert_json_success(result2)
|
|
json2 = ujson.loads(result2.content)
|
|
|
|
self.assertEqual(sorted([s["name"] for s in json["streams"]]),
|
|
sorted([s["name"] for s in json2["subscriptions"]]))
|
|
|
|
# Check it correctly lists all public streams with include_subscribed=false
|
|
result = self.client.get("/api/v1/streams?include_public=true&include_subscribed=false",
|
|
**self.api_auth(email))
|
|
self.assert_json_success(result)
|
|
|
|
json = ujson.loads(result.content)
|
|
all_streams = [stream.name for stream in
|
|
Stream.objects.filter(realm=get_user_profile_by_email(email).realm)]
|
|
self.assertEqual(sorted(s["name"] for s in json["streams"]),
|
|
sorted(all_streams))
|
|
|
|
# Check non-superuser can't use include_all_active
|
|
result = self.client.get("/api/v1/streams?include_all_active=true",
|
|
**self.api_auth(email))
|
|
self.assertEqual(result.status_code, 400)
|
|
|
|
class InviteOnlyStreamTest(AuthedTestCase):
|
|
def test_must_be_subbed_to_send(self):
|
|
# type: () -> None
|
|
"""
|
|
If you try to send a message to an invite-only stream to which
|
|
you aren't subscribed, you'll get a 400.
|
|
"""
|
|
self.login("hamlet@zulip.com")
|
|
# Create Saxony as an invite-only stream.
|
|
self.assert_json_success(
|
|
self.common_subscribe_to_streams("hamlet@zulip.com", ["Saxony"],
|
|
invite_only=True))
|
|
|
|
email = "cordelia@zulip.com"
|
|
with self.assertRaises(JsonableError):
|
|
self.send_message(email, "Saxony", Recipient.STREAM)
|
|
|
|
def test_list_respects_invite_only_bit(self):
|
|
# type: () -> None
|
|
"""
|
|
Make sure that /api/v1/users/me/subscriptions properly returns
|
|
the invite-only bit for streams that are invite-only
|
|
"""
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
|
|
result1 = self.common_subscribe_to_streams(email, ["Saxony"], invite_only=True)
|
|
self.assert_json_success(result1)
|
|
result2 = self.common_subscribe_to_streams(email, ["Normandy"], invite_only=False)
|
|
self.assert_json_success(result2)
|
|
result = self.client.get("/api/v1/users/me/subscriptions", **self.api_auth(email))
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
self.assertIn("subscriptions", json)
|
|
for sub in json["subscriptions"]:
|
|
if sub['name'] == "Normandy":
|
|
self.assertEqual(sub['invite_only'], False, "Normandy was mistakenly marked invite-only")
|
|
if sub['name'] == "Saxony":
|
|
self.assertEqual(sub['invite_only'], True, "Saxony was not properly marked invite-only")
|
|
|
|
@slow(0.15, "lots of queries")
|
|
def test_inviteonly(self):
|
|
# type: () -> None
|
|
# Creating an invite-only stream is allowed
|
|
email = 'hamlet@zulip.com'
|
|
stream_name = "Saxony"
|
|
|
|
result = self.common_subscribe_to_streams(email, [stream_name], invite_only=True)
|
|
self.assert_json_success(result)
|
|
|
|
json = ujson.loads(result.content)
|
|
self.assertEqual(json["subscribed"], {email: [stream_name]})
|
|
self.assertEqual(json["already_subscribed"], {})
|
|
|
|
# Subscribing oneself to an invite-only stream is not allowed
|
|
email = "othello@zulip.com"
|
|
self.login(email)
|
|
result = self.common_subscribe_to_streams(email, [stream_name])
|
|
self.assert_json_error(result, 'Unable to access stream (Saxony).')
|
|
|
|
# authorization_errors_fatal=False works
|
|
email = "othello@zulip.com"
|
|
self.login(email)
|
|
result = self.common_subscribe_to_streams(email, [stream_name],
|
|
extra_post_data={'authorization_errors_fatal': ujson.dumps(False)})
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
self.assertEqual(json["unauthorized"], [stream_name])
|
|
self.assertEqual(json["subscribed"], {})
|
|
self.assertEqual(json["already_subscribed"], {})
|
|
|
|
# Inviting another user to an invite-only stream is allowed
|
|
email = 'hamlet@zulip.com'
|
|
self.login(email)
|
|
result = self.common_subscribe_to_streams(
|
|
email, [stream_name],
|
|
extra_post_data={'principals': ujson.dumps(["othello@zulip.com"])})
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
self.assertEqual(json["subscribed"], {"othello@zulip.com": [stream_name]})
|
|
self.assertEqual(json["already_subscribed"], {})
|
|
|
|
# Make sure both users are subscribed to this stream
|
|
result = self.client.get("/api/v1/streams/%s/members" % (stream_name,),
|
|
**self.api_auth(email))
|
|
self.assert_json_success(result)
|
|
json = ujson.loads(result.content)
|
|
|
|
self.assertTrue('othello@zulip.com' in json['subscribers'])
|
|
self.assertTrue('hamlet@zulip.com' in json['subscribers'])
|
|
|
|
class GetSubscribersTest(AuthedTestCase):
|
|
|
|
def setUp(self):
|
|
# type: () -> None
|
|
self.email = "hamlet@zulip.com"
|
|
self.user_profile = get_user_profile_by_email(self.email)
|
|
self.login(self.email)
|
|
|
|
def check_well_formed_result(self, result, stream_name, domain):
|
|
# type: (Dict[str, Any], text_type, text_type) -> None
|
|
"""
|
|
A successful call to get_subscribers returns the list of subscribers in
|
|
the form:
|
|
|
|
{"msg": "",
|
|
"result": "success",
|
|
"subscribers": ["hamlet@zulip.com", "prospero@zulip.com"]}
|
|
"""
|
|
self.assertIn("subscribers", result)
|
|
self.assertIsInstance(result["subscribers"], list)
|
|
true_subscribers = [user_profile.email for user_profile in self.users_subscribed_to_stream(
|
|
stream_name, domain)]
|
|
self.assertEqual(sorted(result["subscribers"]), sorted(true_subscribers))
|
|
|
|
def make_subscriber_request(self, stream_name, email=None):
|
|
# type: (text_type, Optional[str]) -> HttpResponse
|
|
if email is None:
|
|
email = self.email
|
|
return self.client.get("/api/v1/streams/%s/members" % (stream_name,),
|
|
**self.api_auth(email))
|
|
|
|
def make_successful_subscriber_request(self, stream_name):
|
|
# type: (text_type) -> None
|
|
result = self.make_subscriber_request(stream_name)
|
|
self.assert_json_success(result)
|
|
self.check_well_formed_result(ujson.loads(result.content),
|
|
stream_name, self.user_profile.realm.domain)
|
|
|
|
def test_subscriber(self):
|
|
# type: () -> None
|
|
"""
|
|
get_subscribers returns the list of subscribers.
|
|
"""
|
|
stream_name = gather_subscriptions(self.user_profile)[0][0]['name']
|
|
self.make_successful_subscriber_request(stream_name)
|
|
|
|
@slow(0.15, "common_subscribe_to_streams is slow")
|
|
def test_gather_subscriptions(self):
|
|
# type: () -> None
|
|
"""
|
|
gather_subscriptions returns correct results with only 3 queries
|
|
"""
|
|
realm = get_realm("zulip.com")
|
|
streams = ["stream_%s" % i for i in range(10)]
|
|
for stream in streams:
|
|
create_stream_if_needed(realm, stream)
|
|
users_to_subscribe = [self.email, "othello@zulip.com", "cordelia@zulip.com"]
|
|
ret = self.common_subscribe_to_streams(
|
|
self.email,
|
|
streams,
|
|
dict(principals=ujson.dumps(users_to_subscribe)))
|
|
self.assert_json_success(ret)
|
|
ret = self.common_subscribe_to_streams(
|
|
self.email,
|
|
["stream_invite_only_1"],
|
|
dict(principals=ujson.dumps(users_to_subscribe)),
|
|
invite_only=True)
|
|
self.assert_json_success(ret)
|
|
|
|
with queries_captured() as queries:
|
|
subscriptions = gather_subscriptions(self.user_profile)
|
|
self.assertTrue(len(subscriptions[0]) >= 11)
|
|
for sub in subscriptions[0]:
|
|
if not sub["name"].startswith("stream_"):
|
|
continue
|
|
self.assertTrue(len(sub["subscribers"]) == len(users_to_subscribe))
|
|
self.assert_length(queries, 4, exact=True)
|
|
|
|
@slow(0.15, "common_subscribe_to_streams is slow")
|
|
def test_never_subscribed_streams(self):
|
|
# type: () -> None
|
|
"""
|
|
Check never_subscribed streams are fetched correctly and not include invite_only streams.
|
|
"""
|
|
realm = get_realm("zulip.com")
|
|
streams = ["stream_%s" % i for i in range(10)]
|
|
for stream in streams:
|
|
create_stream_if_needed(realm, stream)
|
|
users_to_subscribe = ["othello@zulip.com", "cordelia@zulip.com"]
|
|
ret = self.common_subscribe_to_streams(
|
|
self.email,
|
|
streams,
|
|
dict(principals=ujson.dumps(users_to_subscribe)))
|
|
self.assert_json_success(ret)
|
|
ret = self.common_subscribe_to_streams(
|
|
self.email,
|
|
["stream_invite_only_1"],
|
|
dict(principals=ujson.dumps(users_to_subscribe)),
|
|
invite_only=True)
|
|
self.assert_json_success(ret)
|
|
with queries_captured() as queries:
|
|
subscribed, unsubscribed, never_subscribed, email_dict = gather_subscriptions_helper(self.user_profile)
|
|
self.assertTrue(len(never_subscribed) >= 10)
|
|
|
|
# Invite only stream should not be there in never_subscribed streams
|
|
for stream_dict in never_subscribed:
|
|
if stream_dict["name"].startswith("stream_"):
|
|
self.assertFalse(stream_dict['name'] == "stream_invite_only_1")
|
|
self.assertTrue(len(stream_dict["subscribers"]) == len(users_to_subscribe))
|
|
self.assert_length(queries, 4, exact=True)
|
|
|
|
@slow(0.15, "common_subscribe_to_streams is slow")
|
|
def test_gather_subscriptions_mit(self):
|
|
# type: () -> None
|
|
"""
|
|
gather_subscriptions returns correct results with only 3 queries
|
|
"""
|
|
# Subscribe only ourself because invites are disabled on mit.edu
|
|
users_to_subscribe = ["starnine@mit.edu", "espuser@mit.edu"]
|
|
for email in users_to_subscribe:
|
|
self.subscribe_to_stream(email, "mit_stream")
|
|
|
|
ret = self.common_subscribe_to_streams(
|
|
"starnine@mit.edu",
|
|
["mit_invite_only"],
|
|
dict(principals=ujson.dumps(users_to_subscribe)),
|
|
invite_only=True)
|
|
self.assert_json_success(ret)
|
|
|
|
with queries_captured() as queries:
|
|
subscriptions = gather_subscriptions(get_user_profile_by_email("starnine@mit.edu"))
|
|
|
|
self.assertTrue(len(subscriptions[0]) >= 2)
|
|
for sub in subscriptions[0]:
|
|
if not sub["name"].startswith("mit_"):
|
|
continue
|
|
if sub["name"] == "mit_invite_only":
|
|
self.assertTrue(len(sub["subscribers"]) == len(users_to_subscribe))
|
|
else:
|
|
self.assertTrue(len(sub["subscribers"]) == 0)
|
|
self.assert_length(queries, 4, exact=True)
|
|
|
|
def test_nonsubscriber(self):
|
|
# type: () -> None
|
|
"""
|
|
Even a non-subscriber to a public stream can query a stream's membership
|
|
with get_subscribers.
|
|
"""
|
|
# Create a stream for which Hamlet is the only subscriber.
|
|
stream_name = "Saxony"
|
|
self.common_subscribe_to_streams(self.email, [stream_name])
|
|
other_email = "othello@zulip.com"
|
|
|
|
# Fetch the subscriber list as a non-member.
|
|
self.login(other_email)
|
|
self.make_successful_subscriber_request(stream_name)
|
|
|
|
def test_subscriber_private_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
A subscriber to a private stream can query that stream's membership.
|
|
"""
|
|
stream_name = "Saxony"
|
|
self.common_subscribe_to_streams(self.email, [stream_name],
|
|
invite_only=True)
|
|
self.make_successful_subscriber_request(stream_name)
|
|
|
|
def test_json_get_subscribers_stream_not_exist(self):
|
|
# type: () -> None
|
|
"""
|
|
json_get_subscribers also returns the list of subscribers for a stream.
|
|
"""
|
|
stream_name = "unknown_stream"
|
|
result = self.client.post("/json/get_subscribers", {"stream": stream_name})
|
|
self.assert_json_error(result, "Stream does not exist: %s" % (stream_name,))
|
|
|
|
def test_json_get_subscribers(self):
|
|
# type: () -> None
|
|
"""
|
|
json_get_subscribers in zerver/views/streams.py
|
|
also returns the list of subscribers for a stream.
|
|
"""
|
|
stream_name = gather_subscriptions(self.user_profile)[0][0]['name']
|
|
expected_subscribers = gather_subscriptions(self.user_profile)[0][0]['subscribers']
|
|
result = self.client.post("/json/get_subscribers", {"stream": stream_name})
|
|
self.assert_json_success(result)
|
|
result_dict = ujson.loads(result.content)
|
|
self.assertIn('subscribers', result_dict)
|
|
self.assertIsInstance(result_dict['subscribers'], list)
|
|
subscribers = [] # type: List[text_type]
|
|
for subscriber in result_dict['subscribers']:
|
|
self.assertIsInstance(subscriber, six.string_types)
|
|
subscribers.append(subscriber)
|
|
self.assertEqual(set(subscribers), set(expected_subscribers))
|
|
|
|
def test_nonsubscriber_private_stream(self):
|
|
# type: () -> None
|
|
"""
|
|
A non-subscriber to a private stream can't query that stream's membership.
|
|
"""
|
|
# Create a private stream for which Hamlet is the only subscriber.
|
|
stream_name = "NewStream"
|
|
self.common_subscribe_to_streams(self.email, [stream_name],
|
|
invite_only=True)
|
|
other_email = "othello@zulip.com"
|
|
|
|
# Try to fetch the subscriber list as a non-member.
|
|
result = self.make_subscriber_request(stream_name, email=other_email)
|
|
self.assert_json_error(result,
|
|
"Unable to retrieve subscribers for invite-only stream")
|