<commit_before># -*- 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!")

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

class MockServiceHandler(OutgoingWebhookServiceInterface):
    def process_success(self, response_json: Dict[str, Any], event: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        # Our tests don't really look at the content yet.
        # They just ensure we use the "success" codepath.
        success_data = dict(
            content="whatever",
        )
        return success_data

service_handler = MockServiceHandler(None, None, None)

class DoRestCallTests(ZulipTestCase):
    def setUp(self) -> None:
        realm = get_realm("zulip")
        user_profile = get_user("<EMAIL>", 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.bot_user = self.example_user('outgoing_webhook_bot')
        logging.disable(logging.WARNING)

    @mock.patch('zerver.lib.outgoing_webhook.send_response_message')
    def test_successful_request(self, mock_send: mock.Mock) -> None:
        response = ResponseMock(200)
        with mock.patch('requests.request', return_value=response):
            do_rest_call('', None, self.mock_event, service_handler)
            self.assertTrue(mock_send.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('',  None, self.mock_event, service_handler)
            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)<commit_msg>bot tests: Replace use of MockServiceHandler.<commit_after># -*- 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, GenericOutgoingWebhookService
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!")

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

service_handler = GenericOutgoingWebhookService(None, None, None)

class DoRestCallTests(ZulipTestCase):
    def setUp(self) -> None:
        realm = get_realm("zulip")
        user_profile = get_user("<EMAIL>", 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.bot_user = self.example_user('outgoing_webhook_bot')
        logging.disable(logging.WARNING)

    @mock.patch('zerver.lib.outgoing_webhook.send_response_message')
    def test_successful_request(self, mock_send: mock.Mock) -> None:
        response = ResponseMock(200, dict(content='whatever'))
        with mock.patch('requests.request', return_value=response):
            do_rest_call('', None, self.mock_event, service_handler)
            self.assertTrue(mock_send.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('',  None, self.mock_event, service_handler)
            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)
{'fraction_non_alphanumeric': 0.10578397212543554, 'fraction_numerical': 0.006132404181184669, 'mean_word_length': 3.174520069808028, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 22, 'https://': 0, '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': '5619386', 'n_tokens_mistral': 2123, 'n_tokens_neox': 2050, 'n_words': 495}