import asyncio
import os
import pathlib
import socket
import zlib
from typing import Any, Iterable

import pytest

import aiohttp
from aiohttp import web

try:
    import ssl
except ImportError:
    ssl = None  # type: ignore


@pytest.fixture
def loop_without_sendfile(loop):
    def sendfile(*args, **kwargs):
        raise NotImplementedError

    loop.sendfile = sendfile
    return loop


@pytest.fixture
def loop_with_mocked_native_sendfile(loop: Any):
    def sendfile(transport, fobj, offset, count):
        if count == 0:
            raise ValueError("count must be a positive integer (got 0)")
        raise NotImplementedError

    loop.sendfile = sendfile
    return loop


@pytest.fixture(params=["sendfile", "no_sendfile"], ids=["sendfile", "no_sendfile"])
def sender(request, loop_without_sendfile):
    def maker(*args, **kwargs):
        ret = web.FileResponse(*args, **kwargs)
        if request.param == "no_sendfile":
            asyncio.set_event_loop(loop_without_sendfile)
        return ret

    return maker


@pytest.fixture
def app_with_static_route(sender):
    filename = "data.unknown_mime_type"
    filepath = pathlib.Path(__file__).parent / filename

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

    app = web.Application()
    app.router.add_get("/", handler)
    return app


