test_responses.py

# coding: utf-8

import inspect
import os
import re
import warnings
from io import BufferedReader
from io import BytesIO
from unittest.mock import Mock
from unittest.mock import patch

import pytest
import requests
from requests.exceptions import ChunkedEncodingError
from requests.exceptions import ConnectionError
from requests.exceptions import HTTPError

import responses
from responses import BaseResponse
from responses import CallbackResponse
from responses import PassthroughResponse
from responses import Response
from responses import matchers


def assert_reset():
    assert len(responses._default_mock.registered()) == 0
    assert len(responses.calls) == 0


def assert_response(resp, body=None, content_type="text/plain"):
    assert resp.status_code == 200
    assert resp.reason == "OK"
    if content_type is not None:
        assert resp.headers["Content-Type"] == content_type
    else:
        assert "Content-Type" not in resp.headers
    assert resp.text == body


def assert_params(resp, expected):
    assert hasattr(resp, "request"), "Missing request"
    assert hasattr(
        resp.request, "params"
    ), "Missing params on request that responses should add"
    assert getattr(resp.request, "params") == expected, "Incorrect parameters"


def test_response():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com", body=b"test")
        resp = requests.get("http://example.com")
        assert_response(resp, "test")
        assert len(responses.calls) == 1
        assert responses.calls[0].request.url == "http://example.com/"
        assert responses.calls[0].response.content == b"test"

        resp = requests.get("http://example.com?foo=bar")
        assert_response(resp, "test")
        assert len(responses.calls) == 2
        assert responses.calls[1].request.url == "http://example.com/?foo=bar"
        assert responses.calls[1].response.content == b"test"

    run()
    assert_reset()


def test_response_encoded():
    @responses.activate
    def run():
        # Path contains urlencoded =/()[]
        url = "http://example.org/foo.bar%3D%2F%28%29%5B%5D"
        responses.add(responses.GET, url, body="it works", status=200)
        resp = requests.get(url)
        assert_response(resp, "it works")

    run()
    assert_reset()


def test_response_with_instance():
    @responses.activate
    def run():
        responses.add(
            responses.Response(method=responses.GET, url="http://example.com")
        )
        resp = requests.get("http://example.com")
        assert_response(resp, "")
        assert len(responses.calls) == 1
        assert responses.calls[0].request.url == "http://example.com/"

        resp = requests.get("http://example.com?foo=bar")
        assert_response(resp, "")
        assert len(responses.calls) == 2
        assert responses.calls[1].request.url == "http://example.com/?foo=bar"

    run()
    assert_reset()


@pytest.mark.parametrize(
    "original,replacement",
    [
        ("http://example.com/two", "http://example.com/two"),
        (
            Response(method=responses.GET, url="http://example.com/two"),
            Response(
                method=responses.GET, url="http://example.com/two", body="testtwo"
            ),
        ),
        (
            re.compile(r"http://example\.com/two"),
            re.compile(r"http://example\.com/two"),
        ),
    ],
)
def test_replace(original, replacement):
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com/one", body="test1")

        if isinstance(original, BaseResponse):
            responses.add(original)
        else:
            responses.add(responses.GET, original, body="test2")

        responses.add(responses.GET, "http://example.com/three", body="test3")
        responses.add(
            responses.GET, re.compile(r"http://example\.com/four"), body="test3"
        )

        if isinstance(replacement, BaseResponse):
            responses.replace(replacement)
        else:
            responses.replace(responses.GET, replacement, body="testtwo")

        resp = requests.get("http://example.com/two")
        assert_response(resp, "testtwo")

    run()
    assert_reset()


@pytest.mark.parametrize(
    "original,replacement",
    [
        ("http://example.com/one", re.compile(r"http://example\.com/one")),
        (re.compile(r"http://example\.com/one"), "http://example.com/one"),
    ],
)
def test_replace_error(original, replacement):
    @responses.activate
    def run():
        responses.add(responses.GET, original)
        with pytest.raises(ValueError) as excinfo:
            responses.replace(responses.GET, replacement)
        assert "Response is not registered for URL %s" % replacement in str(
            excinfo.value
        )

    run()
    assert_reset()


def test_replace_response_object_error():
    @responses.activate
    def run():
        responses.add(Response(method=responses.GET, url="http://example.com/one"))
        with pytest.raises(ValueError) as excinfo:
            responses.replace(
                Response(method=responses.GET, url="http://example.com/two")
            )
        assert "Response is not registered for URL http://example.com/two" in str(
            excinfo.value
        )

    run()
    assert_reset()


@pytest.mark.parametrize(
    "original,replacement",
    [
        ("http://example.com/two", "http://example.com/two"),
        (
            Response(method=responses.GET, url="http://example.com/two"),
            Response(
                method=responses.GET, url="http://example.com/two", body="testtwo"
            ),
        ),
        (
            re.compile(r"http://example\.com/two"),
            re.compile(r"http://example\.com/two"),
        ),
    ],
)
def test_upsert_replace(original, replacement):
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com/one", body="test1")

        if isinstance(original, BaseResponse):
            responses.add(original)
        else:
            responses.add(responses.GET, original, body="test2")

        if isinstance(replacement, BaseResponse):
            responses.upsert(replacement)
        else:
            responses.upsert(responses.GET, replacement, body="testtwo")

        resp = requests.get("http://example.com/two")
        assert_response(resp, "testtwo")

    run()
    assert_reset()


@pytest.mark.parametrize(
    "original,replacement",
    [
        ("http://example.com/two", "http://example.com/two"),
        (
            Response(method=responses.GET, url="http://example.com/two"),
            Response(
                method=responses.GET, url="http://example.com/two", body="testtwo"
            ),
        ),
        (
            re.compile(r"http://example\.com/two"),
            re.compile(r"http://example\.com/two"),
        ),
    ],
)
def test_upsert_add(original, replacement):
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com/one", body="test1")

        if isinstance(replacement, BaseResponse):
            responses.upsert(replacement)
        else:
            responses.upsert(responses.GET, replacement, body="testtwo")

        resp = requests.get("http://example.com/two")
        assert_response(resp, "testtwo")

    run()
    assert_reset()


