Annotate zephyr_mirror_backend.py.

With some tweaks by tabbott to update tuple->Tuple and update comments.
This commit is contained in:
anirudhjain75 2016-12-07 00:33:23 +05:30 committed by Tim Abbott
parent 476a591ef5
commit 92981590f5

View file

@ -21,7 +21,8 @@
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE. # SOFTWARE.
from __future__ import absolute_import from __future__ import absolute_import
from typing import Any, List from typing import IO, Any, Text, Union, Set, Tuple
from types import FrameType
import sys import sys
from six.moves import map from six.moves import map
@ -52,6 +53,7 @@ CURRENT_STATE = States.Startup
logger = None # type: logging.Logger logger = None # type: logging.Logger
def to_zulip_username(zephyr_username): def to_zulip_username(zephyr_username):
# type: (str) -> str
if "@" in zephyr_username: if "@" in zephyr_username:
(user, realm) = zephyr_username.split("@") (user, realm) = zephyr_username.split("@")
else: else:
@ -64,6 +66,7 @@ def to_zulip_username(zephyr_username):
return user.lower() + "|" + realm.upper() + "@mit.edu" return user.lower() + "|" + realm.upper() + "@mit.edu"
def to_zephyr_username(zulip_username): def to_zephyr_username(zulip_username):
# type: (str) -> str
(user, realm) = zulip_username.split("@") (user, realm) = zulip_username.split("@")
if "|" not in user: if "|" not in user:
# Hack to make ctl's fake username setup work :) # Hack to make ctl's fake username setup work :)
@ -86,6 +89,7 @@ def to_zephyr_username(zulip_username):
# or (3) the first word of the next line is longer than this entire # or (3) the first word of the next line is longer than this entire
# line. # line.
def different_paragraph(line, next_line): def different_paragraph(line, next_line):
# type: (str, str) -> bool
words = next_line.split() words = next_line.split()
return (len(line + " " + words[0]) < len(next_line) * 0.8 or return (len(line + " " + words[0]) < len(next_line) * 0.8 or
len(line + " " + words[0]) < 50 or len(line + " " + words[0]) < 50 or
@ -94,6 +98,7 @@ def different_paragraph(line, next_line):
# Linewrapping algorithm based on: # Linewrapping algorithm based on:
# http://gcbenison.wordpress.com/2011/07/03/a-program-to-intelligently-remove-carriage-returns-so-you-can-paste-text-without-having-it-look-awful/ #ignorelongline # http://gcbenison.wordpress.com/2011/07/03/a-program-to-intelligently-remove-carriage-returns-so-you-can-paste-text-without-having-it-look-awful/ #ignorelongline
def unwrap_lines(body): def unwrap_lines(body):
# type: (str) -> str
lines = body.split("\n") lines = body.split("\n")
result = "" result = ""
previous_line = lines[0] previous_line = lines[0]
@ -117,6 +122,7 @@ def unwrap_lines(body):
return result return result
def send_zulip(zeph): def send_zulip(zeph):
# type: (Dict[str, str]) -> Dict[str, str]
message = {} message = {}
if options.forward_class_messages: if options.forward_class_messages:
message["forged"] = "yes" message["forged"] = "yes"
@ -149,6 +155,7 @@ def send_zulip(zeph):
return zulip_client.send_message(message) return zulip_client.send_message(message)
def send_error_zulip(error_msg): def send_error_zulip(error_msg):
# type: (str) -> None
message = {"type": "private", message = {"type": "private",
"sender": zulip_account_email, "sender": zulip_account_email,
"to": zulip_account_email, "to": zulip_account_email,
@ -158,6 +165,7 @@ def send_error_zulip(error_msg):
current_zephyr_subs = set() current_zephyr_subs = set()
def zephyr_bulk_subscribe(subs): def zephyr_bulk_subscribe(subs):
# type: (List[Tuple[str, str, str]]) -> None
try: try:
zephyr._z.subAll(subs) zephyr._z.subAll(subs)
except IOError: except IOError:
@ -193,6 +201,7 @@ def zephyr_bulk_subscribe(subs):
current_zephyr_subs.add(cls) current_zephyr_subs.add(cls)
def update_subscriptions(): def update_subscriptions():
# type: () -> None
try: try:
f = open(options.stream_file_path, "r") f = open(options.stream_file_path, "r")
public_streams = simplejson.loads(f.read()) public_streams = simplejson.loads(f.read())
@ -216,6 +225,7 @@ def update_subscriptions():
zephyr_bulk_subscribe(list(classes_to_subscribe)) zephyr_bulk_subscribe(list(classes_to_subscribe))
def maybe_kill_child(): def maybe_kill_child():
# type: () -> None
try: try:
if child_pid is not None: if child_pid is not None:
os.kill(child_pid, signal.SIGTERM) os.kill(child_pid, signal.SIGTERM)
@ -224,6 +234,7 @@ def maybe_kill_child():
logger.exception("") logger.exception("")
def maybe_restart_mirroring_script(): def maybe_restart_mirroring_script():
# type: () -> None
if os.stat(os.path.join(options.root_path, "stamps", "restart_stamp")).st_mtime > start_time or \ if os.stat(os.path.join(options.root_path, "stamps", "restart_stamp")).st_mtime > start_time or \
((options.user == "tabbott" or options.user == "tabbott/extra") and ((options.user == "tabbott" or options.user == "tabbott/extra") and
os.stat(os.path.join(options.root_path, "stamps", "tabbott_stamp")).st_mtime > start_time): os.stat(os.path.join(options.root_path, "stamps", "tabbott_stamp")).st_mtime > start_time):
@ -243,6 +254,7 @@ def maybe_restart_mirroring_script():
time.sleep(1) time.sleep(1)
def process_loop(log): def process_loop(log):
# type: (IO) -> None
restart_check_count = 0 restart_check_count = 0
last_check_time = time.time() last_check_time = time.time()
while True: while True:
@ -282,6 +294,7 @@ def process_loop(log):
logger.exception("Error updating subscriptions from Zulip:") logger.exception("Error updating subscriptions from Zulip:")
def parse_zephyr_body(zephyr_data): def parse_zephyr_body(zephyr_data):
# type: (str) -> Tuple[str, str]
try: try:
(zsig, body) = zephyr_data.split("\x00", 1) (zsig, body) = zephyr_data.split("\x00", 1)
except ValueError: except ValueError:
@ -289,6 +302,7 @@ def parse_zephyr_body(zephyr_data):
return (zsig, body) return (zsig, body)
def parse_crypt_table(zephyr_class, instance): def parse_crypt_table(zephyr_class, instance):
# type: (Text, str) -> str
try: try:
crypt_table = open(os.path.join(os.environ["HOME"], ".crypt-table")) crypt_table = open(os.path.join(os.environ["HOME"], ".crypt-table"))
except IOError: except IOError:
@ -310,6 +324,7 @@ def parse_crypt_table(zephyr_class, instance):
return None return None
def decrypt_zephyr(zephyr_class, instance, body): def decrypt_zephyr(zephyr_class, instance, body):
# type: (Text, str, str) -> str
keypath = parse_crypt_table(zephyr_class, instance) keypath = parse_crypt_table(zephyr_class, instance)
if keypath is None: if keypath is None:
# We can't decrypt it, so we just return the original body # We can't decrypt it, so we just return the original body
@ -340,6 +355,7 @@ def decrypt_zephyr(zephyr_class, instance, body):
return decrypted return decrypted
def process_notice(notice, log): def process_notice(notice, log):
# type: (zulip, IO) -> None
(zsig, body) = parse_zephyr_body(notice.message) (zsig, body) = parse_zephyr_body(notice.message)
is_personal = False is_personal = False
is_huddle = False is_huddle = False
@ -438,6 +454,8 @@ def process_notice(notice, log):
os._exit(0) os._exit(0)
def decode_unicode_byte_strings(zeph): def decode_unicode_byte_strings(zeph):
# type: (Dict[str, Any]) -> Dict[str, str]
# 'Any' can be of any type of text that is converted to str.
for field in zeph.keys(): for field in zeph.keys():
if isinstance(zeph[field], str): if isinstance(zeph[field], str):
try: try:
@ -448,11 +466,13 @@ def decode_unicode_byte_strings(zeph):
return zeph return zeph
def quit_failed_initialization(message): def quit_failed_initialization(message):
# type: (str) -> str
logger.error(message) logger.error(message)
maybe_kill_child() maybe_kill_child()
sys.exit(1) sys.exit(1)
def zephyr_init_autoretry(): def zephyr_init_autoretry():
# type: () -> None
backoff = zulip.RandomExponentialBackoff() backoff = zulip.RandomExponentialBackoff()
while backoff.keep_going(): while backoff.keep_going():
try: try:
@ -468,6 +488,7 @@ def zephyr_init_autoretry():
quit_failed_initialization("Could not initialize Zephyr library, quitting!") quit_failed_initialization("Could not initialize Zephyr library, quitting!")
def zephyr_load_session_autoretry(session_path): def zephyr_load_session_autoretry(session_path):
# type: (str) -> None
backoff = zulip.RandomExponentialBackoff() backoff = zulip.RandomExponentialBackoff()
while backoff.keep_going(): while backoff.keep_going():
try: try:
@ -483,6 +504,7 @@ def zephyr_load_session_autoretry(session_path):
quit_failed_initialization("Could not load saved Zephyr session, quitting!") quit_failed_initialization("Could not load saved Zephyr session, quitting!")
def zephyr_subscribe_autoretry(sub): def zephyr_subscribe_autoretry(sub):
# type: (Tuple[str, str, str]) -> None
backoff = zulip.RandomExponentialBackoff() backoff = zulip.RandomExponentialBackoff()
while backoff.keep_going(): while backoff.keep_going():
try: try:
@ -498,6 +520,7 @@ def zephyr_subscribe_autoretry(sub):
quit_failed_initialization("Could not subscribe to personals, quitting!") quit_failed_initialization("Could not subscribe to personals, quitting!")
def zephyr_to_zulip(options): def zephyr_to_zulip(options):
# type: (Any) -> None
if options.use_sessions and os.path.exists(options.session_path): if options.use_sessions and os.path.exists(options.session_path):
logger.info("Loading old session") logger.info("Loading old session")
zephyr_load_session_autoretry(options.session_path) zephyr_load_session_autoretry(options.session_path)
@ -549,6 +572,7 @@ def zephyr_to_zulip(options):
process_loop(None) process_loop(None)
def send_zephyr(zwrite_args, content): def send_zephyr(zwrite_args, content):
# type: (list, str) -> Tuple[int, str]
p = subprocess.Popen(zwrite_args, stdin=subprocess.PIPE, p = subprocess.Popen(zwrite_args, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = p.communicate(input=content.encode("utf-8")) stdout, stderr = p.communicate(input=content.encode("utf-8"))
@ -565,12 +589,15 @@ def send_zephyr(zwrite_args, content):
return (p.returncode, stderr) return (p.returncode, stderr)
def send_authed_zephyr(zwrite_args, content): def send_authed_zephyr(zwrite_args, content):
# type: (list[str], str) -> Tuple[int, str]
return send_zephyr(zwrite_args, content) return send_zephyr(zwrite_args, content)
def send_unauthed_zephyr(zwrite_args, content): def send_unauthed_zephyr(zwrite_args, content):
# type: (list[str], str) -> Tuple[int, str]
return send_zephyr(zwrite_args + ["-d"], content) return send_zephyr(zwrite_args + ["-d"], content)
def zcrypt_encrypt_content(zephyr_class, instance, content): def zcrypt_encrypt_content(zephyr_class, instance, content):
# type: (str, str, str) -> str
keypath = parse_crypt_table(zephyr_class, instance) keypath = parse_crypt_table(zephyr_class, instance)
if keypath is None: if keypath is None:
return None return None
@ -596,6 +623,8 @@ def zcrypt_encrypt_content(zephyr_class, instance, content):
return encrypted return encrypted
def forward_to_zephyr(message): def forward_to_zephyr(message):
# type: (Dict[str, Any]) -> None
# 'Any' can be of any type of text
support_heading = "Hi there! This is an automated message from Zulip." support_heading = "Hi there! This is an automated message from Zulip."
support_closing = """If you have any questions, please be in touch through the \ support_closing = """If you have any questions, please be in touch through the \
Feedback button or at support@zulipchat.com.""" Feedback button or at support@zulipchat.com."""
@ -656,7 +685,7 @@ Feedback button or at support@zulipchat.com."""
if message.get("invite_only_stream"): if message.get("invite_only_stream"):
result = zcrypt_encrypt_content(zephyr_class, instance, wrapped_content) result = zcrypt_encrypt_content(zephyr_class, instance, wrapped_content)
if result is None: if result is None:
return send_error_zulip("""%s send_error_zulip("""%s
Your Zulip-Zephyr mirror bot was unable to forward that last message \ Your Zulip-Zephyr mirror bot was unable to forward that last message \
from Zulip to Zephyr because you were sending to a zcrypted Zephyr \ from Zulip to Zephyr because you were sending to a zcrypted Zephyr \
@ -665,6 +694,7 @@ key (perhaps because your AFS tokens expired). That means that while \
Zulip users (like you) received it, Zephyr users did not. Zulip users (like you) received it, Zephyr users did not.
%s""" % (support_heading, support_closing)) %s""" % (support_heading, support_closing))
return
# Proceed with sending a zcrypted message # Proceed with sending a zcrypted message
wrapped_content = result wrapped_content = result
@ -679,7 +709,7 @@ Zulip users (like you) received it, Zephyr users did not.
if code == 0 and stderr == "": if code == 0 and stderr == "":
return return
elif code == 0: elif code == 0:
return send_error_zulip("""%s send_error_zulip("""%s
Your last message was successfully mirrored to zephyr, but zwrite \ Your last message was successfully mirrored to zephyr, but zwrite \
returned the following warning: returned the following warning:
@ -687,6 +717,7 @@ returned the following warning:
%s %s
%s""" % (support_heading, stderr, support_closing)) %s""" % (support_heading, stderr, support_closing))
return
elif code != 0 and (stderr.startswith("zwrite: Ticket expired while sending notice to ") or elif code != 0 and (stderr.startswith("zwrite: Ticket expired while sending notice to ") or
stderr.startswith("zwrite: No credentials cache found while sending notice to ")): stderr.startswith("zwrite: No credentials cache found while sending notice to ")):
# Retry sending the message unauthenticated; if that works, # Retry sending the message unauthenticated; if that works,
@ -695,7 +726,7 @@ returned the following warning:
if code == 0: if code == 0:
if options.ignore_expired_tickets: if options.ignore_expired_tickets:
return return
return send_error_zulip("""%s send_error_zulip("""%s
Your last message was forwarded from Zulip to Zephyr unauthenticated, \ Your last message was forwarded from Zulip to Zephyr unauthenticated, \
because your Kerberos tickets have expired. It was sent successfully, \ because your Kerberos tickets have expired. It was sent successfully, \
@ -704,11 +735,12 @@ are running the Zulip-Zephyr mirroring bot, so we can send \
authenticated Zephyr messages for you again. authenticated Zephyr messages for you again.
%s""" % (support_heading, support_closing)) %s""" % (support_heading, support_closing))
return
# zwrite failed and it wasn't because of expired tickets: This is # zwrite failed and it wasn't because of expired tickets: This is
# probably because the recipient isn't subscribed to personals, # probably because the recipient isn't subscribed to personals,
# but regardless, we should just notify the user. # but regardless, we should just notify the user.
return send_error_zulip("""%s send_error_zulip("""%s
Your Zulip-Zephyr mirror bot was unable to forward that last message \ Your Zulip-Zephyr mirror bot was unable to forward that last message \
from Zulip to Zephyr. That means that while Zulip users (like you) \ from Zulip to Zephyr. That means that while Zulip users (like you) \
@ -717,8 +749,11 @@ received it, Zephyr users did not. The error message from zwrite was:
%s %s
%s""" % (support_heading, stderr, support_closing)) %s""" % (support_heading, stderr, support_closing))
return
def maybe_forward_to_zephyr(message): def maybe_forward_to_zephyr(message):
# type: (Dict[str, Any]) -> None
# The key string can be used to direct any type of text.
if (message["sender_email"] == zulip_account_email): if (message["sender_email"] == zulip_account_email):
if not ((message["type"] == "stream") or if not ((message["type"] == "stream") or
(message["type"] == "private" and (message["type"] == "private" and
@ -740,6 +775,7 @@ def maybe_forward_to_zephyr(message):
logger.exception("Error forwarding message:") logger.exception("Error forwarding message:")
def zulip_to_zephyr(options): def zulip_to_zephyr(options):
# type: (int) -> None
# Sync messages from zulip to zephyr # Sync messages from zulip to zephyr
logger.info("Starting syncing messages.") logger.info("Starting syncing messages.")
while True: while True:
@ -750,6 +786,7 @@ def zulip_to_zephyr(options):
time.sleep(1) time.sleep(1)
def subscribed_to_mail_messages(): def subscribed_to_mail_messages():
# type: () -> bool
# In case we have lost our AFS tokens and those won't be able to # In case we have lost our AFS tokens and those won't be able to
# parse the Zephyr subs file, first try reading in result of this # parse the Zephyr subs file, first try reading in result of this
# query from the environment so we can avoid the filesystem read. # query from the environment so we can avoid the filesystem read.
@ -764,6 +801,7 @@ def subscribed_to_mail_messages():
return False return False
def add_zulip_subscriptions(verbose): def add_zulip_subscriptions(verbose):
# type: (bool) -> None
zephyr_subscriptions = set() zephyr_subscriptions = set()
skipped = set() skipped = set()
for (cls, instance, recipient) in parse_zephyr_subs(verbose=verbose): for (cls, instance, recipient) in parse_zephyr_subs(verbose=verbose):
@ -850,9 +888,11 @@ web interface.
""")) + "\n") """)) + "\n")
def valid_stream_name(name): def valid_stream_name(name):
# type: (str) -> bool
return name != "" return name != ""
def parse_zephyr_subs(verbose=False): def parse_zephyr_subs(verbose=False):
# type: (bool) -> Union[List, Tuple, Set[Tuple[str, str, str]]]
zephyr_subscriptions = set() zephyr_subscriptions = set()
subs_file = os.path.join(os.environ["HOME"], ".zephyr.subs") subs_file = os.path.join(os.environ["HOME"], ".zephyr.subs")
if not os.path.exists(subs_file): if not os.path.exists(subs_file):
@ -907,6 +947,7 @@ def open_logger():
return logger return logger
def configure_logger(logger, direction_name): def configure_logger(logger, direction_name):
# type: (logging.Logger, str) -> None
if direction_name is None: if direction_name is None:
log_format = "%(message)s" log_format = "%(message)s"
else: else:
@ -921,6 +962,7 @@ def configure_logger(logger, direction_name):
handler.setFormatter(formatter) handler.setFormatter(formatter)
def parse_args(): def parse_args():
# type: () -> Tuple
parser = optparse.OptionParser() parser = optparse.OptionParser()
parser.add_option('--forward-class-messages', parser.add_option('--forward-class-messages',
default=False, default=False,
@ -1004,6 +1046,7 @@ def parse_args():
return parser.parse_args() return parser.parse_args()
def die_gracefully(signal, frame): def die_gracefully(signal, frame):
# type: (int, FrameType) -> None
if CURRENT_STATE == States.ZulipToZephyr or CURRENT_STATE == States.ChildSending: if CURRENT_STATE == States.ZulipToZephyr or CURRENT_STATE == States.ChildSending:
# this is a child process, so we want os._exit (no clean-up necessary) # this is a child process, so we want os._exit (no clean-up necessary)
os._exit(1) os._exit(1)
@ -1029,6 +1072,9 @@ if __name__ == "__main__":
signal.signal(signal.SIGINT, die_gracefully) signal.signal(signal.SIGINT, die_gracefully)
# The properties available on 'options' are dynamically
# determined, so we have to treat it as an Any for type
# annotations.
(options, args) = parse_args() # type: Any, List[str] (options, args) = parse_args() # type: Any, List[str]
logger = open_logger() logger = open_logger()