How can I use mocket to test retry logic?

[LilyFoote]

I'm currently trying to convert from httpretty to mocket. Most of the conversions have been straightforward so far, but I have one unusual use case using a callback function, which I think mocket doesn't support.

    from threading import Event

    import httpretty

    with httpretty.enabled():
        with open("tests/api_responses/body.json") as _body:
            body = _body.read()
        with open("tests/api_responses/headers.json") as headers:
            headers = json.load(headers)

        event = Event()
        first = True

        def callback(request, url, headers):
            nonlocal first
            if first:
                first = False
                event.wait(1)
                return 200, headers, "{}"
            return 200, headers, body

        httpretty.register_uri(
            httpretty.GET, "https://api.github.com/users/lilyfoote", body=callback
        )

What I want here is the first response to be delayed (by a second) so the http library attempts a retry and successfully gets the second response immediately. If there a way to achieve this?

[mindflayer]

Hi @LilyFoote,
is the client retrying a second time due to a time-out error?

[mindflayer]

If that's the case, Mocket gives you a way for raising errors:
https://github.com/mindflayer/python-mocket?tab=readme-ov-file#example-of-how-to-fake-socket-errors

[LilyFoote]

Thanks! I'll give that a go!

[LilyFoote]

This isn't working for me. I tried:

    Entry.register(
        Entry.GET,
        "https://api.github.com/users/lilyfoote",
        TimeoutError(),
        Response(body=body, headers=headers, status=200),
    )

and

    Entry.register(
        Entry.GET,
        "https://api.github.com/users/lilyfoote",
        requests.ReadTimeout(),
        Response(body=body, headers=headers, status=200),
    )

But both lead to the exception being caught by requests and re-raised as a ConnectionError.

The code I'm trying to mock is:

def requests_retry_view(request: HttpRequest) -> HttpResponse:
    url = "https://api.github.com/users/lilyfoote"
    try:
        requests.get(url, timeout=0.2)
    except ReadTimeout:
        github_data = requests.get(url)
    return HttpResponse(github_data.content)

Ideally, I would get a ReadTimeout from the first requests call triggering the second call, which would then succeed.

[LilyFoote]

Aha! I was able to get this working by calling urllib3.exceptions.ReadTimeoutError(None, None, None). Thanks for the help!

[mindflayer]

The example shows:

Entry.single_register(Entry.GET, url, exception=socket.error())

You forgot to use exception= and that's why Mocket is not properly raising it. Always be sure you are raising the internal exception that would make the client - requests - fail the way you want.