From: Tom Christie Date: Tue, 1 Sep 2020 20:44:52 +0000 (+0100) Subject: Use sync client in test cases (#1241) X-Git-Tag: 0.14.3~3 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=e39a6d9ef45bdbf148e9d1126cfe427c633be057;p=thirdparty%2Fhttpx.git Use sync client in test cases (#1241) * Use sync client in test cases * Use plain client __init__ style in preference to context manager Co-authored-by: Florimond Manca --- diff --git a/tests/test_multipart.py b/tests/test_multipart.py index fccfdcd2..f4962dab 100644 --- a/tests/test_multipart.py +++ b/tests/test_multipart.py @@ -8,106 +8,106 @@ import httpcore import pytest import httpx -from httpx._content_streams import AsyncIteratorStream, MultipartStream, encode +from httpx._content_streams import MultipartStream, encode from httpx._utils import format_form_param -class MockTransport(httpcore.AsyncHTTPTransport): - async def request( +class MockTransport(httpcore.SyncHTTPTransport): + def request( self, method: bytes, url: typing.Tuple[bytes, bytes, typing.Optional[int], bytes], headers: typing.List[typing.Tuple[bytes, bytes]] = None, - stream: httpcore.AsyncByteStream = None, + stream: httpcore.SyncByteStream = None, timeout: typing.Mapping[str, typing.Optional[float]] = None, ) -> typing.Tuple[ bytes, int, bytes, typing.List[typing.Tuple[bytes, bytes]], - httpcore.AsyncByteStream, + httpcore.SyncByteStream, ]: assert stream is not None - content = AsyncIteratorStream(aiterator=(part async for part in stream)) + content = httpcore.IteratorByteStream(iterator=(part for part in stream)) return b"HTTP/1.1", 200, b"OK", [], content @pytest.mark.parametrize(("value,output"), (("abc", b"abc"), (b"abc", b"abc"))) -@pytest.mark.asyncio -async def test_multipart(value, output): - async with httpx.AsyncClient(transport=MockTransport()) as client: - # Test with a single-value 'data' argument, and a plain file 'files' argument. - data = {"text": value} - files = {"file": io.BytesIO(b"")} - response = await client.post("http://127.0.0.1:8000/", data=data, files=files) - assert response.status_code == 200 - - # We're using the cgi module to verify the behavior here, which is a - # bit grungy, but sufficient just for our testing purposes. - boundary = response.request.headers["Content-Type"].split("boundary=")[-1] - content_length = response.request.headers["Content-Length"] - pdict: dict = { - "boundary": boundary.encode("ascii"), - "CONTENT-LENGTH": content_length, - } - multipart = cgi.parse_multipart(io.BytesIO(response.content), pdict) - - # Note that the expected return type for text fields - # appears to differs from 3.6 to 3.7+ - assert multipart["text"] == [output.decode()] or multipart["text"] == [output] - assert multipart["file"] == [b""] +def test_multipart(value, output): + client = httpx.Client(transport=MockTransport()) + + # Test with a single-value 'data' argument, and a plain file 'files' argument. + data = {"text": value} + files = {"file": io.BytesIO(b"")} + response = client.post("http://127.0.0.1:8000/", data=data, files=files) + assert response.status_code == 200 + + # We're using the cgi module to verify the behavior here, which is a + # bit grungy, but sufficient just for our testing purposes. + boundary = response.request.headers["Content-Type"].split("boundary=")[-1] + content_length = response.request.headers["Content-Length"] + pdict: dict = { + "boundary": boundary.encode("ascii"), + "CONTENT-LENGTH": content_length, + } + multipart = cgi.parse_multipart(io.BytesIO(response.content), pdict) + + # Note that the expected return type for text fields + # appears to differs from 3.6 to 3.7+ + assert multipart["text"] == [output.decode()] or multipart["text"] == [output] + assert multipart["file"] == [b""] @pytest.mark.parametrize(("key"), (b"abc", 1, 2.3, None)) -@pytest.mark.asyncio -async def test_multipart_invalid_key(key): - async with httpx.AsyncClient(transport=MockTransport()) as client: - data = {key: "abc"} - files = {"file": io.BytesIO(b"")} - with pytest.raises(TypeError) as e: - await client.post( - "http://127.0.0.1:8000/", - data=data, - files=files, - ) - assert "Invalid type for name" in str(e.value) +def test_multipart_invalid_key(key): + client = httpx.Client(transport=MockTransport()) + + data = {key: "abc"} + files = {"file": io.BytesIO(b"")} + with pytest.raises(TypeError) as e: + client.post( + "http://127.0.0.1:8000/", + data=data, + files=files, + ) + assert "Invalid type for name" in str(e.value) @pytest.mark.parametrize(("value"), (1, 2.3, None, [None, "abc"], {None: "abc"})) -@pytest.mark.asyncio -async def test_multipart_invalid_value(value): - async with httpx.AsyncClient(transport=MockTransport()) as client: - data = {"text": value} - files = {"file": io.BytesIO(b"")} - with pytest.raises(TypeError) as e: - await client.post("http://127.0.0.1:8000/", data=data, files=files) - assert "Invalid type for value" in str(e.value) - - -@pytest.mark.asyncio -async def test_multipart_file_tuple(): - async with httpx.AsyncClient(transport=MockTransport()) as client: - # Test with a list of values 'data' argument, - # and a tuple style 'files' argument. - data = {"text": ["abc"]} - files = {"file": ("name.txt", io.BytesIO(b""))} - response = await client.post("http://127.0.0.1:8000/", data=data, files=files) - assert response.status_code == 200 - - # We're using the cgi module to verify the behavior here, which is a - # bit grungy, but sufficient just for our testing purposes. - boundary = response.request.headers["Content-Type"].split("boundary=")[-1] - content_length = response.request.headers["Content-Length"] - pdict: dict = { - "boundary": boundary.encode("ascii"), - "CONTENT-LENGTH": content_length, - } - multipart = cgi.parse_multipart(io.BytesIO(response.content), pdict) - - # Note that the expected return type for text fields - # appears to differs from 3.6 to 3.7+ - assert multipart["text"] == ["abc"] or multipart["text"] == [b"abc"] - assert multipart["file"] == [b""] +def test_multipart_invalid_value(value): + client = httpx.Client(transport=MockTransport()) + + data = {"text": value} + files = {"file": io.BytesIO(b"")} + with pytest.raises(TypeError) as e: + client.post("http://127.0.0.1:8000/", data=data, files=files) + assert "Invalid type for value" in str(e.value) + + +def test_multipart_file_tuple(): + client = httpx.Client(transport=MockTransport()) + + # Test with a list of values 'data' argument, + # and a tuple style 'files' argument. + data = {"text": ["abc"]} + files = {"file": ("name.txt", io.BytesIO(b""))} + response = client.post("http://127.0.0.1:8000/", data=data, files=files) + assert response.status_code == 200 + + # We're using the cgi module to verify the behavior here, which is a + # bit grungy, but sufficient just for our testing purposes. + boundary = response.request.headers["Content-Type"].split("boundary=")[-1] + content_length = response.request.headers["Content-Length"] + pdict: dict = { + "boundary": boundary.encode("ascii"), + "CONTENT-LENGTH": content_length, + } + multipart = cgi.parse_multipart(io.BytesIO(response.content), pdict) + + # Note that the expected return type for text fields + # appears to differs from 3.6 to 3.7+ + assert multipart["text"] == ["abc"] or multipart["text"] == [b"abc"] + assert multipart["file"] == [b""] def test_multipart_encode(tmp_path: typing.Any) -> None: