[3.9] Create hello.txt.gz dynamically and improve related assertions (#…

…8136) (#8157)

Co-authored-by: Sviatoslav Sydorenko (Святослав Сидоренко) <wk.cvs.github@sydorenko.org.ua>
Co-authored-by: J. Nick Koston <nick@koston.org>

.gitattributes

@@ -1,3 +1,2 @@
 tests/data.unknown_mime_type binary
-tests/hello.txt.gz binary
 tests/sample.* binary

CHANGES/8136.contrib.rst

@@ -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`

tests/test_web_sendfile_functional.py

@@ -1,8 +1,9 @@
 import asyncio
+import gzip
 import pathlib
 import socket
 import zlib
-from typing import Any, Iterable
+from typing import Any, Iterable, Optional
 
 import pytest
 
@@ -15,6 +16,24 @@
     ssl = None  # type: ignore
 
 
+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):
     def sendfile(*args, **kwargs):
@@ -201,11 +220,14 @@ async def handler(request):
     await client.close()
 
 
-async def test_static_file_custom_content_type(aiohttp_client, sender) -> None:
-    filepath = pathlib.Path(__file__).parent / "hello.txt.gz"
+@pytest.mark.parametrize("hello_txt", ["gzip"], indirect=True)
+async def test_static_file_custom_content_type(
+    hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any
+) -> None:
+    """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
 
@@ -215,22 +237,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, sender):
-    filepath = pathlib.Path(__file__).parent / "hello.txt"
+async def test_static_file_custom_content_type_compress(
+    hello_txt: pathlib.Path, aiohttp_client: Any, sender: Any
+):
+    """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
 
@@ -240,24 +261,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()
@@ -266,35 +289,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()