def test_remove():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com/zero")
        responses.add(responses.GET, "http://example.com/one")
        responses.add(responses.GET, "http://example.com/two")
        responses.add(responses.GET, re.compile(r"http://example\.com/three"))
        responses.add(responses.GET, re.compile(r"http://example\.com/four"))
        re.purge()
        responses.remove(responses.GET, "http://example.com/two")
        responses.remove(Response(method=responses.GET, url="http://example.com/zero"))
        responses.remove(responses.GET, re.compile(r"http://example\.com/four"))

        with pytest.raises(ConnectionError):
            requests.get("http://example.com/zero")
        requests.get("http://example.com/one")
        with pytest.raises(ConnectionError):
            requests.get("http://example.com/two")
        requests.get("http://example.com/three")
        with pytest.raises(ConnectionError):
            requests.get("http://example.com/four")

    run()
    assert_reset()


@pytest.mark.parametrize(
    "args1,kwargs1,args2,kwargs2,expected",
    [
        ((responses.GET, "a"), {}, (responses.GET, "a"), {}, True),
        ((responses.GET, "a"), {}, (responses.GET, "b"), {}, False),
        ((responses.GET, "a"), {}, (responses.POST, "a"), {}, False),
        (
            (responses.GET, "a"),
            {"match_querystring": True},
            (responses.GET, "a"),
            {},
            True,
        ),
    ],
)
def test_response_equality(args1, kwargs1, args2, kwargs2, expected):
    o1 = BaseResponse(*args1, **kwargs1)
    o2 = BaseResponse(*args2, **kwargs2)
    assert (o1 == o2) is expected
    assert (o1 != o2) is not expected


def test_response_equality_different_objects():
    o1 = BaseResponse(method=responses.GET, url="a")
    o2 = "str"
    assert (o1 == o2) is False
    assert (o1 != o2) is True


def test_connection_error():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com")

        with pytest.raises(ConnectionError):
            requests.get("http://example.com/foo")

        assert len(responses.calls) == 1
        assert responses.calls[0].request.url == "http://example.com/foo"
        assert type(responses.calls[0].response) is ConnectionError
        assert responses.calls[0].response.request

    run()
    assert_reset()


def test_match_querystring():
    @responses.activate
    def run():
        url = "http://example.com?test=1&foo=bar"
        responses.add(responses.GET, url, match_querystring=True, body=b"test")
        resp = requests.get("http://example.com?test=1&foo=bar")
        assert_response(resp, "test")
        resp = requests.get("http://example.com?foo=bar&test=1")
        assert_response(resp, "test")
        resp = requests.get("http://example.com/?foo=bar&test=1")
        assert_response(resp, "test")

    run()
    assert_reset()


def test_match_querystring_empty():
    @responses.activate
    def run():
        responses.add(
            responses.GET, "http://example.com", body=b"test", match_querystring=True
        )
        resp = requests.get("http://example.com")
        assert_response(resp, "test")
        resp = requests.get("http://example.com/")
        assert_response(resp, "test")
        with pytest.raises(ConnectionError):
            requests.get("http://example.com?query=foo")

    run()
    assert_reset()


def test_match_querystring_error():
    @responses.activate
    def run():
        responses.add(
            responses.GET, "http://example.com/?test=1", match_querystring=True
        )

        with pytest.raises(ConnectionError):
            requests.get("http://example.com/foo/?test=2")

    run()
    assert_reset()


def test_match_querystring_regex():
    @responses.activate
    def run():
        """Note that `match_querystring` value shouldn't matter when passing a
        regular expression"""

        responses.add(
            responses.GET,
            re.compile(r"http://example\.com/foo/\?test=1"),
            body="test1",
            match_querystring=True,
        )

        resp = requests.get("http://example.com/foo/?test=1")
        assert_response(resp, "test1")

        responses.add(
            responses.GET,
            re.compile(r"http://example\.com/foo/\?test=2"),
            body="test2",
            match_querystring=False,
        )

        resp = requests.get("http://example.com/foo/?test=2")
        assert_response(resp, "test2")

    run()
    assert_reset()


def test_match_querystring_error_regex():
    @responses.activate
    def run():
        """Note that `match_querystring` value shouldn't matter when passing a
        regular expression"""

        responses.add(
            responses.GET,
            re.compile(r"http://example\.com/foo/\?test=1"),
            match_querystring=True,
        )

        with pytest.raises(ConnectionError):
            requests.get("http://example.com/foo/?test=3")

        responses.add(
            responses.GET,
            re.compile(r"http://example\.com/foo/\?test=2"),
            match_querystring=False,
        )

        with pytest.raises(ConnectionError):
            requests.get("http://example.com/foo/?test=4")

    run()
    assert_reset()


def test_match_querystring_auto_activates():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com?test=1", body=b"test")
        resp = requests.get("http://example.com?test=1")
        assert_response(resp, "test")
        with pytest.raises(ConnectionError):
            requests.get("http://example.com/?test=2")

    run()
    assert_reset()


def test_match_querystring_missing_key():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com?foo=1&bar=2", body=b"test")
        with pytest.raises(ConnectionError):
            requests.get("http://example.com/?foo=1&baz=2")

        with pytest.raises(ConnectionError):
            requests.get("http://example.com/?bar=2&fez=1")

    run()
    assert_reset()


def test_accept_string_body():
    @responses.activate
    def run():
        url = "http://example.com/"
        responses.add(responses.GET, url, body="test")
        resp = requests.get(url)
        assert_response(resp, "test")

    run()
    assert_reset()


def test_accept_json_body():
    @responses.activate
    def run():
        content_type = "application/json"

        url = "http://example.com/"
        responses.add(responses.GET, url, json={"message": "success"})
        resp = requests.get(url)
        assert_response(resp, '{"message": "success"}', content_type)

        url = "http://example.com/1/"
        responses.add(responses.GET, url, json=[])
        resp = requests.get(url)
        assert_response(resp, "[]", content_type)

    run()
    assert_reset()


def test_no_content_type():
    @responses.activate
    def run():
        url = "http://example.com/"
        responses.add(responses.GET, url, body="test", content_type=None)
        resp = requests.get(url)
        assert_response(resp, "test", content_type=None)

    run()
    assert_reset()


def test_arbitrary_status_code():
    @responses.activate
    def run():
        url = "http://example.com/"
        responses.add(responses.GET, url, body="test", status=419)
        resp = requests.get(url)
        assert resp.status_code == 419
        assert resp.reason is None

    run()
    assert_reset()


