Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Create hello.txt.gz dynamically and improve related assertions #8136

Merged
merged 8 commits into from
Feb 13, 2024
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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
1 change: 1 addition & 0 deletions CHANGES/8136.misc.rst
steverep marked this conversation as resolved.
Show resolved Hide resolved
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Created ``hello.txt.gz`` dynamically and improved related assertions -- by :user:`steverep`.
steverep marked this conversation as resolved.
Show resolved Hide resolved
Binary file removed tests/hello.txt.gz
Binary file not shown.
68 changes: 37 additions & 31 deletions tests/test_web_sendfile_functional.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
# type: ignore
import asyncio
import gzip
import pathlib
import socket
import zlib
Expand All @@ -16,6 +17,19 @@
ssl = None


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


@pytest.fixture(scope="module")
def hello_txt(tmp_path_factory) -> pathlib.Path:
steverep marked this conversation as resolved.
Show resolved Hide resolved
"""Create a temp path with hello.txt and compressed versions."""
hello = tmp_path_factory.mktemp("hello-") / "hello.txt"
hello.write_bytes(HELLO_AIOHTTP)
hello_gzip = hello.with_suffix(f"{hello.suffix}.gz")
hello_gzip.write_bytes(gzip.compress(HELLO_AIOHTTP))
return hello


@pytest.fixture
def loop_without_sendfile(loop: Any):
def sendfile(*args, **kwargs):
Expand Down Expand Up @@ -198,9 +212,10 @@ async def handler(request):


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."""
filepath = hello_txt.with_suffix(f"{hello_txt.suffix}.gz")

async def handler(request):
resp = sender(filepath, chunk_size=16)
Expand All @@ -213,24 +228,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() == filepath.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 request."""
steverep marked this conversation as resolved.
Show resolved Hide resolved

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,23 +252,21 @@ 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
async def test_static_file_with_encoding_and_enable_compression(
hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any
):
"""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 = sender(hello_txt)
resp.enable_compression()
return resp

Expand All @@ -266,35 +276,31 @@ 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()


async def test_static_file_with_content_encoding(
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 correct encoded files are returned when available."""

async def handler(request):
return sender(filepath)
return sender(hello_txt.with_suffix(f"{hello_txt.suffix}.gz"))

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") == "gzip"
assert resp.headers["Content-Type"] == "text/plain"
assert await resp.read() == HELLO_AIOHTTP
resp.close()

await resp.release()
Expand Down
Loading