From a23f22d365eb6b16a5adf7c1d9d0d28127e9cdaa Mon Sep 17 00:00:00 2001 From: Maksim Beliaev Date: Wed, 28 Jun 2023 17:01:06 +0200 Subject: [PATCH] move all passthru functions into one class (#646) --- responses/tests/test_responses.py | 334 +++++++++++++++--------------- 1 file changed, 169 insertions(+), 165 deletions(-) diff --git a/responses/tests/test_responses.py b/responses/tests/test_responses.py index 154e03a4..507838a9 100644 --- a/responses/tests/test_responses.py +++ b/responses/tests/test_responses.py @@ -1624,196 +1624,213 @@ def run(): assert_reset() -def test_passthrough_flag(httpserver): - httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain") - url = httpserver.url_for("/") - - response = Response(responses.GET, url, body="MOCK") - - @responses.activate - def run_passthrough(): - responses.add(response) - resp = requests.get(url) - assert_response(resp, "OK") +class TestPassthru: + def test_passthrough_flag(self, httpserver): + httpserver.expect_request("/").respond_with_data( + "OK", content_type="text/plain" + ) + url = httpserver.url_for("/") - @responses.activate - def run_mocked(): - responses.add(response) - resp = requests.get(url) - assert_response(resp, "MOCK") + response = Response(responses.GET, url, body="MOCK") - run_mocked() - assert_reset() + @responses.activate + def run_passthrough(): + responses.add(response) + resp = requests.get(url) + assert_response(resp, "OK") - response.passthrough = True - run_passthrough() - assert_reset() + @responses.activate + def run_mocked(): + responses.add(response) + resp = requests.get(url) + assert_response(resp, "MOCK") + run_mocked() + assert_reset() -def test_passthrough_kwarg(httpserver): - httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain") - url = httpserver.url_for("/") + response.passthrough = True + run_passthrough() + assert_reset() - def configure_response(passthrough): - responses.get(url, body="MOCK", passthrough=passthrough) + def test_passthrough_kwarg(self, httpserver): + httpserver.expect_request("/").respond_with_data( + "OK", content_type="text/plain" + ) + url = httpserver.url_for("/") - @responses.activate - def run_passthrough(): - configure_response(passthrough=True) - resp = requests.get(url) - assert_response(resp, "OK") + def configure_response(passthrough): + responses.get(url, body="MOCK", passthrough=passthrough) - @responses.activate - def run_mocked(): - configure_response(passthrough=False) - resp = requests.get(url) - assert_response(resp, "MOCK") + @responses.activate + def run_passthrough(): + configure_response(passthrough=True) + resp = requests.get(url) + assert_response(resp, "OK") - run_mocked() - assert_reset() + @responses.activate + def run_mocked(): + configure_response(passthrough=False) + resp = requests.get(url) + assert_response(resp, "MOCK") - run_passthrough() - assert_reset() + run_mocked() + assert_reset() + run_passthrough() + assert_reset() -def test_passthrough_response(httpserver): - httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain") - url = httpserver.url_for("/") + def test_passthrough_response(self, httpserver): + httpserver.expect_request("/").respond_with_data( + "OK", content_type="text/plain" + ) + url = httpserver.url_for("/") - @responses.activate - def run(): - responses.add(PassthroughResponse(responses.GET, url)) - responses.add(responses.GET, "{}/one".format(url), body="one") - responses.add(responses.GET, "http://example.com/two", body="two") + @responses.activate + def run(): + responses.add(PassthroughResponse(responses.GET, url)) + responses.add(responses.GET, "{}/one".format(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(url)) + assert_response(resp, "one") + resp = requests.get(url) + assert_response(resp, "OK") - resp = requests.get("http://example.com/two") - assert_response(resp, "two") - resp = requests.get("{}/one".format(url)) - assert_response(resp, "one") - resp = requests.get(url) - assert_response(resp, "OK") + assert len(responses.calls) == 3 + responses.assert_call_count(url, 1) - assert len(responses.calls) == 3 - responses.assert_call_count(url, 1) + run() + assert_reset() - run() - assert_reset() + def test_passthrough_response_stream(self, httpserver): + httpserver.expect_request("/").respond_with_data( + "OK", content_type="text/plain" + ) + @responses.activate + def run(): + url = httpserver.url_for("/") + responses.add(PassthroughResponse(responses.GET, url)) + content_1 = requests.get(url).content + with requests.get(url, stream=True) as resp: + content_2 = resp.raw.read() + assert content_1 == content_2 -def test_passthrough_response_stream(httpserver): - httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain") + run() + assert_reset() - @responses.activate - def run(): + def test_passthru_prefixes(self, httpserver): + httpserver.expect_request("/").respond_with_data( + "OK", content_type="text/plain" + ) url = httpserver.url_for("/") - responses.add(PassthroughResponse(responses.GET, url)) - content_1 = requests.get(url).content - with requests.get(url, stream=True) as resp: - content_2 = resp.raw.read() - assert content_1 == content_2 - run() - assert_reset() + @responses.activate + def run_constructor_argument(): + with responses.RequestsMock(passthru_prefixes=(url,)): + resp = requests.get(url) + assert_response(resp, "OK") + @responses.activate + def run_property_setter(): + with responses.RequestsMock() as m: + m.passthru_prefixes = tuple([url]) + resp = requests.get(url) + assert_response(resp, "OK") -def test_passthru_prefixes(httpserver): - httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain") - url = httpserver.url_for("/") + run_constructor_argument() + assert_reset() + run_property_setter() + assert_reset() - @responses.activate - def run_constructor_argument(): - with responses.RequestsMock(passthru_prefixes=(url,)): - resp = requests.get(url) - assert_response(resp, "OK") + def test_passthru(self, httpserver): + httpserver.expect_request("/").respond_with_data( + "OK", content_type="text/plain" + ) + url = httpserver.url_for("/") - @responses.activate - def run_property_setter(): - with responses.RequestsMock() as m: - m.passthru_prefixes = tuple([url]) + @responses.activate + def run(): + responses.add_passthru(url) + responses.add(responses.GET, "{}/one".format(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(url)) + assert_response(resp, "one") resp = requests.get(url) assert_response(resp, "OK") - run_constructor_argument() - assert_reset() - run_property_setter() - assert_reset() - - -def test_passthru(httpserver): - httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain") - url = httpserver.url_for("/") - - @responses.activate - def run(): - responses.add_passthru(url) - responses.add(responses.GET, "{}/one".format(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(url)) - assert_response(resp, "one") - resp = requests.get(url) - assert_response(resp, "OK") - - run() - assert_reset() - - -def test_passthru_regex(httpserver): - httpserver.expect_request(re.compile("^/\\w+")).respond_with_data( - "OK", content_type="text/plain" - ) - url = httpserver.url_for("/") + run() + assert_reset() - @responses.activate - def run(): - responses.add_passthru(re.compile(f"{url}/\\w+")) - responses.add(responses.GET, "{}/one".format(url), body="one") - responses.add(responses.GET, "http://example.com/two", body="two") + def test_passthru_regex(self, httpserver): + httpserver.expect_request(re.compile("^/\\w+")).respond_with_data( + "OK", content_type="text/plain" + ) + url = httpserver.url_for("/") - resp = requests.get("http://example.com/two") - assert_response(resp, "two") - resp = requests.get(f"{url}/one") - assert_response(resp, "one") - resp = requests.get(f"{url}/two") - assert_response(resp, "OK") - resp = requests.get(f"{url}/three") - assert_response(resp, "OK") + @responses.activate + def run(): + responses.add_passthru(re.compile(f"{url}/\\w+")) + responses.add(responses.GET, "{}/one".format(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(f"{url}/one") + assert_response(resp, "one") + resp = requests.get(f"{url}/two") + assert_response(resp, "OK") + resp = requests.get(f"{url}/three") + assert_response(resp, "OK") - run() - assert_reset() + run() + assert_reset() + def test_passthru_does_not_persist_across_tests(self, httpserver): + """ + passthru should be erased on exit from context manager + see: + https://github.com/getsentry/responses/issues/322 + """ + httpserver.expect_request("/").respond_with_data( + "mocked server", status=969, content_type="text/plain" + ) -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.expect_request("/").respond_with_data("OK", content_type="text/plain") + @responses.activate + def with_a_passthru(): + assert not responses.mock.passthru_prefixes + responses.add_passthru(re.compile(".*")) - @responses.activate - def with_a_passthru(): - assert not responses.mock.passthru_prefixes - responses.add_passthru(re.compile(".*")) + url = httpserver.url_for("/") + response = requests.get(url) + assert response.status_code == 969 + assert response.text == "mocked server" - # wrap request that is passed through with another mock. That helps - # to avoid issues if real URL is unavailable, allow to run tests offline - with responses.RequestsMock(target="responses._real_send") as rsp: - rsp.add(responses.GET, "https://example66.ru", status=969) - response = requests.get("https://example66.ru") + @responses.activate + def without_a_passthru(): + assert not responses.mock.passthru_prefixes + with pytest.raises(requests.exceptions.ConnectionError): + requests.get("https://example.com") - assert response.status_code == 969 + with_a_passthru() + without_a_passthru() - @responses.activate - def without_a_passthru(): - assert not responses.mock.passthru_prefixes - with pytest.raises(requests.exceptions.ConnectionError): - requests.get("https://example.com") + def test_passthru_unicode(self): + @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 - with_a_passthru() - without_a_passthru() + run() + assert_reset() def test_method_named_param(): @@ -1827,19 +1844,6 @@ def 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()