def test_throw_connection_error_explicit():
    @responses.activate
    def run():
        url = "http://example.com"
        exception = HTTPError("HTTP Error")
        responses.add(responses.GET, url, exception)

        with pytest.raises(HTTPError) as HE:
            requests.get(url)

        assert str(HE.value) == "HTTP Error"

    run()
    assert_reset()


def test_callback():
    body = b"test callback"
    status = 400
    reason = "Bad Request"
    headers = {
        "foo": "bar",
        "Content-Type": "application/json",
        "Content-Length": "13",
    }
    url = "http://example.com/"

    def request_callback(_request):
        return status, headers, body

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback)
        resp = requests.get(url)
        assert resp.text == "test callback"
        assert resp.status_code == status
        assert resp.reason == reason
        assert "bar" == resp.headers.get("foo")
        assert "application/json" == resp.headers.get("Content-Type")
        assert "13" == resp.headers.get("Content-Length")

    run()
    assert_reset()


def test_callback_deprecated_stream_argument():
    with pytest.deprecated_call():
        CallbackResponse(responses.GET, "url", lambda x: x, stream=False)


def test_callback_deprecated_match_querystring_argument():
    with pytest.deprecated_call():
        CallbackResponse(responses.GET, "url", lambda x: x, match_querystring=False)


def test_callback_match_querystring_default_false():
    """
    Test to ensure that by default 'match_querystring' in 'add_callback' is set to False
    and does not raise deprecation
    see: https://github.com/getsentry/responses/issues/464 and related PR
    """
    body = b"test callback"
    status = 200
    params = {"hello": "world", "I am": "a big test"}
    headers = {"foo": "bar"}
    url = "http://example.com/"

    def request_callback(_request):
        return status, headers, body

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback, content_type=None)
        resp = requests.get(url, params=params)
        assert resp.text == "test callback"
        assert resp.status_code == status
        assert "foo" in resp.headers

    with warnings.catch_warnings():
        warnings.simplefilter("error")
        run()

    assert_reset()


def test_callback_exception_result():
    result = Exception()
    url = "http://example.com/"

    def request_callback(request):
        return result

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback)

        with pytest.raises(Exception) as e:
            requests.get(url)

        assert e.value is result

    run()
    assert_reset()


def test_callback_exception_body():
    body = Exception()
    url = "http://example.com/"

    def request_callback(request):
        return 200, {}, body

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback)

        with pytest.raises(Exception) as e:
            requests.get(url)

        assert e.value is body

    run()
    assert_reset()


def test_callback_no_content_type():
    body = b"test callback"
    status = 400
    reason = "Bad Request"
    headers = {"foo": "bar"}
    url = "http://example.com/"

    def request_callback(_request):
        return status, headers, body

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback, content_type=None)
        resp = requests.get(url)
        assert resp.text == "test callback"
        assert resp.status_code == status
        assert resp.reason == reason
        assert "foo" in resp.headers
        assert "Content-Type" not in resp.headers

    run()
    assert_reset()


def test_callback_content_type_dict():
    def request_callback(request):
        return (
            200,
            {"Content-Type": "application/json"},
            b"foo",
        )

    @responses.activate
    def run():
        responses.add_callback("GET", "http://mockhost/.foo", callback=request_callback)
        resp = requests.get("http://mockhost/.foo")
        assert resp.text == "foo"
        assert resp.headers["content-type"] == "application/json"

    run()
    assert_reset()


def test_callback_matchers():
    def request_callback(request):
        return (
            200,
            {"Content-Type": "application/json"},
            b"foo",
        )

    @responses.activate
    def run():
        req_data = {"some": "other", "data": "fields"}
        req_files = {"file_name": b"Old World!"}

        responses.add_callback(
            responses.POST,
            url="http://httpbin.org/post",
            match=[matchers.multipart_matcher(req_files, data=req_data)],
            callback=request_callback,
        )
        resp = requests.post("http://httpbin.org/post", data=req_data, files=req_files)
        assert resp.text == "foo"
        assert resp.headers["content-type"] == "application/json"

    run()
    assert_reset()


def test_callback_matchers_fail():
    @responses.activate
    def run():
        req_data = {"some": "other", "data": "fields"}
        req_files = {"file_name": b"Old World!"}

        responses.add_callback(
            responses.POST,
            url="http://httpbin.org/post",
            match=[matchers.multipart_matcher(req_files, data=req_data)],
            callback=lambda x: (
                0,
                {"a": ""},
                "",
            ),
        )
        with pytest.raises(ConnectionError) as exc:
            requests.post(
                "http://httpbin.org/post",
                data={"some": "other", "data": "wrong"},
                files=req_files,
            )

        assert "multipart/form-data doesn't match." in str(exc.value)

    run()
    assert_reset()


def test_callback_content_type_tuple():
    def request_callback(request):
        return (
            200,
            [("Content-Type", "application/json")],
            b"foo",
        )

    @responses.activate
    def run():
        responses.add_callback("GET", "http://mockhost/.foo", callback=request_callback)
        resp = requests.get("http://mockhost/.foo")
        assert resp.text == "foo"
        assert resp.headers["content-type"] == "application/json"

    run()
    assert_reset()


def test_regular_expression_url():
    @responses.activate
    def run():
        url = re.compile(r"https?://(.*\.)?example.com")
        responses.add(responses.GET, url, body=b"test")

        resp = requests.get("http://example.com")
        assert_response(resp, "test")

        resp = requests.get("https://example.com")
        assert_response(resp, "test")

        resp = requests.get("https://uk.example.com")
        assert_response(resp, "test")

        with pytest.raises(ConnectionError):
            requests.get("https://uk.exaaample.com")

    run()
    assert_reset()


def test_base_response_get_response():
    resp = BaseResponse("GET", ".com")
    with pytest.raises(NotImplementedError):
        resp.get_response(requests.PreparedRequest())


