bots: Add mypy annotations for bots framework.
This commit adds mypy annotations for both the main
bots and the bots testing runner. It involves a change
to the BotHandlerApi send_message and update_message
funtions, which is compatible with every bot.
Tweaked by tabbott to use more expressive annotations.
MODIFY
contrib_bots/bot_lib.py
contrib_bots/bot_lib.py
@@ -7,6 +7,11 @@ import sys
import time
import re
+if False:
+ from mypy_extensions import NoReturn
+from typing import Any, Optional, List, Dict
+from types import ModuleType
+
our_dir = os.path.dirname(os.path.abspath(__file__))
# For dev setups, we can find the API in the repo itself.
@@ -16,17 +21,20 @@ if os.path.exists(os.path.join(our_dir, '../api/zulip')):
from zulip import Client
def exit_gracefully(signum, frame):
+ # type: (int, Optional[Any]) -> None
sys.exit(0)
class RateLimit(object):
def __init__(self, message_limit, interval_limit):
+ # type: (int, int) -> None
self.message_limit = message_limit
self.interval_limit = interval_limit
- self.message_list = []
+ self.message_list = [] # type: List[float]
self.error_message = '-----> !*!*!*MESSAGE RATE LIMIT REACHED, EXITING*!*!*! <-----\n'
'Is your bot trapped in an infinite loop by reacting to its own messages?'
def is_legal(self):
+ # type: () -> bool
self.message_list.append(time.time())
if len(self.message_list) > self.message_limit:
self.message_list.pop(0)
@@ -36,12 +44,14 @@ class RateLimit(object):
return True
def show_error_and_exit(self):
+ # type: () -> NoReturn
logging.error(self.error_message)
sys.exit(1)
class BotHandlerApi(object):
def __init__(self, client):
+ # type: (Client) -> None
# Only expose a subset of our Client's functionality
user_profile = client.get_profile()
self._rate_limit = RateLimit(20, 5)
@@ -54,19 +64,22 @@ class BotHandlerApi(object):
' up the zuliprc file correctly.')
sys.exit(1)
- def send_message(self, *args, **kwargs):
+ def send_message(self, message):
+ # type: (Dict[str, Any]) -> Dict[str, Any]
if self._rate_limit.is_legal():
- return self._client.send_message(*args, **kwargs)
+ return self._client.send_message(message)
else:
self._rate_limit.show_error_and_exit()
- def update_message(self, *args, **kwargs):
+ def update_message(self, message):
+ # type: (Dict[str, Any]) -> Dict[str, Any]
if self._rate_limit.is_legal():
- return self._client.update_message(*args, **kwargs)
+ return self._client.update_message(message)
else:
self._rate_limit.show_error_and_exit()
def send_reply(self, message, response):
+ # type: (Dict[str, Any], str) -> Dict[str, Any]
if message['type'] == 'private':
return self.send_message(dict(
type='private',
@@ -83,15 +96,25 @@ class BotHandlerApi(object):
class StateHandler(object):
def __init__(self):
- self.state = None
+ # type: () -> None
+ self.state = None # type: Any
def set_state(self, state):
+ # type: (Any) -> None
self.state = state
def get_state(self):
+ # type: () -> Any
return self.state
def run_message_handler_for_bot(lib_module, quiet, config_file):
+ # type: (Any, bool, str) -> Any
+ #
+ # lib_module is of type Any, since it can contain any bot's
+ # handler class. Eventually, we want bot's handler classes to
+ # inherit from a common prototype specifying the handle_message
+ # function.
+ #
# Make sure you set up your ~/.zuliprc
client = Client(config_file=config_file)
restricted_client = BotHandlerApi(client)
@@ -104,6 +127,7 @@ def run_message_handler_for_bot(lib_module, quiet, config_file):
print(message_handler.usage())
def extract_query_without_mention(message, client):
+ # type: (Dict[str, Any], BotHandlerApi) -> str
"""
If the bot is the first @mention in the message, then this function returns
the message with the bot's @mention removed. Otherwise, it returns None.
@@ -116,6 +140,7 @@ def run_message_handler_for_bot(lib_module, quiet, config_file):
return query_without_mention.lstrip()
def is_private(message, client):
+ # type: (Dict[str, Any], BotHandlerApi) -> bool
# bot will not reply if the sender name is the same as the bot name
# to prevent infinite loop
if message['type'] == 'private':
@@ -123,6 +148,7 @@ def run_message_handler_for_bot(lib_module, quiet, config_file):
return False
def handle_message(message):
+ # type: (Dict[str, Any]) -> None
logging.info('waiting for next message')
# is_mentioned is true if the bot is mentioned at ANY position (not necessarily

MODIFY
contrib_bots/bots_test_lib.py
contrib_bots/bots_test_lib.py
@@ -16,20 +16,27 @@ from six.moves import zip
from unittest import TestCase
+from typing import List, Dict, Any
+from types import ModuleType
+
current_dir = os.path.dirname(os.path.abspath(__file__))
class BotTestCase(TestCase):
- bot_name = None
+ bot_name = '' # type: str
def assert_bot_output(self, request, response):
- # type: (str, str) -> None
+ # type: (Dict[str, Any], str) -> None
bot_module = os.path.join(current_dir, "bots",
self.bot_name, self.bot_name + ".py")
self.bot_test(messages=[request], bot_module=bot_module,
bot_response=[response])
def mock_test(self, messages, message_handler, bot_response):
- # type: (List[Dict[str, str]], Function) -> None
+ # message_handler is of type Any, since it can contain any bot's
+ # handler class. Eventually, we want bot's handler classes to
+ # inherit from a common prototype specifying the handle_message
+ # function.
+ # type: (List[Dict[str, Any]], Any, List[str]) -> None
# Mocking BotHandlerApi
with patch('contrib_bots.bot_lib.BotHandlerApi') as MockClass:
instance = MockClass.return_value
@@ -43,11 +50,13 @@ class BotTestCase(TestCase):
instance.send_reply.assert_called_with(message, response)
def bot_to_run(self, bot_module):
- # type: None -> Function
+ # Returning Any, same argument as in mock_test function.
+ # type: (str) -> Any
lib_module = get_lib_module(bot_module)
message_handler = lib_module.handler_class()
return message_handler
def bot_test(self, messages, bot_module, bot_response):
+ # type: (List[Dict[str, Any]], str, List[str]) -> None
message_handler = self.bot_to_run(bot_module)
self.mock_test(messages=messages, message_handler=message_handler, bot_response=bot_response)

MODIFY
contrib_bots/run.py
contrib_bots/run.py
@@ -7,6 +7,7 @@ import logging
import optparse
import os
import sys
+from types import ModuleType
our_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, our_dir)
@@ -14,6 +15,7 @@ sys.path.insert(0, our_dir)
from bot_lib import run_message_handler_for_bot
def get_lib_module(bots_fn):
+ # type: (str) -> ModuleType
bots_fn = os.path.realpath(bots_fn)
if not os.path.dirname(bots_fn).startswith(os.path.join(our_dir, 'bots')):
print('Sorry, we will only import code from contrib_bots/bots.')
@@ -29,6 +31,7 @@ def get_lib_module(bots_fn):
return module
def run():
+ # type: () -> None
usage = '''
./run.py <lib file>
Example: ./run.py lib/followup.py

MODIFY
contrib_bots/test-bots
contrib_bots/test-bots
@@ -20,9 +20,12 @@ if __name__ == '__main__':
sys.path.insert(0, root_dir)
- loader = unittest.TestLoader()
+ # mypy doesn't recognize the TestLoader attribute, even though the code
+ # is executable
+ loader = unittest.TestLoader() # type: ignore
suite = loader.discover(start_dir=bots_test_dir, top_level_dir=root_dir)
runner = unittest.TextTestRunner(verbosity=2)
- result = runner.run(suite)
+ # same issue as for TestLoader
+ result = runner.run(suite) # type: ignore
if result.errors or result.failures:
raise Exception('Test failed!')

bots: Add mypy annotations for bots framework.

This commit adds mypy annotations for both the main
bots and the bots testing runner. It involves a change
to the BotHandlerApi send_message and update_message
funtions, which is compatible with every bot.

Tweaked by tabbott to use more expressive annotations.