tests: Add type annotations to test_lib.

Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg 2023-11-13 13:27:42 -08:00
parent 6aedfe6457
commit 9c44fe5d3a
3 changed files with 55 additions and 47 deletions

View file

@ -21,8 +21,6 @@ exclude = [
# fully annotate their bots. # fully annotate their bots.
"zulip_bots/zulip_bots/bots", "zulip_bots/zulip_bots/bots",
"zulip_bots/zulip_bots/bots_unmaintained", "zulip_bots/zulip_bots/bots_unmaintained",
# Excluded out of laziness:
"zulip_bots/zulip_bots/tests/test_lib.py",
] ]
# These files will be included even if excluded by a rule above. # These files will be included even if excluded by a rule above.

View file

@ -213,8 +213,8 @@ class ExternalBotHandler:
def __init__( def __init__(
self, self,
client: Client, client: Client,
root_dir: str, root_dir: Optional[str],
bot_details: Dict[str, Any], bot_details: Optional[Dict[str, Any]],
bot_config_file: Optional[str] = None, bot_config_file: Optional[str] = None,
bot_config_parser: Optional[configparser.ConfigParser] = None, bot_config_parser: Optional[configparser.ConfigParser] = None,
) -> None: ) -> None:
@ -363,6 +363,7 @@ class ExternalBotHandler:
return self._client.upload_file(file) return self._client.upload_file(file)
def open(self, filepath: str) -> IO[str]: def open(self, filepath: str) -> IO[str]:
assert self._root_dir is not None
filepath = os.path.normpath(filepath) filepath = os.path.normpath(filepath)
abs_filepath = os.path.join(self._root_dir, filepath) abs_filepath = os.path.join(self._root_dir, filepath)
if abs_filepath.startswith(self._root_dir): if abs_filepath.startswith(self._root_dir):
@ -434,8 +435,8 @@ def prepare_message_handler(bot: str, bot_handler: BotHandler, bot_lib_module: A
def run_message_handler_for_bot( def run_message_handler_for_bot(
lib_module: Any, lib_module: Any,
quiet: bool, quiet: bool,
config_file: str, config_file: Optional[str],
bot_config_file: str, bot_config_file: Optional[str],
bot_name: str, bot_name: str,
bot_source: str, bot_source: str,
) -> Any: ) -> Any:
@ -459,6 +460,7 @@ def run_message_handler_for_bot(
try: try:
client = Client(config_file=config_file, client=client_name) client = Client(config_file=config_file, client=client_name)
except configparser.Error as e: except configparser.Error as e:
assert config_file is not None
display_config_file_errors(str(e), config_file) display_config_file_errors(str(e), config_file)
sys.exit(1) sys.exit(1)

View file

@ -1,8 +1,11 @@
import io import io
from typing import IO, Any, Callable, Dict, List, Optional, Set, Tuple, cast
from unittest import TestCase from unittest import TestCase
from unittest.mock import ANY, MagicMock, create_autospec, patch from unittest.mock import ANY, MagicMock, create_autospec, patch
from zulip import Client
from zulip_bots.lib import ( from zulip_bots.lib import (
BotHandler,
ExternalBotHandler, ExternalBotHandler,
StateHandler, StateHandler,
extract_query_without_mention, extract_query_without_mention,
@ -12,10 +15,10 @@ from zulip_bots.lib import (
class FakeClient: class FakeClient:
def __init__(self, *args, **kwargs): def __init__(self, *args: object, **kwargs: object) -> None:
self.storage = dict() self.storage: Dict[str, str] = dict()
def get_profile(self): def get_profile(self) -> Dict[str, Any]:
return dict( return dict(
user_id="alice", user_id="alice",
full_name="Alice", full_name="Alice",
@ -23,7 +26,7 @@ class FakeClient:
id=42, id=42,
) )
def update_storage(self, payload): def update_storage(self, payload: Dict[str, Any]) -> Dict[str, Any]:
new_data = payload["storage"] new_data = payload["storage"]
self.storage.update(new_data) self.storage.update(new_data)
@ -31,45 +34,45 @@ class FakeClient:
result="success", result="success",
) )
def get_storage(self, request): def get_storage(self, request: Dict[str, Any]) -> Dict[str, Any]:
return dict( return dict(
result="success", result="success",
storage=self.storage, storage=self.storage,
) )
def send_message(self, message): def send_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
return dict( return dict(
result="success", result="success",
) )
def upload_file(self, file): def upload_file(self, file: IO[Any]) -> None:
pass pass
class FakeBotHandler: class FakeBotHandler:
def usage(self): def usage(self) -> str:
return """ return """
This is a fake bot handler that is used This is a fake bot handler that is used
to spec BotHandler mocks. to spec BotHandler mocks.
""" """
def handle_message(self, message, bot_handler): def handle_message(self, message: Dict[str, str], bot_handler: BotHandler) -> None:
pass pass
class LibTest(TestCase): class LibTest(TestCase):
def test_basics(self): def test_basics(self) -> None:
client = FakeClient() client = cast(Client, FakeClient())
handler = ExternalBotHandler( handler = ExternalBotHandler(
client=client, root_dir=None, bot_details=None, bot_config_file=None client=client, root_dir=None, bot_details=None, bot_config_file=None
) )
message = None message: Dict[str, Any] = {}
handler.send_message(message) handler.send_message(message)
def test_state_handler(self): def test_state_handler(self) -> None:
client = FakeClient() client = cast(Client, FakeClient())
state_handler = StateHandler(client) state_handler = StateHandler(client)
state_handler.put("key", [1, 2, 3]) state_handler.put("key", [1, 2, 3])
@ -81,7 +84,7 @@ class LibTest(TestCase):
val = state_handler.get("key") val = state_handler.get("key")
self.assertEqual(val, [1, 2, 3]) self.assertEqual(val, [1, 2, 3])
def test_state_handler_by_mock(self): def test_state_handler_by_mock(self) -> None:
client = MagicMock() client = MagicMock()
state_handler = StateHandler(client) state_handler = StateHandler(client)
@ -109,8 +112,8 @@ class LibTest(TestCase):
client.get_storage.assert_not_called() client.get_storage.assert_not_called()
self.assertEqual(val, [5]) self.assertEqual(val, [5])
def test_react(self): def test_react(self) -> None:
client = FakeClient() client = cast(Client, FakeClient())
handler = ExternalBotHandler( handler = ExternalBotHandler(
client=client, root_dir=None, bot_details=None, bot_config_file=None client=client, root_dir=None, bot_details=None, bot_config_file=None
) )
@ -121,18 +124,18 @@ class LibTest(TestCase):
"emoji_name": "wave", "emoji_name": "wave",
"reaction_type": "unicode_emoji", "reaction_type": "unicode_emoji",
} }
client.add_reaction = MagicMock() client.add_reaction = MagicMock() # type: ignore[method-assign]
handler.react(message, emoji_name) handler.react(message, emoji_name)
client.add_reaction.assert_called_once_with(dict(expected)) client.add_reaction.assert_called_once_with(dict(expected))
def test_send_reply(self): def test_send_reply(self) -> None:
client = FakeClient() client = cast(Client, FakeClient())
profile = client.get_profile() profile = client.get_profile()
handler = ExternalBotHandler( handler = ExternalBotHandler(
client=client, root_dir=None, bot_details=None, bot_config_file=None client=client, root_dir=None, bot_details=None, bot_config_file=None
) )
to = {"id": 43} to = {"id": 43}
expected = [ expected: List[Tuple[Dict[str, Any], Dict[str, Any], Optional[str]]] = [
( (
{"type": "private", "display_recipient": [to]}, {"type": "private", "display_recipient": [to]},
{"type": "private", "to": [to["id"]]}, {"type": "private", "to": [to["id"]]},
@ -151,18 +154,18 @@ class LibTest(TestCase):
] ]
response_text = "Response" response_text = "Response"
for test in expected: for test in expected:
client.send_message = MagicMock() client.send_message = MagicMock() # type: ignore[method-assign]
handler.send_reply(test[0], response_text, test[2]) handler.send_reply(test[0], response_text, test[2])
client.send_message.assert_called_once_with( client.send_message.assert_called_once_with(
dict(test[1], content=response_text, widget_content=test[2]) dict(test[1], content=response_text, widget_content=test[2])
) )
def test_content_and_full_content(self): def test_content_and_full_content(self) -> None:
client = FakeClient() client = cast(Client, FakeClient())
client.get_profile() client.get_profile()
ExternalBotHandler(client=client, root_dir=None, bot_details=None, bot_config_file=None) ExternalBotHandler(client=client, root_dir=None, bot_details=None, bot_config_file=None)
def test_run_message_handler_for_bot(self): def test_run_message_handler_for_bot(self) -> None:
with patch("zulip_bots.lib.Client", new=FakeClient) as fake_client: with patch("zulip_bots.lib.Client", new=FakeClient) as fake_client:
mock_lib_module = MagicMock() mock_lib_module = MagicMock()
# __file__ is not mocked by MagicMock(), so we assign a mock value manually. # __file__ is not mocked by MagicMock(), so we assign a mock value manually.
@ -170,8 +173,13 @@ class LibTest(TestCase):
mock_bot_handler = create_autospec(FakeBotHandler) mock_bot_handler = create_autospec(FakeBotHandler)
mock_lib_module.handler_class.return_value = mock_bot_handler mock_lib_module.handler_class.return_value = mock_bot_handler
def call_on_each_event_mock(self, callback, event_types=None, narrow=None): def call_on_each_event_mock(
def test_message(message, flags): self: FakeClient,
callback: Callable[[Dict[str, Any]], None],
event_types: Optional[List[str]] = None,
narrow: Optional[List[List[str]]] = None,
) -> None:
def test_message(message: Dict[str, Any], flags: Set[str]) -> None:
event = {"message": message, "flags": flags, "type": "message"} event = {"message": message, "flags": flags, "type": "message"}
callback(event) callback(event)
@ -188,8 +196,8 @@ class LibTest(TestCase):
message=expected_message, bot_handler=ANY message=expected_message, bot_handler=ANY
) )
fake_client.call_on_each_event = call_on_each_event_mock.__get__( fake_client.call_on_each_event = call_on_each_event_mock.__get__( # type: ignore[attr-defined]
fake_client, fake_client.__class__ fake_client, type(fake_client)
) )
run_message_handler_for_bot( run_message_handler_for_bot(
lib_module=mock_lib_module, lib_module=mock_lib_module,
@ -200,25 +208,25 @@ class LibTest(TestCase):
bot_source="bot code location", bot_source="bot code location",
) )
def test_upload_file(self): def test_upload_file(self) -> None:
client, handler = self._create_client_and_handler_for_file_upload() client, handler = self._create_client_and_handler_for_file_upload()
file = io.BytesIO(b"binary") file = io.BytesIO(b"binary")
handler.upload_file(file) handler.upload_file(file)
client.upload_file.assert_called_once_with(file) client.upload_file.assert_called_once_with(file) # type: ignore[attr-defined]
def test_upload_file_from_path(self): def test_upload_file_from_path(self) -> None:
client, handler = self._create_client_and_handler_for_file_upload() client, handler = self._create_client_and_handler_for_file_upload()
file = io.BytesIO(b"binary") file = io.BytesIO(b"binary")
with patch("builtins.open", return_value=file): with patch("builtins.open", return_value=file):
handler.upload_file_from_path("file.txt") handler.upload_file_from_path("file.txt")
client.upload_file.assert_called_once_with(file) client.upload_file.assert_called_once_with(file) # type: ignore[attr-defined]
def test_extract_query_without_mention(self): def test_extract_query_without_mention(self) -> None:
client = FakeClient() client = cast(Client, FakeClient())
handler = ExternalBotHandler( handler = ExternalBotHandler(
client=client, root_dir=None, bot_details=None, bot_config_file=None client=client, root_dir=None, bot_details=None, bot_config_file=None
) )
@ -231,12 +239,12 @@ class LibTest(TestCase):
message = {"content": "Not at start @**Alice|alice** Hello World"} message = {"content": "Not at start @**Alice|alice** Hello World"}
self.assertEqual(extract_query_without_mention(message, handler), None) self.assertEqual(extract_query_without_mention(message, handler), None)
def test_is_private_message_but_not_group_pm(self): def test_is_private_message_but_not_group_pm(self) -> None:
client = FakeClient() client = cast(Client, FakeClient())
handler = ExternalBotHandler( handler = ExternalBotHandler(
client=client, root_dir=None, bot_details=None, bot_config_file=None client=client, root_dir=None, bot_details=None, bot_config_file=None
) )
message = {} message: Dict[str, Any] = {}
message["display_recipient"] = "some stream" message["display_recipient"] = "some stream"
message["type"] = "stream" message["type"] = "stream"
self.assertFalse(is_private_message_but_not_group_pm(message, handler)) self.assertFalse(is_private_message_but_not_group_pm(message, handler))
@ -249,9 +257,9 @@ class LibTest(TestCase):
message["display_recipient"] = [{"email": "a1@b.com"}, {"email": "a2@b.com"}] message["display_recipient"] = [{"email": "a1@b.com"}, {"email": "a2@b.com"}]
self.assertFalse(is_private_message_but_not_group_pm(message, handler)) self.assertFalse(is_private_message_but_not_group_pm(message, handler))
def _create_client_and_handler_for_file_upload(self): def _create_client_and_handler_for_file_upload(self) -> Tuple[Client, ExternalBotHandler]:
client = FakeClient() client = cast(Client, FakeClient())
client.upload_file = MagicMock() client.upload_file = MagicMock() # type: ignore[method-assign]
handler = ExternalBotHandler( handler = ExternalBotHandler(
client=client, root_dir=None, bot_details=None, bot_config_file=None client=client, root_dir=None, bot_details=None, bot_config_file=None