Skip to content

Commit

Permalink
change pytest fixture to httpserver (#566)
Browse files Browse the repository at this point in the history
  • Loading branch information
@beliaev-maksim
beliaev-maksim authored Jul 15, 2022
1 parent b86e3dd commit d49f829
Show file tree
Hide file tree
Showing 2 changed files with 51 additions and 38 deletions.
87 changes: 50 additions & 37 deletions responses/tests/test_responses.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1566,19 +1566,21 @@ def run():


def test_passthrough_flag(httpserver):
httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
response = Response(responses.GET, httpserver.url, body="MOCK")
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(httpserver.url)
resp = requests.get(url)
assert_response(resp, "OK")

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

run_mocked()
Expand All @@ -1590,21 +1592,22 @@ def run_mocked():


def test_passthrough_kwarg(httpserver):
httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain")
url = httpserver.url_for("/")

def configure_response(passthrough):
responses.get(httpserver.url, body="MOCK", passthrough=passthrough)
responses.get(url, body="MOCK", passthrough=passthrough)

@responses.activate
def run_passthrough():
configure_response(passthrough=True)
resp = requests.get(httpserver.url)
resp = requests.get(url)
assert_response(resp, "OK")

@responses.activate
def run_mocked():
configure_response(passthrough=False)
resp = requests.get(httpserver.url)
resp = requests.get(url)
assert_response(resp, "MOCK")

run_mocked()
Expand All @@ -1615,36 +1618,38 @@ def run_mocked():


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

@responses.activate
def run():
responses.add(PassthroughResponse(responses.GET, httpserver.url))
responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
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(httpserver.url))
resp = requests.get("{}/one".format(url))
assert_response(resp, "one")
resp = requests.get(httpserver.url)
resp = requests.get(url)
assert_response(resp, "OK")

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

run()
assert_reset()


def test_passthrough_response_stream(httpserver):
httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
httpserver.expect_request("/").respond_with_data("OK", 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:
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

Expand All @@ -1653,19 +1658,20 @@ def run():


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

@responses.activate
def run_constructor_argument():
with responses.RequestsMock(passthru_prefixes=(httpserver.url,)):
resp = requests.get(httpserver.url)
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([httpserver.url])
resp = requests.get(httpserver.url)
m.passthru_prefixes = tuple([url])
resp = requests.get(url)
assert_response(resp, "OK")

run_constructor_argument()
Expand All @@ -1675,41 +1681,45 @@ def run_property_setter():


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

@responses.activate
def run():
responses.add_passthru(httpserver.url)
responses.add(responses.GET, "{}/one".format(httpserver.url), body="one")
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(httpserver.url))
resp = requests.get("{}/one".format(url))
assert_response(resp, "one")
resp = requests.get(httpserver.url)
resp = requests.get(url)
assert_response(resp, "OK")

run()
assert_reset()


def test_passthru_regex(httpserver):
httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
httpserver.expect_request(re.compile("^/\\w+")).respond_with_data(
"OK", content_type="text/plain"
)
url = httpserver.url_for("/")

@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_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("{}/one".format(httpserver.url))
resp = requests.get(f"{url}/one")
assert_response(resp, "one")
resp = requests.get("{}/two".format(httpserver.url))
resp = requests.get(f"{url}/two")
assert_response(resp, "OK")
resp = requests.get("{}/three".format(httpserver.url))
resp = requests.get(f"{url}/three")
assert_response(resp, "OK")

run()
Expand All @@ -1722,7 +1732,7 @@ def test_passthru_does_not_persist_across_tests(httpserver):
see:
https://github.com/getsentry/responses/issues/322
"""
httpserver.serve_content("OK", headers={"Content-Type": "text/plain"})
httpserver.expect_request("/").respond_with_data("OK", content_type="text/plain")

@responses.activate
def with_a_passthru():
Expand Down Expand Up @@ -2181,9 +2191,12 @@ def test_mock_not_leaked(self, httpserver):
Mock real HTTP server
"""
httpserver.serve_content("OK", code=969, headers={"Content-Type": "text/plain"})
httpserver.expect_request("/").respond_with_data(
"OK", content_type="text/plain", status=969
)
url = httpserver.url_for("/")

response = requests.get(httpserver.url)
response = requests.get(url)
assert response.status_code == 969


Expand Down
2 changes: 1 addition & 1 deletion setup.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
"coverage >= 6.0.0",
"pytest-cov",
"pytest-asyncio",
"pytest-localserver",
"pytest-httpserver",
"flake8",
"types-requests",
"mypy",
Expand Down

0 comments on commit d49f829

Please sign in to comment.