# -*- coding: utf-8 -*- import ujson import logging import mock import requests from builtins import object from django.test import override_settings from requests import Response from typing import Any, Dict, Tuple, Optional from zerver.lib.outgoing_webhook import do_rest_call, OutgoingWebhookServiceInterface from zerver.lib.test_classes import ZulipTestCase from zerver.models import get_realm, get_user, UserProfile, get_display_recipient class ResponseMock: def __init__(self, status_code: int, content: Optional[Any]=None) -> None: self.status_code = status_code self.content = content self.text = ujson.dumps(content) def request_exception_error(http_method: Any, final_url: Any, data: Any, **request_kwargs: Any) -> Any: raise requests.exceptions.RequestException("I'm a generic exception :(") def timeout_error(http_method: Any, final_url: Any, data: Any, **request_kwargs: Any) -> Any: raise requests.exceptions.Timeout("Time is up!") class MockServiceHandler(OutgoingWebhookServiceInterface): def process_success(self, response: Response, event: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]: return "Success!", None service_handler = MockServiceHandler(None, None, None, None) class DoRestCallTests(ZulipTestCase): def setUp(self) -> None: realm = get_realm("zulip") user_profile = get_user("outgoing-webhook@zulip.com", realm) self.mock_event = { # In the tests there is no active queue processor, so retries don't get processed. # Therefore, we need to emulate `retry_event` in the last stage when the maximum # retries have been exceeded. 'failed_tries': 3, 'message': {'display_recipient': 'Verona', 'stream_id': 999, 'subject': 'Foo', 'id': '', 'type': 'stream'}, 'user_profile_id': user_profile.id, 'command': '', 'service_name': ''} self.rest_operation = {'method': "POST", 'relative_url_path': "", 'request_kwargs': {}, 'base_url': ""} self.bot_user = self.example_user('outgoing_webhook_bot') logging.disable(logging.WARNING) @mock.patch('zerver.lib.outgoing_webhook.succeed_with_message') def test_successful_request(self, mock_succeed_with_message: mock.Mock) -> None: response = ResponseMock(200) with mock.patch('requests.request', return_value=response): do_rest_call(self.rest_operation, None, self.mock_event, service_handler, None) self.assertTrue(mock_succeed_with_message.called) def test_retry_request(self: mock.Mock) -> None: response = ResponseMock(500) self.mock_event['failed_tries'] = 3 with mock.patch('requests.request', return_value=response): do_rest_call(self.rest_operation, None, self.mock_event, service_handler, None) bot_owner_notification = self.get_last_message() self.assertEqual(bot_owner_notification.content, '''[A message](http://zulip.testserver/#narrow/stream/999-Verona/subject/Foo/near/) triggered an outgoing webhook. The webhook got a response with status code *500*.''') self.assertEqual(bot_owner_notification.recipient_id, self.bot_user.bot_owner.id) self.mock_event['failed_tries'] = 0 @mock.patch('zerver.lib.outgoing_webhook.fail_with_message') def test_fail_request(self, mock_fail_with_message: mock.Mock) -> None: response = ResponseMock(400) with mock.patch('requests.request', return_value=response): do_rest_call(self.rest_operation, None, self.mock_event, service_handler, None) bot_owner_notification = self.get_last_message() self.assertTrue(mock_fail_with_message.called) self.assertEqual(bot_owner_notification.content, '''[A message](http://zulip.testserver/#narrow/stream/999-Verona/subject/Foo/near/) triggered an outgoing webhook. The webhook got a response with status code *400*.''') self.assertEqual(bot_owner_notification.recipient_id, self.bot_user.bot_owner.id) @mock.patch('logging.info') @mock.patch('requests.request', side_effect=timeout_error) def test_timeout_request(self, mock_requests_request: mock.Mock, mock_logger: mock.Mock) -> None: do_rest_call(self.rest_operation, None, self.mock_event, service_handler, None) bot_owner_notification = self.get_last_message() self.assertEqual(bot_owner_notification.content, '''[A message](http://zulip.testserver/#narrow/stream/999-Verona/subject/Foo/near/) triggered an outgoing webhook. When trying to send a request to the webhook service, an exception of type Timeout occurred: ``` Time is up! ```''') self.assertEqual(bot_owner_notification.recipient_id, self.bot_user.bot_owner.id) @mock.patch('logging.exception') @mock.patch('requests.request', side_effect=request_exception_error) @mock.patch('zerver.lib.outgoing_webhook.fail_with_message') def test_request_exception(self, mock_fail_with_message: mock.Mock, mock_requests_request: mock.Mock, mock_logger: mock.Mock) -> None: do_rest_call(self.rest_operation, None, self.mock_event, service_handler, None) bot_owner_notification = self.get_last_message() self.assertTrue(mock_fail_with_message.called) self.assertEqual(bot_owner_notification.content, '''[A message](http://zulip.testserver/#narrow/stream/999-Verona/subject/Foo/near/) triggered an outgoing webhook. When trying to send a request to the webhook service, an exception of type RequestException occurred: ``` I'm a generic exception :( ```''') self.assertEqual(bot_owner_notification.recipient_id, self.bot_user.bot_owner.id) class TestOutgoingWebhookMessaging(ZulipTestCase): def setUp(self) -> None: self.user_profile = self.example_user("othello") self.bot_profile = self.create_test_bot('outgoing-webhook', self.user_profile, full_name='Outgoing Webhook bot', bot_type=UserProfile.OUTGOING_WEBHOOK_BOT, service_name='foo-service') @mock.patch('requests.request', return_value=ResponseMock(200, {"response_string": "Hidley ho, I'm a webhook responding!"})) def test_pm_to_outgoing_webhook_bot(self, mock_requests_request: mock.Mock) -> None: self.send_personal_message(self.user_profile.email, self.bot_profile.email, content="foo") last_message = self.get_last_message() self.assertEqual(last_message.content, "Success! Hidley ho, I'm a webhook responding!") self.assertEqual(last_message.sender_id, self.bot_profile.id) display_recipient = get_display_recipient(last_message.recipient) # The next two lines error on mypy because the display_recipient is of type Union[str, List[Dict[str, Any]]]. # In this case, we know that display_recipient will be of type List[Dict[str, Any]]. # Otherwise this test will error, which is wanted behavior anyway. self.assert_length(display_recipient, 1) # type: ignore self.assertEqual(display_recipient[0]['email'], self.user_profile.email) # type: ignore @mock.patch('requests.request', return_value=ResponseMock(200, {"response_string": "Hidley ho, I'm a webhook responding!"})) def test_stream_message_to_outgoing_webhook_bot(self, mock_requests_request: mock.Mock) -> None: self.send_stream_message(self.user_profile.email, "Denmark", content="@**{}** foo".format(self.bot_profile.full_name), topic_name="bar") last_message = self.get_last_message() self.assertEqual(last_message.content, "Success! Hidley ho, I'm a webhook responding!") self.assertEqual(last_message.sender_id, self.bot_profile.id) self.assertEqual(last_message.subject, "bar") display_recipient = get_display_recipient(last_message.recipient) self.assertEqual(display_recipient, "Denmark")