def test_custom_adapter():
    @responses.activate
    def run():
        url = "http://example.com"
        responses.add(responses.GET, url, body=b"test")

        calls = [0]

        class DummyAdapter(requests.adapters.HTTPAdapter):
            def send(self, *a, **k):
                calls[0] += 1
                return super().send(*a, **k)

        # Test that the adapter is actually used
        session = requests.Session()
        session.mount("http://", DummyAdapter())

        resp = session.get(url, allow_redirects=False)
        assert calls[0] == 1

        # Test that the response is still correctly emulated
        session = requests.Session()
        session.mount("http://", DummyAdapter())

        resp = session.get(url)
        assert_response(resp, "test")

    run()


def test_responses_as_context_manager():
    def run():
        with responses.mock:
            responses.add(responses.GET, "http://example.com", body=b"test")
            resp = requests.get("http://example.com")
            assert_response(resp, "test")
            assert len(responses.calls) == 1
            assert responses.calls[0].request.url == "http://example.com/"
            assert responses.calls[0].response.content == b"test"

            resp = requests.get("http://example.com?foo=bar")
            assert_response(resp, "test")
            assert len(responses.calls) == 2
            assert responses.calls[1].request.url == "http://example.com/?foo=bar"
            assert responses.calls[1].response.content == b"test"

    run()
    assert_reset()


def test_activate_doesnt_change_signature():
    def test_function(a, b=None):
        return (a, b)

    decorated_test_function = responses.activate(test_function)
    assert inspect.signature(test_function) == inspect.signature(
        decorated_test_function
    )

    assert decorated_test_function(1, 2) == test_function(1, 2)
    assert decorated_test_function(3) == test_function(3)


@pytest.fixture
def my_fruit():
    return "apple"


@pytest.fixture
def fruit_basket(my_fruit):
    return ["banana", my_fruit]


@pytest.mark.usefixtures("my_fruit", "fruit_basket")
class TestFixtures(object):
    """
    Test that pytest fixtures work well with 'activate' decorator
    """

    def test_function(self, my_fruit, fruit_basket):
        assert my_fruit in fruit_basket
        assert my_fruit == "apple"

    test_function_decorated = responses.activate(test_function)


def test_activate_mock_interaction():
    @patch("sys.stdout")
    def test_function(mock_stdout):
        return mock_stdout

    decorated_test_function = responses.activate(test_function)
    assert inspect.signature(test_function) == inspect.signature(
        decorated_test_function
    )

    value = test_function()
    assert isinstance(value, Mock)

    value = decorated_test_function()
    assert isinstance(value, Mock)


def test_activate_doesnt_change_signature_with_return_type():
    def test_function(a, b=None):
        return a, b

    # Add type annotations as they are syntax errors in py2.
    # Use a class to test for import errors in evaled code.
    test_function.__annotations__["return"] = Mock
    test_function.__annotations__["a"] = Mock

    decorated_test_function = responses.activate(test_function)
    assert inspect.signature(test_function) == inspect.signature(
        decorated_test_function
    )

    assert decorated_test_function(1, 2) == test_function(1, 2)
    assert decorated_test_function(3) == test_function(3)


def test_activate_doesnt_change_signature_for_method():
    class TestCase(object):
        def test_function(self, a, b=None):
            return (self, a, b)

        decorated_test_function = responses.activate(test_function)

    test_case = TestCase()
    assert test_case.decorated_test_function(1, 2) == test_case.test_function(1, 2)
    assert test_case.decorated_test_function(3) == test_case.test_function(3)


def test_response_cookies():
    body = b"test callback"
    status = 200
    headers = {"set-cookie": "session_id=12345; a=b; c=d"}
    url = "http://example.com/"

    def request_callback(request):
        return (status, headers, body)

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback)
        resp = requests.get(url)
        assert resp.text == "test callback"
        assert resp.status_code == status
        assert "session_id" in resp.cookies
        assert resp.cookies["session_id"] == "12345"
        assert set(resp.cookies.keys()) == set(["session_id"])

    run()
    assert_reset()


def test_response_cookies_secure():
    body = b"test callback"
    status = 200
    headers = {"set-cookie": "session_id=12345; a=b; c=d; secure"}
    url = "http://example.com/"

    def request_callback(request):
        return (status, headers, body)

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback)
        resp = requests.get(url)
        assert resp.text == "test callback"
        assert resp.status_code == status
        assert "session_id" in resp.cookies
        assert resp.cookies["session_id"] == "12345"
        assert set(resp.cookies.keys()) == set(["session_id"])

    run()
    assert_reset()


def test_response_cookies_multiple():
    body = b"test callback"
    status = 200
    headers = [
        ("set-cookie", "1P_JAR=2019-12-31-23; path=/; domain=.example.com; HttpOnly"),
        ("set-cookie", "NID=some=value; path=/; domain=.example.com; secure"),
    ]
    url = "http://example.com/"

    def request_callback(request):
        return (status, headers, body)

    @responses.activate
    def run():
        responses.add_callback(responses.GET, url, request_callback)
        resp = requests.get(url)
        assert resp.text == "test callback"
        assert resp.status_code == status
        assert set(resp.cookies.keys()) == set(["1P_JAR", "NID"])
        assert resp.cookies["1P_JAR"] == "2019-12-31-23"
        assert resp.cookies["NID"] == "some=value"

    run()
    assert_reset()


@pytest.mark.parametrize("request_stream", (True, False, None))
@pytest.mark.parametrize("responses_stream", (True, False, None))
def test_response_cookies_session(request_stream, responses_stream):
    @responses.activate
    def run():
        url = "https://example.com/path"
        responses.add(
            responses.GET,
            url,
            headers=[
                ("Set-cookie", "mycookie=cookieval; path=/; secure"),
            ],
            body="ok",
            stream=responses_stream,
        )
        session = requests.session()
        resp = session.get(url, stream=request_stream)
        assert resp.text == "ok"
        assert resp.status_code == 200

        assert "mycookie" in resp.cookies
        assert resp.cookies["mycookie"] == "cookieval"
        assert set(resp.cookies.keys()) == set(["mycookie"])

        assert "mycookie" in session.cookies
        assert session.cookies["mycookie"] == "cookieval"
        assert set(session.cookies.keys()) == set(["mycookie"])

    run()
    assert_reset()


