Skip to content

Commit

Permalink
Create hello.txt.gz dynamically and improve related assertions (aio-l…
Browse files Browse the repository at this point in the history
…ibs#8136)

Co-authored-by: Sviatoslav Sydorenko (Святослав Сидоренко) <[email protected]>
Co-authored-by: J. Nick Koston <[email protected]>
  • Loading branch information
3 people authored Feb 13, 2024
1 parent fb465e1 commit 026e8d9
Show file tree
Hide file tree
Showing 4 changed files with 59 additions and 35 deletions.
1 change: 0 additions & 1 deletion .gitattributes
Original file line number Diff line number Diff line change
@@ -1,3 +1,2 @@
tests/data.unknown_mime_type binary
tests/hello.txt.gz binary
tests/sample.* binary
7 changes: 7 additions & 0 deletions CHANGES/8136.contrib.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
A pytest fixture ``hello_txt`` was introduced to aid
static file serving tests in
:file:`test_web_sendfile_functional.py`. It dynamically
provisions ``hello.txt`` file variants shared across the
tests in the module.

-- by :user:`steverep`
Binary file removed tests/hello.txt.gz
Binary file not shown.
86 changes: 52 additions & 34 deletions tests/test_web_sendfile_functional.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
# type: ignore
import asyncio
import gzip
import pathlib
import socket
import zlib
from typing import Any, Iterable
from typing import Any, Iterable, Optional

import pytest

Expand All @@ -16,6 +17,24 @@
ssl = None


HELLO_AIOHTTP = b"Hello aiohttp! :-)\n"


@pytest.fixture(scope="module")
def hello_txt(request, tmp_path_factory) -> pathlib.Path:
"""Create a temp path with hello.txt and compressed versions.
The uncompressed text file path is returned by default. Alternatively, an
indirect parameter can be passed with an encoding to get a compressed path.
"""
txt = tmp_path_factory.mktemp("hello-") / "hello.txt"
hello = {None: txt, "gzip": txt.with_suffix(f"{txt.suffix}.gz")}
hello[None].write_bytes(HELLO_AIOHTTP)
hello["gzip"].write_bytes(gzip.compress(HELLO_AIOHTTP))
encoding = getattr(request, "param", None)
return hello[encoding]


@pytest.fixture
def loop_without_sendfile(loop: Any):
def sendfile(*args, **kwargs):
Expand Down Expand Up @@ -197,13 +216,14 @@ async def handler(request):
await client.close()


@pytest.mark.parametrize("hello_txt", ["gzip"], indirect=True)
async def test_static_file_custom_content_type(
aiohttp_client: Any, sender: Any
hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any
) -> None:
filepath = pathlib.Path(__file__).parent / "hello.txt.gz"
"""Test that custom type without encoding is returned for encoded request."""

async def handler(request):
resp = sender(filepath, chunk_size=16)
resp = sender(hello_txt, chunk_size=16)
resp.content_type = "application/pdf"
return resp

Expand All @@ -213,24 +233,21 @@ async def handler(request):

resp = await client.get("/")
assert resp.status == 200
body = await resp.read()
with filepath.open("rb") as f:
content = f.read()
assert content == body
assert resp.headers["Content-Type"] == "application/pdf"
assert resp.headers.get("Content-Encoding") is None
assert resp.headers["Content-Type"] == "application/pdf"
assert await resp.read() == hello_txt.read_bytes()
resp.close()
await resp.release()
await client.close()


async def test_static_file_custom_content_type_compress(
aiohttp_client: Any, sender: Any
hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any
):
filepath = pathlib.Path(__file__).parent / "hello.txt"
"""Test that custom type with encoding is returned for unencoded requests."""

async def handler(request):
resp = sender(filepath, chunk_size=16)
resp = sender(hello_txt, chunk_size=16)
resp.content_type = "application/pdf"
return resp

Expand All @@ -240,24 +257,26 @@ async def handler(request):

resp = await client.get("/")
assert resp.status == 200
body = await resp.read()
assert b"hello aiohttp\n" == body
assert resp.headers["Content-Type"] == "application/pdf"
assert resp.headers.get("Content-Encoding") == "gzip"
assert resp.headers["Content-Type"] == "application/pdf"
assert await resp.read() == HELLO_AIOHTTP
resp.close()
await resp.release()
await client.close()


async def test_static_file_with_gziped_counter_part_enable_compression(
aiohttp_client: Any, sender: Any
@pytest.mark.parametrize("forced_compression", [None, web.ContentCoding.gzip])
async def test_static_file_with_encoding_and_enable_compression(
hello_txt: pathlib.Path,
aiohttp_client: Any,
sender: Any,
forced_compression: Optional[web.ContentCoding],
):
"""Test that enable_compression does not double compress when a .gz file is also present."""
filepath = pathlib.Path(__file__).parent / "hello.txt"
"""Test that enable_compression does not double compress when an encoded file is also present."""

async def handler(request):
resp = sender(filepath)
resp.enable_compression()
resp = sender(hello_txt)
resp.enable_compression(forced_compression)
return resp

app = web.Application()
Expand All @@ -266,35 +285,34 @@ async def handler(request):

resp = await client.get("/")
assert resp.status == 200
body = await resp.read()
assert body == b"hello aiohttp\n"
assert resp.headers["Content-Type"] == "text/plain"
assert resp.headers.get("Content-Encoding") == "gzip"
assert resp.headers["Content-Type"] == "text/plain"
assert await resp.read() == HELLO_AIOHTTP
resp.close()
await resp.release()
await client.close()


@pytest.mark.parametrize(
("hello_txt", "expect_encoding"), [["gzip"] * 2], indirect=["hello_txt"]
)
async def test_static_file_with_content_encoding(
aiohttp_client: Any, sender: Any
hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any, expect_encoding: str
) -> None:
filepath = pathlib.Path(__file__).parent / "hello.txt.gz"
"""Test requesting static compressed files returns the correct content type and encoding."""

async def handler(request):
return sender(filepath)
return sender(hello_txt)

app = web.Application()
app.router.add_get("/", handler)
client = await aiohttp_client(app)

resp = await client.get("/")
assert 200 == resp.status
body = await resp.read()
assert b"hello aiohttp\n" == body
ct = resp.headers["CONTENT-TYPE"]
assert "text/plain" == ct
encoding = resp.headers["CONTENT-ENCODING"]
assert "gzip" == encoding
assert resp.status == 200
assert resp.headers.get("Content-Encoding") == expect_encoding
assert resp.headers["Content-Type"] == "text/plain"
assert await resp.read() == HELLO_AIOHTTP
resp.close()

await resp.release()
Expand Down

0 comments on commit 026e8d9

Please sign in to comment.