02a692aba5
* Maybe port is `None` https://www.python.org/dev/peps/pep-3333/#environ-variables > SERVER_NAME, SERVER_PORT > When HTTP_HOST is not set, these variables can be combined to determine a default. See the URL Reconstruction section below for more detail. SERVER_NAME and SERVER_PORT are required strings and must never be empty. * Add unit test * Compute default port Co-authored-by: Florimond Manca <florimond.manca@gmail.com>
146 lines
3.8 KiB
Python
146 lines
3.8 KiB
Python
import sys
|
|
from functools import partial
|
|
|
|
import pytest
|
|
|
|
import httpx
|
|
|
|
|
|
def application_factory(output):
|
|
def application(environ, start_response):
|
|
status = "200 OK"
|
|
|
|
response_headers = [
|
|
("Content-type", "text/plain"),
|
|
]
|
|
|
|
start_response(status, response_headers)
|
|
|
|
for item in output:
|
|
yield item
|
|
|
|
return application
|
|
|
|
|
|
def echo_body(environ, start_response):
|
|
status = "200 OK"
|
|
output = environ["wsgi.input"].read()
|
|
|
|
response_headers = [
|
|
("Content-type", "text/plain"),
|
|
]
|
|
|
|
start_response(status, response_headers)
|
|
|
|
return [output]
|
|
|
|
|
|
def echo_body_with_response_stream(environ, start_response):
|
|
status = "200 OK"
|
|
|
|
response_headers = [("Content-Type", "text/plain")]
|
|
|
|
start_response(status, response_headers)
|
|
|
|
def output_generator(f):
|
|
while True:
|
|
output = f.read(2)
|
|
if not output:
|
|
break
|
|
yield output
|
|
|
|
return output_generator(f=environ["wsgi.input"])
|
|
|
|
|
|
def raise_exc(environ, start_response, exc=ValueError):
|
|
status = "500 Server Error"
|
|
output = b"Nope!"
|
|
|
|
response_headers = [
|
|
("Content-type", "text/plain"),
|
|
]
|
|
|
|
try:
|
|
raise exc()
|
|
except exc:
|
|
exc_info = sys.exc_info()
|
|
start_response(status, response_headers, exc_info=exc_info)
|
|
|
|
return [output]
|
|
|
|
|
|
def test_wsgi():
|
|
client = httpx.Client(app=application_factory([b"Hello, World!"]))
|
|
response = client.get("http://www.example.org/")
|
|
assert response.status_code == 200
|
|
assert response.text == "Hello, World!"
|
|
|
|
|
|
def test_wsgi_upload():
|
|
client = httpx.Client(app=echo_body)
|
|
response = client.post("http://www.example.org/", content=b"example")
|
|
assert response.status_code == 200
|
|
assert response.text == "example"
|
|
|
|
|
|
def test_wsgi_upload_with_response_stream():
|
|
client = httpx.Client(app=echo_body_with_response_stream)
|
|
response = client.post("http://www.example.org/", content=b"example")
|
|
assert response.status_code == 200
|
|
assert response.text == "example"
|
|
|
|
|
|
def test_wsgi_exc():
|
|
client = httpx.Client(app=raise_exc)
|
|
with pytest.raises(ValueError):
|
|
client.get("http://www.example.org/")
|
|
|
|
|
|
def test_wsgi_http_error():
|
|
client = httpx.Client(app=partial(raise_exc, exc=RuntimeError))
|
|
with pytest.raises(RuntimeError):
|
|
client.get("http://www.example.org/")
|
|
|
|
|
|
def test_wsgi_generator():
|
|
output = [b"", b"", b"Some content", b" and more content"]
|
|
client = httpx.Client(app=application_factory(output))
|
|
response = client.get("http://www.example.org/")
|
|
assert response.status_code == 200
|
|
assert response.text == "Some content and more content"
|
|
|
|
|
|
def test_wsgi_generator_empty():
|
|
output = [b"", b"", b"", b""]
|
|
client = httpx.Client(app=application_factory(output))
|
|
response = client.get("http://www.example.org/")
|
|
assert response.status_code == 200
|
|
assert response.text == ""
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"url, expected_server_port",
|
|
[
|
|
pytest.param("http://www.example.org", "80", id="auto-http"),
|
|
pytest.param("https://www.example.org", "443", id="auto-https"),
|
|
pytest.param("http://www.example.org:8000", "8000", id="explicit-port"),
|
|
],
|
|
)
|
|
def test_wsgi_server_port(url: str, expected_server_port: int):
|
|
"""
|
|
SERVER_PORT is populated correctly from the requested URL.
|
|
"""
|
|
hello_world_app = application_factory([b"Hello, World!"])
|
|
server_port: str
|
|
|
|
def app(environ, start_response):
|
|
nonlocal server_port
|
|
server_port = environ["SERVER_PORT"]
|
|
return hello_world_app(environ, start_response)
|
|
|
|
client = httpx.Client(app=app)
|
|
response = client.get(url)
|
|
assert response.status_code == 200
|
|
assert response.text == "Hello, World!"
|
|
assert server_port == expected_server_port
|