Mirror
/
zulip
mirror of https://github.com/zulip/zulip.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

zerver/models.py: Annotate get_display_recipient. 

get_display_recipient's annotation clashes with other wrong annotations.
Fix those wrong annotations.
Since get_display_recipient returns a Union, use isinstance checks and
casts to make mypy checks succeed.
This commit is contained in:
Eklavya Sharma 2016-06-12 04:17:19 +05:30
parent 9f980c64b6
commit c654c4032d
5 changed files with 31 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
zerver/lib/email_mirror.py
View File

@ -126,7 +126,9 @@ def send_to_missed_message_address(address, message):
# Testing with basestring so we don't depend on the list return type from
# get_display_recipient
if not isinstance(display_recipient, six.string_types):
display_recipient = ','.join([user['email'] for user in display_recipient])
recipient_str = ','.join([user['email'] for user in display_recipient])
else:
recipient_str = display_recipient
body = filter_footer(extract_body(message))
body += extract_and_upload_attachments(message, user_profile.realm)
@ -139,7 +141,7 @@ def send_to_missed_message_address(address, message):
recipient_type_name = 'private'
internal_send_message(user_profile.email, recipient_type_name,
display_recipient, subject, body)
recipient_str, subject, body)
## Sending the Zulip ##

26
zerver/lib/notifications.py
View File

@ -1,5 +1,7 @@
from __future__ import print_function
from typing import Any, Tuple, Iterable, Optional
from six import text_type
from typing import cast, Any, Iterable, Mapping, Optional, Sequence, Tuple
import mandrill
from confirmation.models import Confirmation
@ -46,7 +48,7 @@ def one_click_unsubscribe_link(user_profile, endpoint):
return "%s/%s" % (base_url.rstrip("/"), resource_path)
def hashchange_encode(string):
# type: (str) -> str
# type: (text_type) -> text_type
# Do the same encoding operation as hashchange.encodeHashComponent on the
# frontend.
# `safe` has a default value of "/", but we want those encoded, too.
@ -54,20 +56,20 @@ def hashchange_encode(string):
string.encode("utf-8"), safe="").replace(".", "%2E").replace("%", ".")
def pm_narrow_url(participants):
# type: (List[str]) -> str
# type: (List[text_type]) -> text_type
participants.sort()
base_url = "https://%s/#narrow/pm-with/" % (settings.EXTERNAL_HOST,)
base_url = u"https://%s/#narrow/pm-with/" % (settings.EXTERNAL_HOST,)
return base_url + hashchange_encode(",".join(participants))
def stream_narrow_url(stream):
# type: (str) -> str
base_url = "https://%s/#narrow/stream/" % (settings.EXTERNAL_HOST,)
# type: (text_type) -> text_type
base_url = u"https://%s/#narrow/stream/" % (settings.EXTERNAL_HOST,)
return base_url + hashchange_encode(stream)
def topic_narrow_url(stream, topic):
# type: (str, str) -> str
base_url = "https://%s/#narrow/stream/" % (settings.EXTERNAL_HOST,)
return "%s%s/topic/%s" % (base_url, hashchange_encode(stream),
# type: (text_type, text_type) -> text_type
base_url = u"https://%s/#narrow/stream/" % (settings.EXTERNAL_HOST,)
return u"%s%s/topic/%s" % (base_url, hashchange_encode(stream),
hashchange_encode(topic))
def build_message_list(user_profile, messages):
@ -142,13 +144,14 @@ def build_message_list(user_profile, messages):
'content': [build_message_payload(message)]}
def message_header(user_profile, message):
# type: (UserProfile, Message) -> Dict[str, str]
# type: (UserProfile, Message) -> Dict[str, text_type]
disp_recipient = get_display_recipient(message.recipient)
if message.recipient.type == Recipient.PERSONAL:
header = "You and %s" % (message.sender.full_name)
html_link = pm_narrow_url([message.sender.email])
header_html = "<a style='color: #ffffff;' href='%s'>%s</a>" % (html_link, header)
elif message.recipient.type == Recipient.HUDDLE:
assert not isinstance(disp_recipient, text_type)
other_recipients = [r['full_name'] for r in disp_recipient
if r['email'] != user_profile.email]
header = "You and %s" % (", ".join(other_recipients),)
@ -156,6 +159,7 @@ def build_message_list(user_profile, messages):
if r["email"] != user_profile.email])
header_html = "<a style='color: #ffffff;' href='%s'>%s</a>" % (html_link, header)
else:
assert isinstance(disp_recipient, text_type)
header = "%s > %s" % (disp_recipient, message.subject)
stream_link = stream_narrow_url(disp_recipient)
topic_link = topic_narrow_url(disp_recipient, message.subject)
@ -303,7 +307,7 @@ def do_send_missedmessage_events(user_profile, missed_messages, message_count):
# If we have one huddle, set a reply-to to all of the members
# of the huddle except the user herself
disp_recipients = [", ".join(recipient['email']
for recipient in get_display_recipient(mesg.recipient)
for recipient in cast(Sequence[Mapping[str, Any]], get_display_recipient(mesg.recipient))
if recipient['email'] != user_profile.email)
for mesg in missed_messages]
if all(msg.recipient.type == Recipient.HUDDLE for msg in missed_messages) and \

