RequestsMock doesn't catch requests inside loop.run_in_executor(None, ...)

Describe the bug

RequestsMock doesn't catch requests that are sent inside the default asyncio executor. As a result, real HTTP requests are sent.

Additional context

Python 3.9 - 3.11

Version of `responses`

0.24.1

Steps to Reproduce

import asyncio
import responses
import requests

def send_request():
    # Here a real HTTP request will be sent
    response = requests.get('https://example.com/')
    print(response.content)

def main():
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
        rsps.get('https://example.com/', body='test')

        loop = asyncio.get_event_loop()
        loop.run_in_executor(None, send_request)
        # but   send_request()  works correct and DOES NOT send a real HTTP request

main()

Expected Result

test should be printed.

Actual Result

A real HTTP request is sent to https://example.com/, and a page body b'<!doctype html>\n<html>\n<head>\n <title>Example Domain</title>.... is printed.

@andrei-shabanski I think there is an issue with the code you provided. In this case none of the context managers will work. As you can see you use a coroutine without waiting for the completion. How that works:

You set up the context manager
you call for the executor
code exits the context manager
Code exits main function
function send_request works outside the context manager in another thread
since it is in another thread outside of the decorator, responses cannot intercept it

see illustrated example:

import asyncio
import responses
import requests

def send_request():
    # Here a real HTTP request will be sent
    response = requests.get('https://example.com/')
    print(response.content)

def main():
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
        rsps.get('https://example.com/', body='test')

        loop = asyncio.get_event_loop()
        loop.run_in_executor(None, send_request)
    print("exit CtxMgr")

main()
print("exit main")

will return you:

exit CtxMgr
exit main
b'<!doctype html>\n<html>\n<head>\n    <title>Example Domain</title>\n\n    <meta charset="utf-8" />\n    <meta http-equiv="Content-type" content="text/html; charset=utf-8" />\n    <meta name="viewport" content="width=device-width, initial-scale=1" />\n    <style type="text/css">\n    body {\n        background-color: #f0f0f2;\n        margin: 0;\n        padding: 0;\n        font-family: -apple-system, system-ui, BlinkMacSystemFont, "Segoe UI", "Open Sans", "Helvetica Neue", Helvetica, Arial, sans-serif;\n        \n    }\n    div {\n        width: 600px;\n        margin: 5em auto;\n        padding: 2em;\n        background-color: #fdfdff;\n        border-radius: 0.5em;\n        box-shadow: 2px 3px 7px 2px rgba(0,0,0,0.02);\n    }\n    a:link, a:visited {\n        color: #38488f;\n        text-decoration: none;\n    }\n    @media (max-width: 700px) {\n        div {\n            margin: 0 auto;\n            width: auto;\n        }\n    }\n    </style>    \n</head>\n\n<body>\n<div>\n    <h1>Example Domain</h1>\n    <p>This domain is for use in illustrative examples in documents. You may use this\n    domain in literature without prior coordination or asking for permission.</p>\n    <p><a href="https://www.iana.org/domains/example">More information...</a></p>\n</div>\n</body>\n</html>\n'

what you should do instead:

import asyncio
import responses
import requests

def send_request():
    response = requests.get('https://example.com/')
    print(response.content)

async def main():
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
        rsps.get('https://example.com/', body='test')

        loop = asyncio.get_running_loop()
        await loop.run_in_executor(None, send_request)

asyncio.run(main())

that is similar to the documented example: https://github.com/getsentry/responses?tab=readme-ov-file#coroutines-and-multithreading

also see https://docs.python.org/3/library/asyncio-runner.html#asyncio.run

getsentry / responses