def test_response_callback():
    """adds a callback to decorate the response, then checks it"""

    def run():
        def response_callback(resp):
            resp._is_mocked = True
            return resp

        with responses.RequestsMock(response_callback=response_callback) as m:
            m.add(responses.GET, "http://example.com", body=b"test")
            resp = requests.get("http://example.com")
            assert resp.text == "test"
            assert hasattr(resp, "_is_mocked")
            assert getattr(resp, "_is_mocked") is True

    run()
    assert_reset()


def test_response_filebody():
    """ Adds the possibility to use actual (binary) files as responses """

    def run():
        current_file = os.path.abspath(__file__)
        with responses.RequestsMock() as m:
            with open(current_file, "r") as out:
                m.add(responses.GET, "http://example.com", body=out.read(), stream=True)
                resp = requests.get("http://example.com", stream=True)
            with open(current_file, "r") as out:
                assert resp.text == out.read()

    run()
    assert_reset()


def test_use_stream_twice_to_double_raw_io():
    @responses.activate
    def run():
        url = "http://example.com"
        responses.add(responses.GET, url, body=b"42", stream=True)
        resp = requests.get(url, stream=True)
        assert resp.raw.read() == b"42"

    run()
    assert_reset()


def test_assert_all_requests_are_fired():
    def request_callback(request):
        raise BaseException()

    def run():
        with pytest.raises(AssertionError) as excinfo:
            with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
                m.add(responses.GET, "http://example.com", body=b"test")
        assert "http://example.com" in str(excinfo.value)
        assert responses.GET in str(excinfo.value)

        # check that assert_all_requests_are_fired default to True
        with pytest.raises(AssertionError):
            with responses.RequestsMock() as m:
                m.add(responses.GET, "http://example.com", body=b"test")

        # check that assert_all_requests_are_fired doesn't swallow exceptions
        with pytest.raises(ValueError):
            with responses.RequestsMock() as m:
                m.add(responses.GET, "http://example.com", body=b"test")
                raise ValueError()

        # check that assert_all_requests_are_fired=True doesn't remove urls
        with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
            m.add(responses.GET, "http://example.com", body=b"test")
            assert len(m.registered()) == 1
            requests.get("http://example.com")
            assert len(m.registered()) == 1

        # check that assert_all_requests_are_fired=True counts mocked errors
        with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
            m.add(responses.GET, "http://example.com", body=Exception())
            assert len(m.registered()) == 1
            with pytest.raises(Exception):
                requests.get("http://example.com")
            assert len(m.registered()) == 1

        with responses.RequestsMock(assert_all_requests_are_fired=True) as m:
            m.add_callback(responses.GET, "http://example.com", request_callback)
            assert len(m.registered()) == 1
            with pytest.raises(BaseException):
                requests.get("http://example.com")
            assert len(m.registered()) == 1

    run()
    assert_reset()


def test_allow_redirects_samehost():
    redirecting_url = "http://example.com"
    final_url_path = "/1"
    final_url = "{0}{1}".format(redirecting_url, final_url_path)
    url_re = re.compile(r"^http://example.com(/)?(\d+)?$")

    def request_callback(request):
        # endpoint of chained redirect
        if request.url.endswith(final_url_path):
            return 200, (), b"test"

        # otherwise redirect to an integer path
        else:
            if request.url.endswith("/0"):
                n = 1
            else:
                n = 0
            redirect_headers = {"location": "/{0!s}".format(n)}
            return 301, redirect_headers, None

    def run():
        # setup redirect
        with responses.mock:
            responses.add_callback(responses.GET, url_re, request_callback)
            resp_no_redirects = requests.get(redirecting_url, allow_redirects=False)
            assert resp_no_redirects.status_code == 301
            assert len(responses.calls) == 1  # 1x300
            assert responses.calls[0][1].status_code == 301
        assert_reset()

        with responses.mock:
            responses.add_callback(responses.GET, url_re, request_callback)
            resp_yes_redirects = requests.get(redirecting_url, allow_redirects=True)
            assert len(responses.calls) == 3  # 2x300 + 1x200
            assert len(resp_yes_redirects.history) == 2
            assert resp_yes_redirects.status_code == 200
            assert final_url == resp_yes_redirects.url
            status_codes = [call[1].status_code for call in responses.calls]
            assert status_codes == [301, 301, 200]
        assert_reset()

    run()
    assert_reset()


def test_handles_unicode_querystring():
    url = "http://example.com/test?type=2&ie=utf8&query=汉字"

    @responses.activate
    def run():
        responses.add(responses.GET, url, body="test", match_querystring=True)

        resp = requests.get(url)

        assert_response(resp, "test")

    run()
    assert_reset()


def test_handles_unicode_url():
    url = "http://www.संजाल.भारत/hi/वेबसाइट-डिजाइन"

    @responses.activate
    def run():
        responses.add(responses.GET, url, body="test")

        resp = requests.get(url)

        assert_response(resp, "test")

    run()
    assert_reset()


def test_handles_unicode_body():
    url = "http://example.com/test"

    @responses.activate
    def run():
        responses.add(responses.GET, url, body="михољско лето")

        resp = requests.get(url)

        assert_response(resp, "михољско лето", content_type="text/plain; charset=utf-8")

    run()
    assert_reset()


def test_handles_buffered_reader_body():
    url = "http://example.com/test"

    @responses.activate
    def run():
        responses.add(responses.GET, url, body=BufferedReader(BytesIO(b"test")))  # type: ignore

        resp = requests.get(url)

        assert_response(resp, "test")

    run()
    assert_reset()


def test_headers():
    @responses.activate
    def run():
        responses.add(
            responses.GET, "http://example.com", body="", headers={"X-Test": "foo"}
        )
        resp = requests.get("http://example.com")
        assert resp.headers["X-Test"] == "foo"

    run()
    assert_reset()


