<commit_before>from typing import Any, Dict
from unittest import mock

import orjson
import requests

from version import ZULIP_VERSION
from zerver.lib.actions import do_create_user
from zerver.lib.outgoing_webhook import (
    GenericOutgoingWebhookService,
    SlackOutgoingWebhookService,
    do_rest_call,
)
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.topic import TOPIC_NAME
from zerver.lib.url_encoding import near_message_url
from zerver.lib.users import add_service
from zerver.models import Recipient, Service, UserProfile, get_display_recipient, get_realm


class ResponseMock:
    def __init__(self, status_code: int, content: bytes = b"") -> None:
        self.status_code = status_code
        self.content = content
        self.text = content.decode()


def request_exception_error(http_method: Any, final_url: Any, **request_kwargs: Any) -> Any:
    raise requests.exceptions.RequestException("I'm a generic exception :(")


def timeout_error(http_method: Any, final_url: Any, **request_kwargs: Any) -> Any:
    raise requests.exceptions.Timeout("Time is up!")


def connection_error(http_method: Any, final_url: Any, **request_kwargs: Any) -> Any:
    raise requests.exceptions.ConnectionError()


class DoRestCallTests(ZulipTestCase):
    def mock_event(self, bot_user: UserProfile) -> Dict[str, Any]:
        return {
            # 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,
                TOPIC_NAME: "Foo",
                "id": "",
                "type": "stream",
            },
            "user_profile_id": bot_user.id,
            "command": "",
            "service_name": "",
        }

    def test_successful_request(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        response = ResponseMock(200, orjson.dumps(dict(content="whatever")))
        expect_200 = mock.patch("requests.request", return_value=response)

        expect_send_response = mock.patch("zerver.lib.outgoing_webhook.send_response_message")
        with expect_200, expect_send_response as mock_send:
            do_rest_call("", None, mock_event, service_handler)
        self.assertTrue(mock_send.called)

        for service_class in [GenericOutgoingWebhookService, SlackOutgoingWebhookService]:
            handler = service_class("token", bot_user, "service")

            with expect_200:
                do_rest_call("", None, mock_event, handler)

            # TODO: assert something interesting here?

    def test_retry_request(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        response = ResponseMock(500)
        with mock.patch("requests.request", return_value=response), self.assertLogs(
            level="WARNING"
        ) as m:
            final_response = do_rest_call("", None, mock_event, service_handler)
            assert final_response is not None

            self.assertEqual(
                m.output,
                [
                    f'WARNING:root:Message http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/ triggered an outgoing webhook, returning status code 500.\n Content of response (in quotes): "{final_response.text}"'
                ],
            )
        bot_owner_notification = self.get_last_message()
        self.assertEqual(
            bot_owner_notification.content,
            """[A message](http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/) to your bot @_**Outgoing Webhook** triggered an outgoing webhook.
The webhook got a response with status code *500*.""",
        )

        assert bot_user.bot_owner is not None
        self.assertEqual(bot_owner_notification.recipient_id, bot_user.bot_owner.recipient_id)

    def test_fail_request(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        response = ResponseMock(400)
        expect_400 = mock.patch("requests.request", return_value=response)
        expect_fail = mock.patch("zerver.lib.outgoing_webhook.fail_with_message")

        with expect_400, expect_fail as mock_fail, self.assertLogs(level="WARNING") as m:
            final_response = do_rest_call("", None, mock_event, service_handler)
            assert final_response is not None

            self.assertEqual(
                m.output,
                [
                    f'WARNING:root:Message http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/ triggered an outgoing webhook, returning status code 400.\n Content of response (in quotes): "{final_response.text}"'
                ],
            )

        self.assertTrue(mock_fail.called)

        bot_owner_notification = self.get_last_message()
        self.assertEqual(
            bot_owner_notification.content,
            """[A message](http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/) to your bot @_**Outgoing Webhook** triggered an outgoing webhook.
The webhook got a response with status code *400*.""",
        )

        assert bot_user.bot_owner is not None
        self.assertEqual(bot_owner_notification.recipient_id, bot_user.bot_owner.recipient_id)

    def test_headers(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        with mock.patch("requests.sessions.Session.send") as mock_send, self.assertLogs(
            level="WARNING"
        ) as m:
            final_response = do_rest_call(
                "https://example.com/", "payload-stub", mock_event, service_handler
            )
            assert final_response is not None

            self.assertEqual(
                m.output,
                [
                    f'WARNING:root:Message http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/ triggered an outgoing webhook, returning status code {final_response.status_code}.\n Content of response (in quotes): "{final_response.text}"'
                ],
            )

            mock_send.assert_called_once()
            prepared_request = mock_send.call_args[0][0]
            user_agent = "ZulipOutgoingWebhook/" + ZULIP_VERSION
            headers = {
                "Content-Type": "application/json",
                "User-Agent": user_agent,
            }
            self.assertLessEqual(headers.items(), prepared_request.headers.items())

    def test_error_handling(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")<commit_msg>outgoing_webhook: Remove warnings from test by giving a response code.<commit_after>from typing import Any, Dict
from unittest import mock

import orjson
import requests

from version import ZULIP_VERSION
from zerver.lib.actions import do_create_user
from zerver.lib.outgoing_webhook import (
    GenericOutgoingWebhookService,
    SlackOutgoingWebhookService,
    do_rest_call,
)
from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.topic import TOPIC_NAME
from zerver.lib.url_encoding import near_message_url
from zerver.lib.users import add_service
from zerver.models import Recipient, Service, UserProfile, get_display_recipient, get_realm


class ResponseMock:
    def __init__(self, status_code: int, content: bytes = b"") -> None:
        self.status_code = status_code
        self.content = content
        self.text = content.decode()


def request_exception_error(http_method: Any, final_url: Any, **request_kwargs: Any) -> Any:
    raise requests.exceptions.RequestException("I'm a generic exception :(")


def timeout_error(http_method: Any, final_url: Any, **request_kwargs: Any) -> Any:
    raise requests.exceptions.Timeout("Time is up!")


def connection_error(http_method: Any, final_url: Any, **request_kwargs: Any) -> Any:
    raise requests.exceptions.ConnectionError()


class DoRestCallTests(ZulipTestCase):
    def mock_event(self, bot_user: UserProfile) -> Dict[str, Any]:
        return {
            # 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,
                TOPIC_NAME: "Foo",
                "id": "",
                "type": "stream",
            },
            "user_profile_id": bot_user.id,
            "command": "",
            "service_name": "",
        }

    def test_successful_request(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        response = ResponseMock(200, orjson.dumps(dict(content="whatever")))
        expect_200 = mock.patch("requests.request", return_value=response)

        expect_send_response = mock.patch("zerver.lib.outgoing_webhook.send_response_message")
        with expect_200, expect_send_response as mock_send:
            do_rest_call("", None, mock_event, service_handler)
        self.assertTrue(mock_send.called)

        for service_class in [GenericOutgoingWebhookService, SlackOutgoingWebhookService]:
            handler = service_class("token", bot_user, "service")

            with expect_200:
                do_rest_call("", None, mock_event, handler)

            # TODO: assert something interesting here?

    def test_retry_request(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        response = ResponseMock(500)
        with mock.patch("requests.request", return_value=response), self.assertLogs(
            level="WARNING"
        ) as m:
            final_response = do_rest_call("", None, mock_event, service_handler)
            assert final_response is not None

            self.assertEqual(
                m.output,
                [
                    f'WARNING:root:Message http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/ triggered an outgoing webhook, returning status code 500.\n Content of response (in quotes): "{final_response.text}"'
                ],
            )
        bot_owner_notification = self.get_last_message()
        self.assertEqual(
            bot_owner_notification.content,
            """[A message](http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/) to your bot @_**Outgoing Webhook** triggered an outgoing webhook.
The webhook got a response with status code *500*.""",
        )

        assert bot_user.bot_owner is not None
        self.assertEqual(bot_owner_notification.recipient_id, bot_user.bot_owner.recipient_id)

    def test_fail_request(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        response = ResponseMock(400)
        expect_400 = mock.patch("requests.request", return_value=response)
        expect_fail = mock.patch("zerver.lib.outgoing_webhook.fail_with_message")

        with expect_400, expect_fail as mock_fail, self.assertLogs(level="WARNING") as m:
            final_response = do_rest_call("", None, mock_event, service_handler)
            assert final_response is not None

            self.assertEqual(
                m.output,
                [
                    f'WARNING:root:Message http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/ triggered an outgoing webhook, returning status code 400.\n Content of response (in quotes): "{final_response.text}"'
                ],
            )

        self.assertTrue(mock_fail.called)

        bot_owner_notification = self.get_last_message()
        self.assertEqual(
            bot_owner_notification.content,
            """[A message](http://zulip.testserver/#narrow/stream/999-Verona/topic/Foo/near/) to your bot @_**Outgoing Webhook** triggered an outgoing webhook.
The webhook got a response with status code *400*.""",
        )

        assert bot_user.bot_owner is not None
        self.assertEqual(bot_owner_notification.recipient_id, bot_user.bot_owner.recipient_id)

    def test_headers(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
        mock_event = self.mock_event(bot_user)
        service_handler = GenericOutgoingWebhookService("token", bot_user, "service")

        with mock.patch("requests.sessions.Session.send") as mock_send:
            mock_send.return_value = ResponseMock(200)
            final_response = do_rest_call(
                "https://example.com/", "payload-stub", mock_event, service_handler
            )
            assert final_response is not None

            mock_send.assert_called_once()
            prepared_request = mock_send.call_args[0][0]
            user_agent = "ZulipOutgoingWebhook/" + ZULIP_VERSION
            headers = {
                "Content-Type": "application/json",
                "User-Agent": user_agent,
            }
            self.assertLessEqual(headers.items(), prepared_request.headers.items())

    def test_error_handling(self) -> None:
        bot_user = self.example_user("outgoing_webhook_bot")
{'fraction_non_alphanumeric': 0.09776516495211068, 'fraction_numerical': 0.008087974459028023, 'mean_word_length': 2.779088471849866, 'pattern_counts': {'":': 22, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6531213', 'n_tokens_mistral': 4166, 'n_tokens_neox': 3932, 'n_words': 864}