path: root/tests/test_http_request.py

from textwrap import dedent
from unittest.mock import MagicMock, patch

import pytest
import weechat

from slack.http import HttpError, http_request, http_request_process, http_request_url
from slack.task import FutureProcess, FutureTimer, FutureUrl, weechat_task_cb
from slack.util import get_callback_name


@patch.object(weechat, "hook_url")
def test_http_request_success(mock_method: MagicMock):
    url = "http://example.com"
    options = {"option": "1"}
    timeout = 123
    coroutine = http_request(url, options, timeout)

    future = coroutine.send(None)
    assert isinstance(future, FutureUrl)

    mock_method.assert_called_once_with(
        url,
        options,
        timeout,
        get_callback_name(weechat_task_cb),
        future.id,
    )

    future.set_result(
        (url, options, {"response_code": "200", "headers": "", "output": "response"})
    )

    with pytest.raises(StopIteration) as excinfo:
        coroutine.send(None)
    assert excinfo.value.value == "response"


@patch.object(weechat, "hook_process_hashtable")
def test_http_request_process_success(mock_method: MagicMock):
    url = "http://example.com"
    options = {"option": "1"}
    timeout = 123
    coroutine = http_request_process(url, options, timeout)

    future = coroutine.send(None)
    assert isinstance(future, FutureProcess)

    mock_method.assert_called_once_with(
        f"url:{url}",
        {**options, "header": "1"},
        timeout,
        get_callback_name(weechat_task_cb),
        future.id,
    )

    body = f"HTTP/2 200\r\n\r\nresponse"
    future.set_result(("", 0, body, ""))

    with pytest.raises(StopIteration) as excinfo:
        coroutine.send(None)
    assert excinfo.value.value == (200, "HTTP/2 200", "response")


def test_http_request_url_error():
    url = "http://example.com"
    coroutine = http_request_url(url, {}, 0)

    future = coroutine.send(None)
    assert isinstance(future, FutureUrl)
    future.set_result((url, {}, {"error": "error"}))

    with pytest.raises(HttpError) as excinfo:
        coroutine.send(None)

    assert excinfo.value.url == url
    assert excinfo.value.return_code == None
    assert excinfo.value.http_status_code == None
    assert excinfo.value.error == "error"


def test_http_request_process_error_return_code():
    url = "http://example.com"
    coroutine = http_request_process(url, {}, 0)

    future = coroutine.send(None)
    assert isinstance(future, FutureProcess)
    future.set_result(("", -2, "", ""))

    with pytest.raises(HttpError) as excinfo:
        coroutine.send(None)

    assert excinfo.value.url == url
    assert excinfo.value.return_code == -2
    assert excinfo.value.http_status_code == None
    assert excinfo.value.error == ""


def test_http_request_process_error_stderr():
    url = "http://example.com"
    coroutine = http_request_process(url, {}, 0)

    future = coroutine.send(None)
    assert isinstance(future, FutureProcess)
    future.set_result(("", 0, "", "err"))

    with pytest.raises(HttpError) as excinfo:
        coroutine.send(None)

    assert excinfo.value.url == url
    assert excinfo.value.return_code == 0
    assert excinfo.value.http_status_code == None
    assert excinfo.value.error == "err"


def test_http_request_error_http_status():
    url = "http://example.com"
    coroutine = http_request(url, {}, 0)

    future = coroutine.send(None)
    assert isinstance(future, FutureUrl)

    future.set_result(
        (url, {}, {"response_code": "400", "headers": "", "output": "response"})
    )

    with pytest.raises(HttpError) as excinfo:
        coroutine.send(None)

    assert excinfo.value.url == url
    assert excinfo.value.return_code == None
    assert excinfo.value.http_status_code == 400
    assert excinfo.value.error == "response"