def test_content_length_error(monkeypatch):
    """
    Currently 'requests' does not enforce content length validation,
    (validation that body length matches header). However, this could
    be expected in next major version, see
    https://github.com/psf/requests/pull/3563

    Now user can manually patch URL3 lib to achieve the same

    See discussion in
    https://github.com/getsentry/responses/issues/394
    """

    @responses.activate
    def run():
        responses.add(
            responses.GET,
            "http://example.com/api/123",
            json={"message": "this body is too large"},
            adding_headers={"content-length": "2"},
        )
        with pytest.raises(ChunkedEncodingError) as exc:
            requests.get("http://example.com/api/123")

        assert "IncompleteRead" in str(exc.value)

    # Type errors here and on 1250 are ignored because the stubs for requests
    # are off https://github.com/python/typeshed/blob/f8501d33c737482a829c6db557a0be26895c5941
    #   /stubs/requests/requests/packages/__init__.pyi#L1
    original_init = getattr(requests.packages.urllib3.HTTPResponse, "__init__")  # type: ignore

    def patched_init(self, *args, **kwargs):
        kwargs["enforce_content_length"] = True
        original_init(self, *args, **kwargs)

    monkeypatch.setattr(
        requests.packages.urllib3.HTTPResponse, "__init__", patched_init  # type: ignore
    )

    run()
    assert_reset()


def test_stream_with_none_chunk_size():
    """
    See discussion in
    https://github.com/getsentry/responses/issues/438
    """

    @responses.activate
    def run():
        responses.add(
            responses.GET,
            "https://example.com",
            status=200,
            content_type="application/octet-stream",
            body=b"This is test",
            auto_calculate_content_length=True,
        )
        res = requests.get("https://example.com", stream=True)
        for chunk in res.iter_content(chunk_size=None):
            assert chunk == b"This is test"

    run()
    assert_reset()


def test_legacy_adding_headers():
    @responses.activate
    def run():
        responses.add(
            responses.GET,
            "http://example.com",
            body="",
            adding_headers={"X-Test": "foo"},
        )
        resp = requests.get("http://example.com")
        assert resp.headers["X-Test"] == "foo"

    run()
    assert_reset()


def test_legacy_adding_headers_with_content_type():
    @responses.activate
    def run():
        with pytest.raises(RuntimeError) as excinfo:
            responses.add(
                responses.GET,
                "http://example.com",
                body="test",
                content_type="text/html",
                adding_headers={"Content-Type": "text/html; charset=utf-8"},
            )
        assert (
            "You cannot define both `content_type` and `headers[Content-Type]`"
            in str(excinfo.value)
        )

    run()
    assert_reset()


def test_auto_calculate_content_length_string_body():
    @responses.activate
    def run():
        url = "http://example.com/"
        responses.add(
            responses.GET, url, body="test", auto_calculate_content_length=True
        )
        resp = requests.get(url)
        assert_response(resp, "test")
        assert resp.headers["Content-Length"] == "4"

    run()
    assert_reset()


def test_auto_calculate_content_length_bytes_body():
    @responses.activate
    def run():
        url = "http://example.com/"
        responses.add(
            responses.GET, url, body=b"test bytes", auto_calculate_content_length=True
        )
        resp = requests.get(url)
        assert_response(resp, "test bytes")
        assert resp.headers["Content-Length"] == "10"

    run()
    assert_reset()


def test_auto_calculate_content_length_json_body():
    @responses.activate
    def run():
        content_type = "application/json"

        url = "http://example.com/"
        responses.add(
            responses.GET,
            url,
            json={"message": "success"},
            auto_calculate_content_length=True,
        )
        resp = requests.get(url)
        assert_response(resp, '{"message": "success"}', content_type)
        assert resp.headers["Content-Length"] == "22"

        url = "http://example.com/1/"
        responses.add(responses.GET, url, json=[], auto_calculate_content_length=True)
        resp = requests.get(url)
        assert_response(resp, "[]", content_type)
        assert resp.headers["Content-Length"] == "2"

    run()
    assert_reset()


def test_auto_calculate_content_length_unicode_body():
    @responses.activate
    def run():
        url = "http://example.com/test"
        responses.add(
            responses.GET, url, body="михољско лето", auto_calculate_content_length=True
        )
        resp = requests.get(url)
        assert_response(resp, "михољско лето", content_type="text/plain; charset=utf-8")
        assert resp.headers["Content-Length"] == "25"

    run()
    assert_reset()


def test_auto_calculate_content_length_doesnt_work_for_buffered_reader_body():
    @responses.activate
    def run():
        url = "http://example.com/test"
        responses.add(
            responses.GET,
            url,
            body=BufferedReader(BytesIO(b"testing")),  # type: ignore
            auto_calculate_content_length=True,
        )
        resp = requests.get(url)
        assert_response(resp, "testing")
        assert "Content-Length" not in resp.headers

    run()
    assert_reset()


def test_auto_calculate_content_length_doesnt_override_existing_value():
    @responses.activate
    def run():
        url = "http://example.com/"
        responses.add(
            responses.GET,
            url,
            body="test",
            headers={"Content-Length": "2"},
            auto_calculate_content_length=True,
        )
        resp = requests.get(url)
        assert_response(resp, "test")
        assert resp.headers["Content-Length"] == "2"

    run()
    assert_reset()


def test_multiple_responses():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com", body="test")
        responses.add(responses.GET, "http://example.com", body="rest")
        responses.add(responses.GET, "http://example.com", body="fest")
        responses.add(responses.GET, "http://example.com", body="best")

        resp = requests.get("http://example.com")
        assert_response(resp, "test")

        resp = requests.get("http://example.com")
        assert_response(resp, "rest")

        resp = requests.get("http://example.com")
        assert_response(resp, "fest")

        resp = requests.get("http://example.com")
        assert_response(resp, "best")

        # After all responses are used, last response should be repeated
        resp = requests.get("http://example.com")
        assert_response(resp, "best")

    run()
    assert_reset()


def test_multiple_responses_intermixed():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com", body="test")
        resp = requests.get("http://example.com")
        assert_response(resp, "test")

        responses.add(responses.GET, "http://example.com", body="rest")
        resp = requests.get("http://example.com")
        assert_response(resp, "rest")

        responses.add(responses.GET, "http://example.com", body="best")
        resp = requests.get("http://example.com")
        assert_response(resp, "best")

        # After all responses are used, last response should be repeated
        resp = requests.get("http://example.com")
        assert_response(resp, "best")

    run()
    assert_reset()


def test_multiple_urls():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com/one", body="one")
        responses.add(responses.GET, "http://example.com/two", body="two")

        resp = requests.get("http://example.com/two")
        assert_response(resp, "two")
        resp = requests.get("http://example.com/one")
        assert_response(resp, "one")

    run()
    assert_reset()


