mirror of
https://github.com/zulip/zulip.git
synced 2025-11-03 05:23:35 +00:00
fead14951c
This commit was split by tabbott; this piece covers the vast majority
of files in Zulip, but excludes scripts/, tools/, and puppet/ to help
ensure we at least show the right error messages for Xenial systems.
We can likely further refine the remaining pieces with some testing.
Generated by com2ann, with whitespace fixes and various manual fixes
for runtime issues:
- invoiced_through: Optional[LicenseLedger] = models.ForeignKey(
+ invoiced_through: Optional["LicenseLedger"] = models.ForeignKey(
-_apns_client: Optional[APNsClient] = None
+_apns_client: Optional["APNsClient"] = None
- notifications_stream: Optional[Stream] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE)
- signup_notifications_stream: Optional[Stream] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE)
+ notifications_stream: Optional["Stream"] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE)
+ signup_notifications_stream: Optional["Stream"] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE)
- author: Optional[UserProfile] = models.ForeignKey('UserProfile', blank=True, null=True, on_delete=CASCADE)
+ author: Optional["UserProfile"] = models.ForeignKey('UserProfile', blank=True, null=True, on_delete=CASCADE)
- bot_owner: Optional[UserProfile] = models.ForeignKey('self', null=True, on_delete=models.SET_NULL)
+ bot_owner: Optional["UserProfile"] = models.ForeignKey('self', null=True, on_delete=models.SET_NULL)
- default_sending_stream: Optional[Stream] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE)
- default_events_register_stream: Optional[Stream] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE)
+ default_sending_stream: Optional["Stream"] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE)
+ default_events_register_stream: Optional["Stream"] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE)
-descriptors_by_handler_id: Dict[int, ClientDescriptor] = {}
+descriptors_by_handler_id: Dict[int, "ClientDescriptor"] = {}
-worker_classes: Dict[str, Type[QueueProcessingWorker]] = {}
-queues: Dict[str, Dict[str, Type[QueueProcessingWorker]]] = {}
+worker_classes: Dict[str, Type["QueueProcessingWorker"]] = {}
+queues: Dict[str, Dict[str, Type["QueueProcessingWorker"]]] = {}
-AUTH_LDAP_REVERSE_EMAIL_SEARCH: Optional[LDAPSearch] = None
+AUTH_LDAP_REVERSE_EMAIL_SEARCH: Optional["LDAPSearch"] = None
Signed-off-by: Anders Kaseorg <anders@zulipchat.com>
320 lines
12 KiB
Python
320 lines
12 KiB
Python
import ujson
|
|
from typing import Any, Mapping, List
|
|
|
|
from zerver.lib.test_helpers import (
|
|
tornado_redirected_to_list,
|
|
queries_captured,
|
|
)
|
|
from zerver.lib.test_classes import (
|
|
ZulipTestCase,
|
|
)
|
|
from zerver.models import (
|
|
Huddle,
|
|
get_huddle_hash,
|
|
)
|
|
|
|
class TypingValidateOperatorTest(ZulipTestCase):
|
|
def test_missing_parameter(self) -> None:
|
|
"""
|
|
Sending typing notification without op parameter fails
|
|
"""
|
|
sender = self.example_user("hamlet")
|
|
params = dict(
|
|
to=ujson.dumps([sender.id]),
|
|
)
|
|
result = self.api_post(sender, '/api/v1/typing', params)
|
|
self.assert_json_error(result, 'Missing \'op\' argument')
|
|
|
|
def test_invalid_parameter(self) -> None:
|
|
"""
|
|
Sending typing notification with invalid value for op parameter fails
|
|
"""
|
|
sender = self.example_user("hamlet")
|
|
params = dict(
|
|
to=ujson.dumps([sender.id]),
|
|
op='foo'
|
|
)
|
|
result = self.api_post(sender, '/api/v1/typing', params)
|
|
self.assert_json_error(result, 'Invalid \'op\' value (should be start or stop)')
|
|
|
|
class TypingValidateUsersTest(ZulipTestCase):
|
|
def test_empty_array(self) -> None:
|
|
"""
|
|
Sending typing notification without recipient fails
|
|
"""
|
|
sender = self.example_user("hamlet")
|
|
result = self.api_post(sender, '/api/v1/typing', {'op': 'start', 'to': '[]'})
|
|
self.assert_json_error(result, 'Missing parameter: \'to\' (recipient)')
|
|
|
|
def test_missing_recipient(self) -> None:
|
|
"""
|
|
Sending typing notification without recipient fails
|
|
"""
|
|
sender = self.example_user("hamlet")
|
|
result = self.api_post(sender, '/api/v1/typing', {'op': 'start'})
|
|
self.assert_json_error(result, "Missing parameter: 'to' (recipient)")
|
|
|
|
def test_argument_to_is_not_valid_json(self) -> None:
|
|
"""
|
|
Sending typing notification to invalid recipient fails
|
|
"""
|
|
sender = self.example_user("hamlet")
|
|
invalid = 'bad email'
|
|
result = self.api_post(sender, '/api/v1/typing', {'op': 'start', 'to': invalid})
|
|
self.assert_json_error(result, "Invalid email 'bad email'")
|
|
|
|
def test_bogus_user_id(self) -> None:
|
|
"""
|
|
Sending typing notification to invalid recipient fails
|
|
"""
|
|
sender = self.example_user("hamlet")
|
|
invalid = '[9999999]'
|
|
result = self.api_post(sender, '/api/v1/typing', {'op': 'start', 'to': invalid})
|
|
self.assert_json_error(result, 'Invalid user ID 9999999')
|
|
|
|
class TypingHappyPathTest(ZulipTestCase):
|
|
def test_start_to_single_recipient(self) -> None:
|
|
sender = self.example_user('hamlet')
|
|
recipient_user = self.example_user('othello')
|
|
expected_recipients = {sender, recipient_user}
|
|
expected_recipient_emails = {user.email for user in expected_recipients}
|
|
expected_recipient_ids = {user.id for user in expected_recipients}
|
|
|
|
params = dict(
|
|
to=ujson.dumps([recipient_user.id]),
|
|
op='start',
|
|
)
|
|
|
|
events: List[Mapping[str, Any]] = []
|
|
with queries_captured() as queries:
|
|
with tornado_redirected_to_list(events):
|
|
result = self.api_post(sender, '/api/v1/typing', params)
|
|
|
|
self.assert_json_success(result)
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(len(queries), 4)
|
|
|
|
event = events[0]['event']
|
|
event_recipient_emails = {user['email'] for user in event['recipients']}
|
|
event_user_ids = set(events[0]['users'])
|
|
event_recipient_user_ids = {user['user_id'] for user in event['recipients']}
|
|
|
|
self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
|
|
self.assertEqual(expected_recipient_ids, event_user_ids)
|
|
self.assertEqual(event['sender']['email'], sender.email)
|
|
self.assertEqual(event_recipient_emails, expected_recipient_emails)
|
|
self.assertEqual(event['type'], 'typing')
|
|
self.assertEqual(event['op'], 'start')
|
|
|
|
def test_start_to_multiple_recipients(self) -> None:
|
|
sender = self.example_user('hamlet')
|
|
recipient_users = [self.example_user('othello'), self.example_user('cordelia')]
|
|
expected_recipients = set(recipient_users) | {sender}
|
|
expected_recipient_emails = {user.email for user in expected_recipients}
|
|
expected_recipient_ids = {user.id for user in expected_recipients}
|
|
|
|
huddle_hash = get_huddle_hash(list(expected_recipient_ids))
|
|
self.assertFalse(Huddle.objects.filter(huddle_hash=huddle_hash).exists())
|
|
|
|
events: List[Mapping[str, Any]] = []
|
|
|
|
params = dict(
|
|
to=ujson.dumps([user.id for user in recipient_users]),
|
|
op='start',
|
|
)
|
|
|
|
with queries_captured() as queries:
|
|
with tornado_redirected_to_list(events):
|
|
result = self.api_post(sender, '/api/v1/typing', params)
|
|
self.assert_json_success(result)
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(len(queries), 5)
|
|
|
|
# We should not be adding new Huddles just because
|
|
# a user started typing in the compose box. Let's
|
|
# wait till they send an actual message.
|
|
self.assertFalse(Huddle.objects.filter(huddle_hash=huddle_hash).exists())
|
|
|
|
event = events[0]['event']
|
|
event_recipient_emails = {user['email'] for user in event['recipients']}
|
|
event_user_ids = set(events[0]['users'])
|
|
event_recipient_user_ids = {user['user_id'] for user in event['recipients']}
|
|
|
|
self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
|
|
self.assertEqual(expected_recipient_ids, event_user_ids)
|
|
self.assertEqual(event['sender']['email'], sender.email)
|
|
self.assertEqual(event_recipient_emails, expected_recipient_emails)
|
|
self.assertEqual(event['type'], 'typing')
|
|
self.assertEqual(event['op'], 'start')
|
|
|
|
def test_start_to_self(self) -> None:
|
|
"""
|
|
Sending typing notification to yourself (using user IDs)
|
|
is successful.
|
|
"""
|
|
user = self.example_user('hamlet')
|
|
email = user.email
|
|
expected_recipient_emails = {email}
|
|
expected_recipient_ids = {user.id}
|
|
events: List[Mapping[str, Any]] = []
|
|
with tornado_redirected_to_list(events):
|
|
result = self.api_post(
|
|
user,
|
|
'/api/v1/typing',
|
|
{
|
|
'to': ujson.dumps([user.id]),
|
|
'op': 'start'
|
|
}
|
|
)
|
|
self.assert_json_success(result)
|
|
self.assertEqual(len(events), 1)
|
|
|
|
event = events[0]['event']
|
|
event_recipient_emails = {user['email'] for user in event['recipients']}
|
|
event_user_ids = set(events[0]['users'])
|
|
event_recipient_user_ids = {user['user_id'] for user in event['recipients']}
|
|
|
|
self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
|
|
self.assertEqual(expected_recipient_ids, event_user_ids)
|
|
self.assertEqual(event_recipient_emails, expected_recipient_emails)
|
|
self.assertEqual(event['sender']['email'], email)
|
|
self.assertEqual(event['type'], 'typing')
|
|
self.assertEqual(event['op'], 'start')
|
|
|
|
def test_start_to_another_user(self) -> None:
|
|
"""
|
|
Sending typing notification to another user
|
|
is successful.
|
|
"""
|
|
sender = self.example_user('hamlet')
|
|
recipient = self.example_user('othello')
|
|
expected_recipients = {sender, recipient}
|
|
expected_recipient_emails = {user.email for user in expected_recipients}
|
|
expected_recipient_ids = {user.id for user in expected_recipients}
|
|
|
|
params = dict(
|
|
to=ujson.dumps([recipient.id]),
|
|
op='start'
|
|
)
|
|
|
|
events: List[Mapping[str, Any]] = []
|
|
with tornado_redirected_to_list(events):
|
|
result = self.api_post(sender, '/api/v1/typing', params)
|
|
|
|
self.assert_json_success(result)
|
|
self.assertEqual(len(events), 1)
|
|
|
|
event = events[0]['event']
|
|
event_recipient_emails = {user['email'] for user in event['recipients']}
|
|
event_user_ids = set(events[0]['users'])
|
|
event_recipient_user_ids = {user['user_id'] for user in event['recipients']}
|
|
|
|
self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
|
|
self.assertEqual(expected_recipient_ids, event_user_ids)
|
|
self.assertEqual(event_recipient_emails, expected_recipient_emails)
|
|
self.assertEqual(event['sender']['email'], sender.email)
|
|
self.assertEqual(event['type'], 'typing')
|
|
self.assertEqual(event['op'], 'start')
|
|
|
|
def test_stop_to_self(self) -> None:
|
|
"""
|
|
Sending stopped typing notification to yourself
|
|
is successful.
|
|
"""
|
|
user = self.example_user('hamlet')
|
|
email = user.email
|
|
expected_recipient_emails = {email}
|
|
expected_recipient_ids = {user.id}
|
|
|
|
events: List[Mapping[str, Any]] = []
|
|
with tornado_redirected_to_list(events):
|
|
params = dict(
|
|
to=ujson.dumps([user.id]),
|
|
op='stop'
|
|
)
|
|
result = self.api_post(user, '/api/v1/typing', params)
|
|
|
|
self.assert_json_success(result)
|
|
self.assertEqual(len(events), 1)
|
|
|
|
event = events[0]['event']
|
|
event_recipient_emails = {user['email'] for user in event['recipients']}
|
|
event_user_ids = set(events[0]['users'])
|
|
event_recipient_user_ids = {user['user_id'] for user in event['recipients']}
|
|
|
|
self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
|
|
self.assertEqual(expected_recipient_ids, event_user_ids)
|
|
self.assertEqual(event_recipient_emails, expected_recipient_emails)
|
|
self.assertEqual(event['sender']['email'], email)
|
|
self.assertEqual(event['type'], 'typing')
|
|
self.assertEqual(event['op'], 'stop')
|
|
|
|
def test_stop_to_another_user(self) -> None:
|
|
"""
|
|
Sending stopped typing notification to another user
|
|
is successful.
|
|
"""
|
|
sender = self.example_user('hamlet')
|
|
recipient = self.example_user('othello')
|
|
expected_recipients = {sender, recipient}
|
|
expected_recipient_emails = {user.email for user in expected_recipients}
|
|
expected_recipient_ids = {user.id for user in expected_recipients}
|
|
|
|
events: List[Mapping[str, Any]] = []
|
|
with tornado_redirected_to_list(events):
|
|
params = dict(
|
|
to=ujson.dumps([recipient.id]),
|
|
op='stop'
|
|
)
|
|
result = self.api_post(sender, '/api/v1/typing', params)
|
|
|
|
self.assert_json_success(result)
|
|
self.assertEqual(len(events), 1)
|
|
|
|
event = events[0]['event']
|
|
event_recipient_emails = {user['email'] for user in event['recipients']}
|
|
event_user_ids = set(events[0]['users'])
|
|
event_recipient_user_ids = {user['user_id'] for user in event['recipients']}
|
|
|
|
self.assertEqual(expected_recipient_ids, event_recipient_user_ids)
|
|
self.assertEqual(expected_recipient_ids, event_user_ids)
|
|
self.assertEqual(event_recipient_emails, expected_recipient_emails)
|
|
self.assertEqual(event['sender']['email'], sender.email)
|
|
self.assertEqual(event['type'], 'typing')
|
|
self.assertEqual(event['op'], 'stop')
|
|
|
|
class TypingLegacyMobileSupportTest(ZulipTestCase):
|
|
def test_legacy_email_interface(self) -> None:
|
|
'''
|
|
We are keeping the email interface on life support
|
|
for a couple months until we get some of our
|
|
mobile users upgraded.
|
|
'''
|
|
sender = self.example_user('hamlet')
|
|
othello = self.example_user('othello')
|
|
cordelia = self.example_user('cordelia')
|
|
|
|
emails = [othello.email, cordelia.email]
|
|
|
|
params = dict(
|
|
to=ujson.dumps(emails),
|
|
op='start',
|
|
)
|
|
|
|
events: List[Mapping[str, Any]] = []
|
|
with tornado_redirected_to_list(events):
|
|
result = self.api_post(sender, '/api/v1/typing', params)
|
|
|
|
self.assert_json_success(result)
|
|
event = events[0]['event']
|
|
|
|
event_recipient_user_ids = {
|
|
user['user_id']
|
|
for user in event['recipients']
|
|
}
|
|
|
|
self.assertEqual(
|
|
event_recipient_user_ids,
|
|
{sender.id, othello.id, cordelia.id}
|
|
)
|