def test_http_request_error_retry_success():
    url = "http://example.com"
    coroutine = http_request(url, {}, 0, max_retries=2)

    future_1 = coroutine.send(None)
    assert isinstance(future_1, FutureUrl)
    future_1.set_result((url, {}, {"error": "error"}))

    future_2 = coroutine.send(None)
    assert isinstance(future_2, FutureTimer)
    future_2.set_result((0,))

    future_3 = coroutine.send(None)
    assert isinstance(future_3, FutureUrl)

    future_3.set_result(
        (url, {}, {"response_code": "200", "headers": "", "output": "response"})
    )

    with pytest.raises(StopIteration) as excinfo:
        coroutine.send(None)
    assert excinfo.value.value == "response"


def test_http_request_error_retry_error():
    url = "http://example.com"
    coroutine = http_request(url, {}, 0, max_retries=2)

    future_1 = coroutine.send(None)
    assert isinstance(future_1, FutureUrl)
    future_1.set_result((url, {}, {"error": "error"}))

    future_2 = coroutine.send(None)
    assert isinstance(future_2, FutureTimer)
    future_2.set_result((0,))

    future_3 = coroutine.send(None)
    assert isinstance(future_3, FutureUrl)
    future_3.set_result((url, {}, {"error": "error"}))

    future_4 = coroutine.send(None)
    assert isinstance(future_4, FutureTimer)
    future_4.set_result((0,))

    future_5 = coroutine.send(None)
    assert isinstance(future_5, FutureUrl)
    future_5.set_result((url, {}, {"error": "error"}))

    with pytest.raises(HttpError) as excinfo:
        coroutine.send(None)

    assert excinfo.value.url == url
    assert excinfo.value.return_code == None
    assert excinfo.value.http_status_code == None
    assert excinfo.value.error == "error"


def test_http_request_url_multiple_headers():
    url = "http://example.com"
    coroutine = http_request_url(url, {}, 0)
    future = coroutine.send(None)
    assert isinstance(future, FutureUrl)

    headers = (
        dedent(
            f"""
            HTTP/1.1 200 Connection established

            HTTP/2 200
            content-type: application/json; charset=utf-8
            """
        )
        .strip()
        .replace("\n", "\r\n")
    )
    future.set_result(
        (url, {}, {"response_code": "200", "headers": headers, "output": "response"})
    )

    with pytest.raises(StopIteration) as excinfo:
        coroutine.send(future)
    assert excinfo.value.value == (
        200,
        "2 200\r\ncontent-type: application/json; charset=utf-8",
        "response",
    )


def test_http_request_process_multiple_headers():
    url = "http://example.com"
    coroutine = http_request_process(url, {}, 0)
    future = coroutine.send(None)
    assert isinstance(future, FutureProcess)

    body = (
        dedent(
            f"""
            HTTP/1.1 200 Connection established

            HTTP/2 200
            content-type: application/json; charset=utf-8

            response
            """
        )
        .strip()
        .replace("\n", "\r\n")
    )
    future.set_result(("", 0, body, ""))

    with pytest.raises(StopIteration) as excinfo:
        coroutine.send(future)
    assert excinfo.value.value == (
        200,
        "2 200\r\ncontent-type: application/json; charset=utf-8",
        "response",
    )


@patch.object(weechat, "hook_timer")
def test_http_request_ratelimit(mock_method: MagicMock):
    url = "http://example.com"
    coroutine = http_request(url, {}, 0)

    future_1 = coroutine.send(None)
    assert isinstance(future_1, FutureUrl)

    future_1.set_result(
        (
            url,
            {},
            {
                "response_code": "429",
                "headers": "HTTP/2 429\r\nRetry-After: 12",
                "output": "response",
            },
        )
    )

    future_2 = coroutine.send(None)
    assert isinstance(future_2, FutureTimer)
    future_2.set_result((0,))

    mock_method.assert_called_once_with(
        12000, 0, 1, get_callback_name(weechat_task_cb), future_2.id
    )

    future_3 = coroutine.send(None)
    assert isinstance(future_3, FutureUrl)

    future_3.set_result(
        (url, {}, {"response_code": "200", "headers": "", "output": "response"})
    )

    with pytest.raises(StopIteration) as excinfo:
        coroutine.send(None)
    assert excinfo.value.value == "response"