async def test_static_file_ok(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    resp = await client.get("/")
    assert resp.status == 200
    txt = await resp.text()
    assert "file content" == txt.rstrip()
    assert "application/octet-stream" == resp.headers["Content-Type"]
    assert resp.headers.get("Content-Encoding") is None
    await resp.release()
    await client.close()


async def test_zero_bytes_file_ok(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "data.zero_bytes"

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

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

    # Run the request multiple times to ensure
    # that an untrapped exception is not hidden
    # because there is no read of the zero bytes
    for i in range(2):
        resp = await client.get("/")
        assert resp.status == 200
        txt = await resp.text()
        assert "" == txt.rstrip()
        assert "application/octet-stream" == resp.headers["Content-Type"]
        assert resp.headers.get("Content-Encoding") is None
        await resp.release()

    await client.close()


async def test_zero_bytes_file_mocked_native_sendfile(
    aiohttp_client: Any, loop_with_mocked_native_sendfile: Any
) -> None:
    filepath = pathlib.Path(__file__).parent / "data.zero_bytes"

    async def handler(request):
        asyncio.set_event_loop(loop_with_mocked_native_sendfile)
        return web.FileResponse(filepath)

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

    # Run the request multiple times to ensure
    # that an untrapped exception is not hidden
    # because there is no read of the zero bytes
    for i in range(2):
        resp = await client.get("/")
        assert resp.status == 200
        txt = await resp.text()
        assert "" == txt.rstrip()
        assert "application/octet-stream" == resp.headers["Content-Type"]
        assert resp.headers.get("Content-Encoding") is None
        assert resp.headers.get("Content-Length") == "0"
        await resp.release()

    await client.close()


async def test_static_file_ok_string_path(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    resp = await client.get("/")
    assert resp.status == 200
    txt = await resp.text()
    assert "file content" == txt.rstrip()
    assert "application/octet-stream" == resp.headers["Content-Type"]
    assert resp.headers.get("Content-Encoding") is None
    await resp.release()
    await client.close()


async def test_static_file_not_exists(aiohttp_client) -> None:

    app = web.Application()
    client = await aiohttp_client(app)

    resp = await client.get("/fake")
    assert resp.status == 404
    await resp.release()
    await client.close()


async def test_static_file_name_too_long(aiohttp_client) -> None:

    app = web.Application()
    client = await aiohttp_client(app)

    resp = await client.get("/x*500")
    assert resp.status == 404
    await resp.release()
    await client.close()


async def test_static_file_upper_directory(aiohttp_client) -> None:

    app = web.Application()
    client = await aiohttp_client(app)

    resp = await client.get("/../../")
    assert resp.status == 404
    await resp.release()
    await client.close()


async def test_static_file_with_content_type(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "aiohttp.jpg"

    async def handler(request):
        return sender(filepath, chunk_size=16)

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

    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"] == "image/jpeg"
    assert resp.headers.get("Content-Encoding") is None
    resp.close()
    await resp.release()
    await client.close()


async def test_static_file_custom_content_type(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "hello.txt.gz"

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

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

    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
    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 handler(request):
        resp = sender(filepath, chunk_size=16)
        resp.content_type = "application/pdf"
        return resp

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

    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"
    resp.close()
    await resp.release()
    await client.close()


async def test_static_file_with_content_encoding(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "hello.txt.gz"

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

    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
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_modified_since(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    resp = await client.get("/")
    assert 200 == resp.status
    lastmod = resp.headers.get("Last-Modified")
    assert lastmod is not None
    resp.close()
    await resp.release()

    resp = await client.get("/", headers={"If-Modified-Since": lastmod})
    body = await resp.read()
    assert 304 == resp.status
    assert resp.headers.get("Content-Length") is None
    assert resp.headers.get("Last-Modified") == lastmod
    assert b"" == body
    resp.close()
    await resp.release()
    await client.close()


async def test_static_file_if_modified_since_past_date(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Mon, 1 Jan 1990 01:01:01 GMT"

    resp = await client.get("/", headers={"If-Modified-Since": lastmod})
    assert 200 == resp.status
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_modified_since_invalid_date(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "not a valid HTTP-date"

    resp = await client.get("/", headers={"If-Modified-Since": lastmod})
    assert 200 == resp.status
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_modified_since_future_date(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Fri, 31 Dec 9999 23:59:59 GMT"

    resp = await client.get("/", headers={"If-Modified-Since": lastmod})
    body = await resp.read()
    assert 304 == resp.status
    assert resp.headers.get("Content-Length") is None
    assert resp.headers.get("Last-Modified")
    assert b"" == body
    resp.close()

    await resp.release()
    await client.close()


@pytest.mark.parametrize("if_unmodified_since", ("", "Fri, 31 Dec 0000 23:59:59 GMT"))
async def test_static_file_if_match(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
    if_unmodified_since: str,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    resp = await client.get("/")
    assert 200 == resp.status
    original_etag = resp.headers.get("ETag")

    assert original_etag is not None
    resp.close()
    await resp.release()

    headers = {"If-Match": original_etag, "If-Unmodified-Since": if_unmodified_since}
    resp = await client.head("/", headers=headers)
    body = await resp.read()
    assert 200 == resp.status
    assert resp.headers.get("ETag")
    assert resp.headers.get("Last-Modified")
    assert b"" == body
    resp.close()
    await resp.release()

    await client.close()


@pytest.mark.parametrize("if_unmodified_since", ("", "Fri, 31 Dec 0000 23:59:59 GMT"))
@pytest.mark.parametrize(
    "etags,expected_status",
    [
        (("*",), 200),
        (('"example-tag"', 'W/"weak-tag"'), 412),
    ],
)
async def test_static_file_if_match_custom_tags(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
    if_unmodified_since: str,
    etags: Iterable[str],
    expected_status: Iterable[int],
) -> None:
    client = await aiohttp_client(app_with_static_route)

    if_match = ", ".join(etags)
    headers = {"If-Match": if_match, "If-Unmodified-Since": if_unmodified_since}
    resp = await client.head("/", headers=headers)
    body = await resp.read()
    assert expected_status == resp.status
    assert b"" == body
    resp.close()

    await resp.release()
    await client.close()


@pytest.mark.parametrize("if_modified_since", ("", "Fri, 31 Dec 9999 23:59:59 GMT"))
@pytest.mark.parametrize(
    "additional_etags",
    (
        (),
        ('"some-other-strong-etag"', 'W/"weak-tag"', "invalid-tag"),
    ),
)
async def test_static_file_if_none_match(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
    if_modified_since: str,
    additional_etags: Iterable[str],
) -> None:
    client = await aiohttp_client(app_with_static_route)

    resp = await client.get("/")
    assert 200 == resp.status
    original_etag = resp.headers.get("ETag")

    assert resp.headers.get("Last-Modified") is not None
    assert original_etag is not None
    resp.close()
    await resp.release()

    etag = ",".join((original_etag, *additional_etags))

    resp = await client.get(
        "/", headers={"If-None-Match": etag, "If-Modified-Since": if_modified_since}
    )
    body = await resp.read()
    assert 304 == resp.status
    assert resp.headers.get("Content-Length") is None
    assert resp.headers.get("ETag") == original_etag
    assert b"" == body
    resp.close()
    await resp.release()

    await client.close()


async def test_static_file_if_none_match_star(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    resp = await client.head("/", headers={"If-None-Match": "*"})
    body = await resp.read()
    assert 304 == resp.status
    assert resp.headers.get("Content-Length") is None
    assert resp.headers.get("ETag")
    assert resp.headers.get("Last-Modified")
    assert b"" == body
    resp.close()

    await resp.release()
    await client.close()


@pytest.mark.skipif(not ssl, reason="ssl not supported")
async def test_static_file_ssl(
    aiohttp_server,
    ssl_ctx,
    aiohttp_client,
    client_ssl_ctx,
) -> None:
    dirname = os.path.dirname(__file__)
    filename = "data.unknown_mime_type"
    app = web.Application()
    app.router.add_static("/static", dirname)
    server = await aiohttp_server(app, ssl=ssl_ctx)
    conn = aiohttp.TCPConnector(ssl=client_ssl_ctx)
    client = await aiohttp_client(server, connector=conn)

    resp = await client.get("/static/" + filename)
    assert 200 == resp.status
    txt = await resp.text()
    assert "file content" == txt.rstrip()
    ct = resp.headers["CONTENT-TYPE"]
    assert "application/octet-stream" == ct
    assert resp.headers.get("CONTENT-ENCODING") is None

    await resp.release()
    await client.close()


async def test_static_file_directory_traversal_attack(aiohttp_client) -> None:
    dirname = os.path.dirname(__file__)
    relpath = "../README.rst"
    assert os.path.isfile(os.path.join(dirname, relpath))

    app = web.Application()
    app.router.add_static("/static", dirname)
    client = await aiohttp_client(app)

    resp = await client.get("/static/" + relpath)
    assert 404 == resp.status
    await resp.release()

    url_relpath2 = "/static/dir/../" + relpath
    resp = await client.get(url_relpath2)
    assert 404 == resp.status
    await resp.release()

    url_abspath = "/static/" + os.path.abspath(os.path.join(dirname, relpath))
    resp = await client.get(url_abspath)
    assert 403 == resp.status
    await resp.release()

    await client.close()


def test_static_route_path_existence_check() -> None:
    directory = os.path.dirname(__file__)
    web.StaticResource("/", directory)

    nodirectory = os.path.join(directory, "nonexistent-uPNiOEAg5d")
    with pytest.raises(ValueError):
        web.StaticResource("/", nodirectory)


async def test_static_file_huge(aiohttp_client, tmpdir) -> None:
    filename = "huge_data.unknown_mime_type"

    # fill 20MB file
    with tmpdir.join(filename).open("wb") as f:
        for i in range(1024 * 20):
            f.write((chr(i % 64 + 0x20) * 1024).encode())

    file_st = os.stat(str(tmpdir.join(filename)))

    app = web.Application()
    app.router.add_static("/static", str(tmpdir))
    client = await aiohttp_client(app)

    resp = await client.get("/static/" + filename)
    assert 200 == resp.status
    ct = resp.headers["CONTENT-TYPE"]
    assert "application/octet-stream" == ct
    assert resp.headers.get("CONTENT-ENCODING") is None
    assert int(resp.headers.get("CONTENT-LENGTH")) == file_st.st_size

    f = tmpdir.join(filename).open("rb")
    off = 0
    cnt = 0
    while off < file_st.st_size:
        chunk = await resp.content.readany()
        expected = f.read(len(chunk))
        assert chunk == expected
        off += len(chunk)
        cnt += 1
    f.close()

    await resp.release()
    await client.close()


async def test_static_file_range(aiohttp_client: Any, sender: Any) -> None:
    filepath = pathlib.Path(__file__).parent / "sample.txt"

    filesize = filepath.stat().st_size

    async def handler(request):
        return sender(filepath, chunk_size=16)

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

    with filepath.open("rb") as f:
        content = f.read()

    # Ensure the whole file requested in parts is correct
    responses = await asyncio.gather(
        client.get("/", headers={"Range": "bytes=0-999"}),
        client.get("/", headers={"Range": "bytes=1000-1999"}),
        client.get("/", headers={"Range": "bytes=2000-"}),
    )
    assert len(responses) == 3
    assert responses[0].status == 206, "failed 'bytes=0-999': %s" % responses[0].reason
    assert responses[0].headers["Content-Range"] == "bytes 0-999/{}".format(
        filesize
    ), "failed: Content-Range Error"
    assert responses[1].status == 206, (
        "failed 'bytes=1000-1999': %s" % responses[1].reason
    )
    assert responses[1].headers["Content-Range"] == "bytes 1000-1999/{}".format(
        filesize
    ), "failed: Content-Range Error"
    assert responses[2].status == 206, "failed 'bytes=2000-': %s" % responses[2].reason
    assert responses[2].headers["Content-Range"] == "bytes 2000-{}/{}".format(
        filesize - 1, filesize
    ), "failed: Content-Range Error"

    body = await asyncio.gather(
        *(resp.read() for resp in responses),
    )

    assert len(body[0]) == 1000, "failed 'bytes=0-999', received %d bytes" % len(
        body[0]
    )
    assert len(body[1]) == 1000, "failed 'bytes=1000-1999', received %d bytes" % len(
        body[1]
    )
    responses[0].close()
    responses[1].close()
    responses[2].close()

    await asyncio.gather(
        *(resp.release() for resp in responses),
    )

    assert content == b"".join(body)

    await client.close()


async def test_static_file_range_end_bigger_than_size(aiohttp_client, sender):
    filepath = pathlib.Path(__file__).parent / "aiohttp.png"

    async def handler(request):
        return sender(filepath, chunk_size=16)

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

    with filepath.open("rb") as f:
        content = f.read()

        # Ensure the whole file requested in parts is correct
        response = await client.get("/", headers={"Range": "bytes=54000-55000"})

        assert response.status == 206, (
            "failed 'bytes=54000-55000': %s" % response.reason
        )
        assert (
            response.headers["Content-Range"] == "bytes 54000-54996/54997"
        ), "failed: Content-Range Error"

        body = await response.read()
        assert len(body) == 997, "failed 'bytes=54000-55000', received %d bytes" % len(
            body
        )

        assert content[54000:] == body

    await response.release()
    await client.close()


async def test_static_file_range_beyond_eof(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "aiohttp.png"

    async def handler(request):
        return sender(filepath, chunk_size=16)

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

    # Ensure the whole file requested in parts is correct
    response = await client.get("/", headers={"Range": "bytes=1000000-1200000"})

    assert response.status == 416, (
        "failed 'bytes=1000000-1200000': %s" % response.reason
    )

    await response.release()
    await client.close()


async def test_static_file_range_tail(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "aiohttp.png"

    async def handler(request):
        return sender(filepath, chunk_size=16)

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

    with filepath.open("rb") as f:
        content = f.read()

    # Ensure the tail of the file is correct
    resp = await client.get("/", headers={"Range": "bytes=-500"})
    assert resp.status == 206, resp.reason
    assert (
        resp.headers["Content-Range"] == "bytes 54497-54996/54997"
    ), "failed: Content-Range Error"
    body4 = await resp.read()
    resp.close()
    await resp.release()
    assert content[-500:] == body4

    # Ensure out-of-range tails could be handled
    resp2 = await client.get("/", headers={"Range": "bytes=-99999999999999"})
    assert resp2.status == 206, resp.reason
    assert (
        resp2.headers["Content-Range"] == "bytes 0-54996/54997"
    ), "failed: Content-Range Error"
    await resp2.release()

    await client.close()


async def test_static_file_invalid_range(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "aiohttp.png"

    async def handler(request):
        return sender(filepath, chunk_size=16)

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

    # range must be in bytes
    resp = await client.get("/", headers={"Range": "blocks=0-10"})
    assert resp.status == 416, "Range must be in bytes"
    resp.close()
    await resp.release()

    # start > end
    resp = await client.get("/", headers={"Range": "bytes=100-0"})
    assert resp.status == 416, "Range start can't be greater than end"
    resp.close()
    await resp.release()

    # start > end
    resp = await client.get("/", headers={"Range": "bytes=10-9"})
    assert resp.status == 416, "Range start can't be greater than end"
    resp.close()
    await resp.release()

    # non-number range
    resp = await client.get("/", headers={"Range": "bytes=a-f"})
    assert resp.status == 416, "Range must be integers"
    resp.close()
    await resp.release()

    # double dash range
    resp = await client.get("/", headers={"Range": "bytes=0--10"})
    assert resp.status == 416, "double dash in range"
    resp.close()
    await resp.release()

    # no range
    resp = await client.get("/", headers={"Range": "bytes=-"})
    assert resp.status == 416, "no range given"
    resp.close()
    await resp.release()

    await client.close()


async def test_static_file_if_unmodified_since_past_with_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Mon, 1 Jan 1990 01:01:01 GMT"

    resp = await client.get(
        "/", headers={"If-Unmodified-Since": lastmod, "Range": "bytes=2-"}
    )
    assert 412 == resp.status
    resp.close()
    await resp.release()

    await client.close()


async def test_static_file_if_unmodified_since_future_with_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Fri, 31 Dec 9999 23:59:59 GMT"

    resp = await client.get(
        "/", headers={"If-Unmodified-Since": lastmod, "Range": "bytes=2-"}
    )
    assert 206 == resp.status
    assert resp.headers["Content-Range"] == "bytes 2-12/13"
    assert resp.headers["Content-Length"] == "11"
    resp.close()
    await resp.release()

    await client.close()


async def test_static_file_if_range_past_with_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Mon, 1 Jan 1990 01:01:01 GMT"

    resp = await client.get("/", headers={"If-Range": lastmod, "Range": "bytes=2-"})
    assert 200 == resp.status
    assert resp.headers["Content-Length"] == "13"
    resp.close()
    await resp.release()
    await client.close()


async def test_static_file_if_range_future_with_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Fri, 31 Dec 9999 23:59:59 GMT"

    resp = await client.get("/", headers={"If-Range": lastmod, "Range": "bytes=2-"})
    assert 206 == resp.status
    assert resp.headers["Content-Range"] == "bytes 2-12/13"
    assert resp.headers["Content-Length"] == "11"
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_unmodified_since_past_without_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Mon, 1 Jan 1990 01:01:01 GMT"

    resp = await client.get("/", headers={"If-Unmodified-Since": lastmod})
    assert 412 == resp.status
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_unmodified_since_future_without_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Fri, 31 Dec 9999 23:59:59 GMT"

    resp = await client.get("/", headers={"If-Unmodified-Since": lastmod})
    assert 200 == resp.status
    assert resp.headers["Content-Length"] == "13"
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_range_past_without_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Mon, 1 Jan 1990 01:01:01 GMT"

    resp = await client.get("/", headers={"If-Range": lastmod})
    assert 200 == resp.status
    assert resp.headers["Content-Length"] == "13"
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_range_future_without_range(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "Fri, 31 Dec 9999 23:59:59 GMT"

    resp = await client.get("/", headers={"If-Range": lastmod})
    assert 200 == resp.status
    assert resp.headers["Content-Length"] == "13"
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_unmodified_since_invalid_date(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "not a valid HTTP-date"

    resp = await client.get("/", headers={"If-Unmodified-Since": lastmod})
    assert 200 == resp.status
    resp.close()

    await resp.release()
    await client.close()


async def test_static_file_if_range_invalid_date(
    aiohttp_client: Any,
    app_with_static_route: web.Application,
) -> None:
    client = await aiohttp_client(app_with_static_route)

    lastmod = "not a valid HTTP-date"

    resp = await client.get("/", headers={"If-Range": lastmod})
    assert 200 == resp.status
    resp.close()
    await resp.release()

    await client.close()


async def test_static_file_compression(aiohttp_client, sender) -> None:
    filepath = pathlib.Path(__file__).parent / "data.unknown_mime_type"

    async def handler(request):
        ret = sender(filepath)
        ret.enable_compression()
        return ret

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

    resp = await client.get("/")
    assert resp.status == 200
    zcomp = zlib.compressobj(wbits=zlib.MAX_WBITS)
    expected_body = zcomp.compress(b"file content\n") + zcomp.flush()
    assert expected_body == await resp.read()
    assert "application/octet-stream" == resp.headers["Content-Type"]
    assert resp.headers.get("Content-Encoding") == "deflate"
    await resp.release()

    await client.close()


async def test_static_file_huge_cancel(aiohttp_client, tmpdir) -> None:
    filename = "huge_data.unknown_mime_type"

    # fill 100MB file
    with tmpdir.join(filename).open("wb") as f:
        for i in range(1024 * 20):
            f.write((chr(i % 64 + 0x20) * 1024).encode())

    task = None

    async def handler(request):
        nonlocal task
        task = request.task
        # reduce send buffer size
        tr = request.transport
        sock = tr.get_extra_info("socket")
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024)
        ret = web.FileResponse(pathlib.Path(str(tmpdir.join(filename))))
        return ret

    app = web.Application()

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

    resp = await client.get("/")
    assert resp.status == 200
    task.cancel()
    await asyncio.sleep(0)
    data = b""
    while True:
        try:
            data += await resp.content.read(1024)
        except aiohttp.ClientPayloadError:
            break
    assert len(data) < 1024 * 1024 * 20

    await resp.release()
    await client.close()


async def test_static_file_huge_error(aiohttp_client, tmpdir) -> None:
    filename = "huge_data.unknown_mime_type"

    # fill 20MB file
    with tmpdir.join(filename).open("wb") as f:
        f.seek(20 * 1024 * 1024)
        f.write(b"1")

    async def handler(request):
        # reduce send buffer size
        tr = request.transport
        sock = tr.get_extra_info("socket")
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024)
        ret = web.FileResponse(pathlib.Path(str(tmpdir.join(filename))))
        return ret

    app = web.Application()

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

    resp = await client.get("/")
    assert resp.status == 200
    # raise an exception on server side
    resp.close()

    await resp.release()
    await client.close()