Ask for forcing the retry policy to be 0

[vilit1]

Describe the bug

I am currently using the requests mock as so:

@pytest.fixture
def mocked_response():
    with responses.RequestMock() as rsps:
        yeild rsps

With the update from 0.20.0 to 0.21.0 and 0.22.0, the retry mechanism (for any calls that result in 500) forces our tests to fail. We want to force the retry mechanism to not activate and one way I have come up with is to overwrite the _on_request with a partial function as so:

@pytest.fixture
def mocked_response():
    with responses.RequestMock() as rsps:
        on_request_with_no_retry = partial(rsps._on_request, retries=Retry(0, read=False))
        rsps._on_request = on_request_with_no_retry
        yeild rsps

Is there a cleaner way to do this? I would prefer having the responses.RequestMock() initializer take in either a Retry object or an HttpAdapter (that I can attach the Retry object to).

Additional context

No response

Version of responses

0.22.0

Steps to Reproduce

@pytest.fixture
def mocked_response():
    with responses.RequestMock() as rsps:
        on_request_with_no_retry = partial(rsps._on_request, retries=Retry(0, read=False))
        rsps._on_request = on_request_with_no_retry
        yeild rsps

Expected Result

Would like to have the constructor for responses.RequestMock() take in a Retry or HttpAdapter so that I do not have to overwrite an internal method.

Actual Result

Have to overwrite the internal method or restructure all test cases.

[beliaev-maksim]

@vilit1 please read the description of the field for "Steps to Reproduce"

Provide a minimal reproducible self-contained code snippet. Snippet must be as small as possible and ready to run.

Please, respect time of project maintainers, I am willing to help but I do not want to guess what is happening, it should be crystal clear

[kerenkhatiwada]

Moving status based on the above. Please change if needed.

[vilit1]

I wanted to know if it is possible to modify the constructor for responses.RequestMock() so I would not have to overwrite an internal method.

Here is a small runnable snippet:

import pytest
import responses
import requests
import re
import json
from functools import partial
from urllib3.util.retry import Retry

@pytest.fixture
def mocked_response_no_retry():
    with responses.RequestsMock() as rsps:
        on_request_with_no_retry = partial(rsps._on_request, retries=Retry(0, read=False))
        rsps._on_request = on_request_with_no_retry
        yield rsps

@pytest.mark.parametrize("error_code", [500])
def test_device_show_error(mocked_response_no_retry, error_code):
    device_id = "potato"
    mocked_response_no_retry.add(
        method=responses.GET,
        url=re.compile(
            "https://example.com"
        ),
        body=json.dumps({"device": device_id}),
        status=error_code,
        content_type="application/json",
        match_querystring=False,
    )

    requests.get("https://example.com")
    assert len(mocked_response_no_retry.calls) == 1

You can see here that I am changing an internal method (_on_request) to a partial method that takes in a Retry object that I have defined.

If you want an example of a failure - I can provide a longer snippet that involves the CLI that I am working with. It will require installing some packages though.

[beliaev-maksim]

what do you expect to receive in real request if you set Retry on the adapter to 0 and getting 500 ? why do you think that the current implementation does not follow the real requests process?

https://github.com/getsentry/responses/blob/ec6c75544b3be7f800223964ccd6b7dedac1a42d/responses/tests/test_responses.py#L2414

[vilit1]

In a real request, the call will retry 4 times before ending. However, for testing purposes, I want to cut this short.

If the Retry on the adapter is 0 and the 500 is called the client will receive that 500 and error properly. If the Retry on the adapter is 4 and the 500 is called, the client will retry 4 more times before giving up.

Here is the longer snippet (you will require to install the azure-cli and the azure-iot extension to run this):

import pytest
import responses
import requests
import json
import re
from functools import partial
from urllib3.util.retry import Retry

from azure.cli.core.commands import AzCliCommand
from azure.cli.core.mock import DummyCli
from azext_iot.operations import hub as subject

# Mock Iot Hub Target
mock_target = {}
mock_target["entity"] = "myhub.azure-devices.net"
mock_target["name"] = "myhub"
mock_target["primarykey"] = "key1"
mock_target["secondarykey"] = "key2"
mock_target["policy"] = "policy"
mock_target["subscription"] = "sub"
mock_target["location"] = "westus2"
mock_target["sku_tier"] = "Standard"
mock_target["resourcegroup"] = "myresourcegroup"

path_iot_hub_service_factory = "azext_iot._factory.iot_hub_service_factory"
path_ghcs = "azext_iot.iothub.providers.discovery.IotHubDiscovery.get_target"
path_discovery_init = (
    "azext_iot.iothub.providers.discovery.IotHubDiscovery._initialize_client"
)

@pytest.fixture()
def fixture_ghcs(mocker):
    ghcs = mocker.patch(path_ghcs)
    ghcs.return_value = mock_target
    mocker.patch(path_iot_hub_service_factory)
    mocker.patch(path_discovery_init)

    return ghcs

@pytest.fixture()
def fixture_cmd(mocker):
    cli = DummyCli()
    cli.loader = mocker.MagicMock()
    cli.loader.cli_ctx = cli

    def test_handler1():
        pass

    return AzCliCommand(cli.loader, "iot-extension command", test_handler1)

@pytest.fixture
def mocked_response():
    with responses.RequestsMock() as rsps:
        on_request_with_no_retry = partial(rsps._on_request, retries=Retry(0, read=False))
        rsps._on_request = on_request_with_no_retry
        yield rsps

@pytest.mark.parametrize("error_code", [500])
def test_device_show_error(fixture_cmd, mocked_response, fixture_ghcs, error_code):
    device_id = "potato"
    mocked_response.add(
        method=responses.GET,
            url=re.compile(
                "https://{}/devices/{}".format(mock_target["entity"], device_id)
            ),
        body=json.dumps({"device": device_id}),
        status=error_code,
        content_type="application/json",
        match_querystring=False,
    )

    with pytest.raises(Exception) as e:
        subject.iot_device_show(
            cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"]
        )

    assert len(mocked_response.calls) == 1

@pytest.fixture
def mocked_response2():
    with responses.RequestsMock() as rsps:
        yield rsps

@pytest.mark.parametrize("error_code", [500])
def test_device_show_error2(fixture_cmd, mocked_response2, fixture_ghcs, error_code):
    device_id = "potato"
    mocked_response2.add(
        method=responses.GET,
            url=re.compile(
                "https://{}/devices/{}".format(mock_target["entity"], device_id)
            ),
        body=json.dumps({"device": device_id}),
        status=error_code,
        content_type="application/json",
        match_querystring=False,
    )

    with pytest.raises(Exception) as e:
        subject.iot_device_show(
            cmd=fixture_cmd, device_id=device_id, hub_name=mock_target["entity"]
        )

    assert len(mocked_response2.calls) == 1

Note that the call for test_device_show_error2 will have a retry of 4 so without changing _on_request, the call will try to call for a total of 5 times before giving up as shown here:

[beliaev-maksim]

However, for testing purposes, I want to cut this short.

what is the reason? test duration will increase by 2ms ?

if your real request will try 4 times, then you need to simulate it 4 times

I am -1, this is a general purpose library that emulates real requests.

[vilit1]

Ok, thank you for your time!