def test_multiple_methods():
    @responses.activate
    def run():
        responses.add(responses.GET, "http://example.com/one", body="gotcha")
        responses.add(responses.POST, "http://example.com/one", body="posted")

        resp = requests.get("http://example.com/one")
        assert_response(resp, "gotcha")
        resp = requests.post("http://example.com/one")
        assert_response(resp, "posted")

    run()
    assert_reset()


def test_passthrough_flag(httpserver):
    httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
    response = Response(responses.GET, httpserver.url, body="MOCK")

    @responses.activate
    def run_passthrough():
        responses.add(response)
        resp = requests.get(httpserver.url)
        assert_response(resp, "OK")

    @responses.activate
    def run_mocked():
        responses.add(response)
        resp = requests.get(httpserver.url)
        assert_response(resp, "MOCK")

    run_mocked()
    assert_reset()

    response.passthrough = True
    run_passthrough()
    assert_reset()


def test_passthrough_response(httpserver):
    httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})

    @responses.activate
    def run():
        responses.add(PassthroughResponse(responses.GET, httpserver.url))
        responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
        responses.add(responses.GET, "http://example.com/two", body="two")

        resp = requests.get("http://example.com/two")
        assert_response(resp, "two")
        resp = requests.get("{}/one".format(httpserver.url))
        assert_response(resp, "one")
        resp = requests.get(httpserver.url)
        assert_response(resp, "OK")

        assert len(responses.calls) == 3
        responses.assert_call_count(httpserver.url, 1)

    run()
    assert_reset()


def test_passthrough_response_stream(httpserver):
    httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})

    @responses.activate
    def run():
        responses.add(PassthroughResponse(responses.GET, httpserver.url))
        content_1 = requests.get(httpserver.url).content
        with requests.get(httpserver.url, stream=True) as resp:
            content_2 = resp.raw.read()
        assert content_1 == content_2

    run()
    assert_reset()


def test_passthru_prefixes(httpserver):
    httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})

    @responses.activate
    def run_constructor_argument():
        with responses.RequestsMock(passthru_prefixes=(httpserver.url,)):
            resp = requests.get(httpserver.url)
            assert_response(resp, "OK")

    @responses.activate
    def run_property_setter():
        with responses.RequestsMock() as m:
            m.passthru_prefixes = tuple([httpserver.url])
            resp = requests.get(httpserver.url)
            assert_response(resp, "OK")

    run_constructor_argument()
    assert_reset()
    run_property_setter()
    assert_reset()


def test_passthru(httpserver):
    httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})

    @responses.activate
    def run():
        responses.add_passthru(httpserver.url)
        responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
        responses.add(responses.GET, "http://example.com/two", body="two")

        resp = requests.get("http://example.com/two")
        assert_response(resp, "two")
        resp = requests.get("{}/one".format(httpserver.url))
        assert_response(resp, "one")
        resp = requests.get(httpserver.url)
        assert_response(resp, "OK")

    run()
    assert_reset()


def test_passthru_regex(httpserver):
    httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})

    @responses.activate
    def run():
        responses.add_passthru(re.compile("{}/\\w+".format(httpserver.url)))
        responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
        responses.add(responses.GET, "http://example.com/two", body="two")

        resp = requests.get("http://example.com/two")
        assert_response(resp, "two")
        resp = requests.get("{}/one".format(httpserver.url))
        assert_response(resp, "one")
        resp = requests.get("{}/two".format(httpserver.url))
        assert_response(resp, "OK")
        resp = requests.get("{}/three".format(httpserver.url))
        assert_response(resp, "OK")

    run()
    assert_reset()


def test_passthru_does_not_persist_across_tests(httpserver):
    """
    passthru should be erased on exit from context manager
    see:
    https://github.com/getsentry/responses/issues/322
    """
    httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})

    @responses.activate
    def with_a_passthru():
        assert not responses._default_mock.passthru_prefixes
        responses.add_passthru(re.compile(".*"))
        try:
            response = requests.get("https://example.com")
        except ConnectionError as err:  # pragma: no cover
            if "Failed to establish" in str(err):  # pragma: no cover
                pytest.skip("Cannot resolve DNS for example.com")  # pragma: no cover
            raise err  # pragma: no cover

        assert response.status_code == 200

    @responses.activate
    def without_a_passthru():
        assert not responses._default_mock.passthru_prefixes
        with pytest.raises(requests.exceptions.ConnectionError):
            requests.get("https://example.com")

    with_a_passthru()
    without_a_passthru()


def test_method_named_param():
    @responses.activate
    def run():
        responses.add(method=responses.GET, url="http://example.com", body="OK")
        resp = requests.get("http://example.com")
        assert_response(resp, "OK")

    run()
    assert_reset()


def test_passthru_unicode():
    @responses.activate
    def run():
        with responses.RequestsMock() as m:
            url = "http://موقع.وزارة-الاتصالات.مصر/"
            clean_url = "http://xn--4gbrim.xn----ymcbaaajlc6dj7bxne2c.xn--wgbh1c/"
            m.add_passthru(url)
            assert m.passthru_prefixes[0] == clean_url

    run()
    assert_reset()


def test_custom_target(monkeypatch):
    requests_mock = responses.RequestsMock(target="something.else")
    std_mock_mock = responses.std_mock.MagicMock()
    patch_mock = std_mock_mock.patch
    monkeypatch.setattr(responses, "std_mock", std_mock_mock)
    requests_mock.start()
    assert len(patch_mock.call_args_list) == 1
    assert patch_mock.call_args[1]["target"] == "something.else"


def test_cookies_from_headers():
    text = "こんにちは/世界"
    quoted_text = responses.quote(text)
    expected = {"x": "a", "y": quoted_text}
    headers = {"set-cookie": "; ".join(k + "=" + v for k, v in expected.items())}
    cookiejar = responses._cookies_from_headers(headers)
    for k, v in cookiejar.items():
        assert isinstance(v, str)
        assert v == expected[k]


@pytest.mark.parametrize(
    "url",
    (
        "http://example.com",
        "http://example.com/some/path",
        "http://example.com/other/path/",
    ),
)
def test_request_param(url):
    @responses.activate
    def run():
        params = {"hello": "world", "example": "params"}
        responses.add(
            method=responses.GET,
            url="{0}?hello=world".format(url),
            body="test",
            match_querystring=False,
        )
        resp = requests.get(url, params=params)
        assert_response(resp, "test")
        assert_params(resp, params)

        resp = requests.get(url)
        assert_response(resp, "test")
        assert_params(resp, {})

    run()
    assert_reset()


