import json import os import unittest from collections import OrderedDict from importlib import import_module from pathlib import Path from types import ModuleType from typing import Any, Dict from unittest import mock from zulip_bots.finder import metadata from zulip_bots.lib import BotHandler from zulip_botserver import server from zulip_botserver.input_parameters import parse_args from .server_test_lib import BotServerTestCase class BotServerTests(BotServerTestCase): class MockMessageHandler: def handle_message(self, message: Dict[str, str], bot_handler: BotHandler) -> None: assert message == {"key": "test message"} class MockLibModule: def handler_class(self) -> Any: return BotServerTests.MockMessageHandler() def setUp(self) -> None: # Since initializing Client invokes `get_server_settings` that fails in the test # environment, we need to mock it to pretend that there exists a backend. super().setUp() self.patch = mock.patch("zulip.Client.get_server_settings", return_value=mock.Mock()) self.patch.start() def test_successful_request(self) -> None: available_bots = ["helloworld"] bots_config = { "helloworld": { "email": "helloworld-bot@zulip.com", "key": "123456789qwertyuiop", "site": "http://localhost", "token": "abcd1234", } } self.assert_bot_server_response( available_bots=available_bots, bots_config=bots_config, event=dict( message={"content": "@**test** test message"}, bot_email="helloworld-bot@zulip.com", trigger="mention", token="abcd1234", ), expected_response="beep boop", check_success=True, ) def test_successful_request_from_two_bots(self) -> None: available_bots = ["helloworld", "help"] bots_config = { "helloworld": { "email": "helloworld-bot@zulip.com", "key": "123456789qwertyuiop", "site": "http://localhost", "token": "abcd1234", }, "help": { "email": "help-bot@zulip.com", "key": "123456789qwertyuiop", "site": "http://localhost", "token": "abcd1234", }, } self.assert_bot_server_response( available_bots=available_bots, event=dict( message={"content": "@**test** test message"}, bot_email="helloworld-bot@zulip.com", trigger="mention", token="abcd1234", ), expected_response="beep boop", bots_config=bots_config, check_success=True, ) def test_request_for_unkown_bot(self) -> None: bots_config = { "helloworld": { "email": "helloworld-bot@zulip.com", "key": "123456789qwertyuiop", "site": "http://localhost", "token": "abcd1234", }, } self.assert_bot_server_response( available_bots=["helloworld"], event=dict(message={"content": "test message"}, bot_email="unknown-bot@zulip.com"), bots_config=bots_config, check_success=False, ) def test_wrong_bot_token(self) -> None: available_bots = ["helloworld"] bots_config = { "helloworld": { "email": "helloworld-bot@zulip.com", "key": "123456789qwertyuiop", "site": "http://localhost", "token": "abcd1234", } } self.assert_bot_server_response( available_bots=available_bots, bots_config=bots_config, event=dict( message={"content": "@**test** test message"}, bot_email="helloworld-bot@zulip.com", trigger="mention", token="wrongtoken", ), check_success=False, ) @mock.patch("logging.error") @mock.patch("zulip_bots.lib.StateHandler") def test_wrong_bot_credentials( self, mock_StateHandler: mock.Mock, mock_LoggingError: mock.Mock ) -> None: available_bots = ["nonexistent-bot"] bots_config = { "nonexistent-bot": { "email": "helloworld-bot@zulip.com", "key": "123456789qwertyuiop", "site": "http://localhost", "token": "abcd1234", } } with self.assertRaisesRegex( SystemExit, 'Error: Bot "nonexistent-bot" doesn\'t exist. Please make ' "sure you have set up the botserverrc file correctly.", ): self.assert_bot_server_response( available_bots=available_bots, event=dict( message={"content": "@**test** test message"}, bot_email="helloworld-bot@zulip.com", trigger="mention", token="abcd1234", ), bots_config=bots_config, ) @mock.patch("sys.argv", ["zulip-botserver", "--config-file", "/foo/bar/baz.conf"]) def test_argument_parsing_defaults(self) -> None: opts = parse_args() assert opts.config_file == "/foo/bar/baz.conf" assert opts.bot_name is None assert opts.bot_config_file is None assert opts.hostname == "127.0.0.1" assert opts.port == 5002 def test_read_config_from_env_vars(self) -> None: # We use an OrderedDict so that the order of the entries in # the stringified environment variable is standard even on # Python 3.7 and earlier. bots_config = OrderedDict() bots_config["hello_world"] = { "email": "helloworld-bot@zulip.com", "key": "value", "site": "http://localhost", "token": "abcd1234", } bots_config["giphy"] = { "email": "giphy-bot@zulip.com", "key": "value2", "site": "http://localhost", "token": "abcd1234", } os.environ["ZULIP_BOTSERVER_CONFIG"] = json.dumps(bots_config) # No bot specified; should read all bot configs assert server.read_config_from_env_vars() == bots_config # Specified bot exists; should read only that section. assert server.read_config_from_env_vars("giphy") == {"giphy": bots_config["giphy"]} # Specified bot doesn't exist; should read the first section of the config. assert server.read_config_from_env_vars("redefined_bot") == { "redefined_bot": bots_config["hello_world"] } def test_read_config_file(self) -> None: with self.assertRaises(IOError): server.read_config_file("nonexistentfile.conf") current_dir = os.path.dirname(os.path.abspath(__file__)) # No bot specified; should read all bot configs. bot_conf1 = server.read_config_file(os.path.join(current_dir, "test.conf")) expected_config1 = { "helloworld": { "email": "helloworld-bot@zulip.com", "key": "value", "site": "http://localhost", "token": "abcd1234", }, "giphy": { "email": "giphy-bot@zulip.com", "key": "value2", "site": "http://localhost", "token": "abcd1234", }, } assert json.dumps(bot_conf1, sort_keys=True) == json.dumps(expected_config1, sort_keys=True) # Specified bot exists; should read only that section. bot_conf3 = server.read_config_file(os.path.join(current_dir, "test.conf"), "giphy") expected_config3 = { "giphy": { "email": "giphy-bot@zulip.com", "key": "value2", "site": "http://localhost", "token": "abcd1234", } } assert json.dumps(bot_conf3, sort_keys=True) == json.dumps(expected_config3, sort_keys=True) # Specified bot doesn't exist; should read the first section of the config. bot_conf2 = server.read_config_file(os.path.join(current_dir, "test.conf"), "redefined_bot") expected_config2 = { "redefined_bot": { "email": "helloworld-bot@zulip.com", "key": "value", "site": "http://localhost", "token": "abcd1234", } } assert json.dumps(bot_conf2, sort_keys=True) == json.dumps(expected_config2, sort_keys=True) def test_load_lib_modules(self) -> None: # This testcase requires hardcoded paths, which here is a good thing so if we ever # restructure zulip_bots, this test would fail and we would also update Botserver # at the same time. helloworld = import_module("zulip_bots.bots.{bot}.{bot}".format(bot="helloworld")) root_dir = Path(__file__).parents[2].as_posix() # load valid module name module = server.load_lib_modules(["helloworld"])["helloworld"] assert module == helloworld # load valid file path path = Path( root_dir, "zulip_bots/zulip_bots/bots/{bot}/{bot}.py".format(bot="helloworld") ).as_posix() module = server.load_lib_modules([path])[path] assert module.__name__ == "custom_bot_module" assert module.__file__ == path assert isinstance(module, ModuleType) # load invalid module name with self.assertRaisesRegex( SystemExit, 'Error: Bot "botserver-test-case-random-bot" doesn\'t exist. ' "Please make sure you have set up the botserverrc file correctly.", ): module = server.load_lib_modules(["botserver-test-case-random-bot"])[ "botserver-test-case-random-bot" ] # load invalid file path with self.assertRaisesRegex( SystemExit, 'Error: Bot "{}/zulip_bots/zulip_bots/bots/helloworld.py" doesn\'t exist. ' "Please make sure you have set up the botserverrc file correctly.".format(root_dir), ): path = Path( root_dir, "zulip_bots/zulip_bots/bots/{bot}.py".format(bot="helloworld") ).as_posix() module = server.load_lib_modules([path])[path] @mock.patch("zulip_botserver.server.app") @mock.patch("sys.argv", ["zulip-botserver", "--config-file", "/foo/bar/baz.conf"]) def test_load_from_registry(self, mock_app: mock.Mock) -> None: packaged_bot_module = mock.MagicMock(__version__="1.0.0", __file__="asd") packaged_bot_entrypoint = metadata.EntryPoint( "packaged_bot", "module_name", "zulip_bots.registry" ) bots_config = { "packaged_bot": { "email": "packaged-bot@zulip.com", "key": "value", "site": "http://localhost", "token": "abcd1234", } } with mock.patch( "zulip_botserver.server.read_config_file", return_value=bots_config ), mock.patch("zulip_botserver.server.lib.ExternalBotHandler", new=mock.Mock()), mock.patch( "zulip_bots.finder.metadata.EntryPoint.load", return_value=packaged_bot_module, ), mock.patch( "zulip_bots.finder.metadata.entry_points", return_value=(packaged_bot_entrypoint,), ): server.main() mock_app.config.__setitem__.assert_any_call( "BOTS_LIB_MODULES", {"packaged_bot": packaged_bot_module} ) if __name__ == "__main__": unittest.main()