4
zerver/lib/test_helpers.py
View File

@ -1,5 +1,5 @@
from __future__ import absolute_import
from typing import Any, Callable, Generator, Iterable, Tuple, Sized, Union, Optional
from typing import cast, Any, Callable, Generator, Iterable, Tuple, Sized, Union, Optional
from django.test import TestCase
from django.template import loader
@ -292,7 +292,7 @@ class AuthedTestCase(TestCase):
user_profile=user_profile,
active=True,
recipient__type=Recipient.STREAM)
return [get_display_recipient(sub.recipient) for sub in subs]
return [cast(text_type, get_display_recipient(sub.recipient)) for sub in subs]
def send_message(self, sender_name, raw_recipients, message_type,
content=u"test content", subject=u"test", **kwargs):

9
zerver/models.py
View File

@ -50,16 +50,16 @@ STREAM_NAMES = TypeVar('STREAM_NAMES', Sequence[text_type], AbstractSet[text_typ
# Doing 1000 remote cache requests to get_display_recipient is quite slow,
# so add a local cache as well as the remote cache cache.
per_request_display_recipient_cache = {} # type: Dict[int, List[Dict[text_type, Any]]]
per_request_display_recipient_cache = {} # type: Dict[int, List[Dict[str, Any]]]
def get_display_recipient_by_id(recipient_id, recipient_type, recipient_type_id):
## type: (int, int, int) -> Union[text_type, List[Dict[text_type, Any]]]
# type: (int, int, int) -> Union[text_type, List[Dict[str, Any]]]
if recipient_id not in per_request_display_recipient_cache:
result = get_display_recipient_remote_cache(recipient_id, recipient_type, recipient_type_id)
per_request_display_recipient_cache[recipient_id] = result
return per_request_display_recipient_cache[recipient_id]
def get_display_recipient(recipient):
## type: (Recipient) -> Union[text_type, List[Dict[text_type, Any]]]
# type: (Recipient) -> Union[text_type, List[Dict[str, Any]]]
return get_display_recipient_by_id(
recipient.id,
recipient.type,
@ -76,7 +76,7 @@ def flush_per_request_caches():
@cache_with_key(lambda *args: display_recipient_cache_key(args[0]),
timeout=3600*24*7)
def get_display_recipient_remote_cache(recipient_id, recipient_type, recipient_type_id):
## type: (int, int, int) -> Union[text_type, List[Dict[text_type, Any]]]
# type: (int, int, int) -> Union[text_type, List[Dict[str, Any]]]
"""
returns: an appropriate object describing the recipient. For a
stream this will be the stream name as a string. For a huddle or
@ -962,6 +962,7 @@ class Message(ModelReprMixin, models.Model):
if recipient_type == Recipient.STREAM:
display_type = "stream"
elif recipient_type in (Recipient.HUDDLE, Recipient.PERSONAL):
assert not isinstance(display_recipient, text_type)
display_type = "private"
if len(display_recipient) == 1:
# add the sender in if this isn't a message between

6
zerver/tests/test_subs.py
View File

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Mapping, Optional, Sequence
from zerver.lib import cache
@ -974,6 +974,10 @@ class SubscriptionAPITest(AuthedTestCase):
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):