def test_request_param_with_multiple_values_for_the_same_key():
    @responses.activate
    def run():
        url = "http://example.com"
        params = {"key1": ["one", "two"], "key2": "three"}
        responses.add(
            method=responses.GET,
            url=url,
            body="test",
        )
        resp = requests.get(url, params=params)
        assert_response(resp, "test")
        assert_params(resp, params)

    run()
    assert_reset()


@pytest.mark.parametrize(
    "url", ("http://example.com", "http://example.com?hello=world")
)
def test_assert_call_count(url):
    @responses.activate
    def run():
        responses.add(responses.GET, url)
        responses.add(responses.GET, "http://example1.com")

        assert responses.assert_call_count(url, 0) is True

        with pytest.raises(AssertionError) as excinfo:
            responses.assert_call_count(url, 2)
        assert "Expected URL '{0}' to be called 2 times. Called 0 times.".format(
            url
        ) in str(excinfo.value)

        requests.get(url)
        assert responses.assert_call_count(url, 1) is True

        requests.get("http://example1.com")
        assert responses.assert_call_count(url, 1) is True

        requests.get(url)
        with pytest.raises(AssertionError) as excinfo:
            responses.assert_call_count(url, 3)
        assert "Expected URL '{0}' to be called 3 times. Called 2 times.".format(
            url
        ) in str(excinfo.value)

    run()
    assert_reset()


def test_fail_request_error():
    """
    Validate that exception is raised if request URL/Method/kwargs don't match
    :return:
    """

    def run():
        with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
            rsps.add("POST", "http://example1.com")
            rsps.add("GET", "http://example.com")

            with pytest.raises(ConnectionError) as excinfo:
                requests.post("http://example.com", data={"id": "bad"})

            msg = str(excinfo.value)
            assert "- POST http://example1.com/ URL does not match" in msg
            assert "- GET http://example.com/ Method does not match" in msg

    run()
    assert_reset()


@pytest.mark.parametrize(
    "response_params, expected_representation",
    [
        (
            {"method": responses.GET, "url": "http://example.com/"},
            (
                "<Response(url='http://example.com/' status=200 "
                "content_type='text/plain' headers='null')>"
            ),
        ),
        (
            {
                "method": responses.POST,
                "url": "http://another-domain.com/",
                "content_type": "application/json",
                "status": 404,
            },
            (
                "<Response(url='http://another-domain.com/' status=404 "
                "content_type='application/json' headers='null')>"
            ),
        ),
        (
            {
                "method": responses.PUT,
                "url": "http://abcd.com/",
                "content_type": "text/html",
                "status": 500,
                "headers": {"X-Test": "foo"},
                "body": {"it_wont_be": "considered"},
            },
            (
                "<Response(url='http://abcd.com/' status=500 "
                "content_type='text/html' headers='{\"X-Test\": \"foo\"}')>"
            ),
        ),
    ],
)
def test_response_representations(response_params, expected_representation):
    response = Response(**response_params)

    assert str(response) == expected_representation
    assert repr(response) == expected_representation


def test_mocked_responses_list_registered():
    @responses.activate
    def run():
        first_response = Response(
            responses.GET,
            "http://example.com/",
            body="",
            headers={"X-Test": "foo"},
            status=404,
        )
        second_response = Response(
            responses.GET, "http://example.com/", body="", headers={"X-Test": "foo"}
        )
        third_response = Response(
            responses.POST,
            "http://anotherdomain.com/",
        )
        responses.add(first_response)
        responses.add(second_response)
        responses.add(third_response)

        mocks_list = responses.registered()

        assert mocks_list == responses.mock.registered()
        assert mocks_list == [first_response, second_response, third_response]

    run()
    assert_reset()


@pytest.mark.parametrize(
    "url,other_url",
    [
        ("http://service-A/foo?q=fizz", "http://service-a/foo?q=fizz"),
        ("http://service-a/foo", "http://service-A/foo"),
        ("http://someHost-AwAy/", "http://somehost-away/"),
        ("http://fizzbuzz/foo", "http://fizzbuzz/foo"),
    ],
)
def test_rfc_compliance(url, other_url):
    @responses.activate
    def run():
        responses.add(method=responses.GET, url=url)
        resp = requests.request("GET", other_url)
        assert_response(resp, "")

    run()
    assert_reset()


def test_requests_between_add():
    @responses.activate
    def run():
        responses.add(responses.GET, "https://example.com/", json={"response": "old"})
        assert requests.get("https://example.com/").content == b'{"response": "old"}'
        assert requests.get("https://example.com/").content == b'{"response": "old"}'
        assert requests.get("https://example.com/").content == b'{"response": "old"}'

        responses.add(responses.GET, "https://example.com/", json={"response": "new"})

        assert requests.get("https://example.com/").content == b'{"response": "new"}'
        assert requests.get("https://example.com/").content == b'{"response": "new"}'
        assert requests.get("https://example.com/").content == b'{"response": "new"}'

    run()
    assert_reset()


def test_responses_reuse():
    @responses.activate
    def run():
        url = "https://someapi.com/"
        fail_response = responses.Response(
            method="GET", url=url, body="fail", status=500
        )
        responses.add(responses.GET, url, "success", status=200)
        responses.add(fail_response)
        responses.add(fail_response)
        responses.add(fail_response)
        responses.add(responses.GET, url, "success", status=200)
        responses.add(responses.GET, url, "", status=302)

        response = requests.get(url)
        assert response.content == b"success"

        for _ in range(3):
            response = requests.get(url)
            assert response.content == b"fail"

    run()
    assert_reset()


async def test_async_calls():
    @responses.activate
    async def run():
        responses.add(
            responses.GET,
            "http://twitter.com/api/1/foobar",
            json={"error": "not found"},
            status=404,
        )

        resp = requests.get("http://twitter.com/api/1/foobar")
        assert resp.json() == {"error": "not found"}
        assert responses.calls[0].request.url == "http://twitter.com/api/1/foobar"

    await run()
    assert_reset()