import asyncio
from collections.abc import Awaitable, Callable, Iterable
from contextlib import suppress
from typing import NoReturn

import pytest
from pytest_aiohttp import AiohttpClient, AiohttpServer
from yarl import URL

from aiohttp import web, web_app
from aiohttp.test_utils import TestClient
from aiohttp.typedefs import Handler, Middleware

CLI = Callable[
    [Iterable[Middleware]], Awaitable[TestClient[web.Request, web.Application]]
]


async def test_middleware_modifies_response(aiohttp_client: AiohttpClient) -> None:
    async def handler(request: web.Request) -> web.Response:
        return web.Response(body=b"OK")

    async def middleware(request: web.Request, handler: Handler) -> web.Response:
        resp = await handler(request)
        assert 200 == resp.status
        resp.set_status(201)
        assert isinstance(resp, web.Response)
        assert resp.text is not None
        resp.text = resp.text + "[MIDDLEWARE]"
        return resp

    app = web.Application()
    app.middlewares.append(middleware)
    app.router.add_route("GET", "/", handler)
    client = await aiohttp_client(app)

    # Call twice to verify cache works
    for _ in range(2):
        resp = await client.get("/")
        assert 201 == resp.status
        txt = await resp.text()
        assert "OK[MIDDLEWARE]" == txt


async def test_middleware_handles_exception(aiohttp_client: AiohttpClient) -> None:
    async def handler(request: web.Request) -> NoReturn:
        raise RuntimeError("Error text")

    async def middleware(request: web.Request, handler: Handler) -> web.Response:
        with pytest.raises(RuntimeError) as ctx:
            await handler(request)
        return web.Response(status=501, text=str(ctx.value) + "[MIDDLEWARE]")

    app = web.Application()
    app.middlewares.append(middleware)
    app.router.add_route("GET", "/", handler)
    client = await aiohttp_client(app)

    # Call twice to verify cache works
    for _ in range(2):
        resp = await client.get("/")
        assert 501 == resp.status
        txt = await resp.text()
        assert "Error text[MIDDLEWARE]" == txt


async def test_middleware_chain(aiohttp_client: AiohttpClient) -> None:
    async def handler(request: web.Request) -> web.Response:
        return web.Response(text="OK")

    handler.annotation = "annotation_value"  # type: ignore[attr-defined]

    async def handler2(request: web.Request) -> web.Response:
        return web.Response(text="OK")

    middleware_annotation_seen_values = []

    def make_middleware(num: int) -> Middleware:
        async def middleware(request: web.Request, handler: Handler) -> web.Response:
            middleware_annotation_seen_values.append(
                getattr(handler, "annotation", None)
            )
            resp = await handler(request)
            assert isinstance(resp, web.Response)
            assert resp.text is not None
            resp.text = resp.text + f"[{num}]"
            return resp

        return middleware

    app = web.Application()
    app.middlewares.append(make_middleware(1))
    app.middlewares.append(make_middleware(2))
    app.router.add_route("GET", "/", handler)
    app.router.add_route("GET", "/r2", handler2)
    client = await aiohttp_client(app)
    resp = await client.get("/")
    assert 200 == resp.status
    txt = await resp.text()
    assert "OK[2][1]" == txt
    assert middleware_annotation_seen_values == ["annotation_value", "annotation_value"]

    # check that attributes from handler are not applied to handler2
    resp = await client.get("/r2")
    assert 200 == resp.status
    assert middleware_annotation_seen_values == [
        "annotation_value",
        "annotation_value",
        None,
        None,
    ]


async def test_middleware_subapp(aiohttp_client: AiohttpClient) -> None:
    async def sub_handler(request: web.Request) -> web.Response:
        return web.Response(text="OK")

    sub_handler.annotation = "annotation_value"  # type: ignore[attr-defined]

    async def handler(request: web.Request) -> web.Response:
        return web.Response(text="OK")

    middleware_annotation_seen_values = []

    def make_middleware(num: int) -> Middleware:
        async def middleware(
            request: web.Request, handler: Handler
        ) -> web.StreamResponse:
            annotation = getattr(handler, "annotation", None)
            if annotation is not None:
                middleware_annotation_seen_values.append(f"{annotation}/{num}")
            return await handler(request)

        return middleware

    app = web.Application()
    app.middlewares.append(make_middleware(1))
    app.router.add_route("GET", "/r2", handler)

    subapp = web.Application()
    subapp.middlewares.append(make_middleware(2))
    subapp.router.add_route("GET", "/", sub_handler)
    app.add_subapp("/sub", subapp)

    client = await aiohttp_client(app)
    resp = await client.get("/sub/")
    assert 200 == resp.status
    await resp.text()
    assert middleware_annotation_seen_values == [
        "annotation_value/1",
        "annotation_value/2",
    ]

    # check that attributes from sub_handler are not applied to handler
    del middleware_annotation_seen_values[:]
    resp = await client.get("/r2")
    assert 200 == resp.status
    assert middleware_annotation_seen_values == []


@pytest.fixture
def cli(aiohttp_client: AiohttpClient) -> CLI:
    async def handler(request: web.Request) -> web.Response:
        return web.Response(text="OK")

    def wrapper(
        extra_middlewares: Iterable[Middleware],
    ) -> Awaitable[TestClient[web.Request, web.Application]]:
        app = web.Application()
        app.router.add_route("GET", "/resource1", handler)
        app.router.add_route("GET", "/resource2/", handler)
        app.router.add_route("GET", "/resource1/a/b", handler)
        app.router.add_route("GET", "/resource2/a/b/", handler)
        app.router.add_route("GET", "/resource2/a/b%2Fc/", handler)
        app.middlewares.extend(extra_middlewares)
        return aiohttp_client(app, server_kwargs={"skip_url_asserts": True})

    return wrapper


class TestNormalizePathMiddleware:
    @pytest.mark.parametrize(
        "path, status",
        [
            ("/resource1", 200),
            ("/resource1/", 404),
            ("/resource2", 200),
            ("/resource2/", 200),
            ("/resource1?p1=1&p2=2", 200),
            ("/resource1/?p1=1&p2=2", 404),
            ("/resource2?p1=1&p2=2", 200),
            ("/resource2/?p1=1&p2=2", 200),
            ("/resource2/a/b%2Fc", 200),
            ("/resource2/a/b%2Fc/", 200),
        ],
    )
    async def test_add_trailing_when_necessary(
        self, path: str, status: int, cli: CLI
    ) -> None:
        extra_middlewares = [web.normalize_path_middleware(merge_slashes=False)]
        client = await cli(extra_middlewares)

        resp = await client.get(path)
        assert resp.status == status
        assert resp.url.query == URL(path).query

    @pytest.mark.parametrize(
        "path, status",
        [
            ("/resource1", 200),
            ("/resource1/", 200),
            ("/resource2", 404),
            ("/resource2/", 200),
            ("/resource1?p1=1&p2=2", 200),
            ("/resource1/?p1=1&p2=2", 200),
            ("/resource2?p1=1&p2=2", 404),
            ("/resource2/?p1=1&p2=2", 200),
            ("/resource2/a/b%2Fc", 404),
            ("/resource2/a/b%2Fc/", 200),
            ("/resource12", 404),
            ("/resource12345", 404),
        ],
    )
    async def test_remove_trailing_when_necessary(
        self, path: str, status: int, cli: CLI
    ) -> None:
        extra_middlewares = [
            web.normalize_path_middleware(
                append_slash=False, remove_slash=True, merge_slashes=False
            )
        ]
        client = await cli(extra_middlewares)

        resp = await client.get(path)
        assert resp.status == status
        assert resp.url.query == URL(path).query

    @pytest.mark.parametrize(
        "path, status",
        [
            ("/resource1", 200),
            ("/resource1/", 404),
            ("/resource2", 404),
            ("/resource2/", 200),
            ("/resource1?p1=1&p2=2", 200),
            ("/resource1/?p1=1&p2=2", 404),
            ("/resource2?p1=1&p2=2", 404),
            ("/resource2/?p1=1&p2=2", 200),
            ("/resource2/a/b%2Fc", 404),
            ("/resource2/a/b%2Fc/", 200),
        ],
    )
    async def test_no_trailing_slash_when_disabled(
        self, path: str, status: int, cli: CLI
    ) -> None:
        extra_middlewares = [
            web.normalize_path_middleware(append_slash=False, merge_slashes=False)
        ]
        client = await cli(extra_middlewares)

        resp = await client.get(path)
        assert resp.status == status
        assert resp.url.query == URL(path).query

    @pytest.mark.parametrize(
        "path, status",
        [
            ("/resource1/a/b", 200),
            ("//resource1//a//b", 200),
            ("//resource1//a//b/", 404),
            ("///resource1//a//b", 200),
            ("/////resource1/a///b", 200),
            ("/////resource1/a//b/", 404),
            ("/resource1/a/b?p=1", 200),
            ("//resource1//a//b?p=1", 200),
            ("//resource1//a//b/?p=1", 404),
            ("///resource1//a//b?p=1", 200),
            ("/////resource1/a///b?p=1", 200),
            ("/////resource1/a//b/?p=1", 404),
        ],
    )
    async def test_merge_slash(self, path: str, status: int, cli: CLI) -> None:
        extra_middlewares = [web.normalize_path_middleware(append_slash=False)]
        client = await cli(extra_middlewares)

        resp = await client.get(path)
        assert resp.status == status
        assert resp.url.query == URL(path).query

    @pytest.mark.parametrize(
        "path, status",
        [
            ("/resource1/a/b", 200),
            ("/resource1/a/b/", 404),
            ("//resource2//a//b", 200),
            ("//resource2//a//b/", 200),
            ("///resource1//a//b", 200),
            ("///resource1//a//b/", 404),
            ("/////resource1/a///b", 200),
            ("/////resource1/a///b/", 404),
            ("/resource2/a/b", 200),
            ("//resource2//a//b", 200),
            ("//resource2//a//b/", 200),
            ("///resource2//a//b", 200),
            ("///resource2//a//b/", 200),
            ("/////resource2/a///b", 200),
            ("/////resource2/a///b/", 200),
            ("/resource1/a/b?p=1", 200),
            ("/resource1/a/b/?p=1", 404),
            ("//resource2//a//b?p=1", 200),
            ("//resource2//a//b/?p=1", 200),
            ("///resource1//a//b?p=1", 200),
            ("///resource1//a//b/?p=1", 404),
            ("/////resource1/a///b?p=1", 200),
            ("/////resource1/a///b/?p=1", 404),
            ("/resource2/a/b?p=1", 200),
            ("//resource2//a//b?p=1", 200),
            ("//resource2//a//b/?p=1", 200),
            ("///resource2//a//b?p=1", 200),
            ("///resource2//a//b/?p=1", 200),
            ("/////resource2/a///b?p=1", 200),
            ("/////resource2/a///b/?p=1", 200),
        ],
    )
    async def test_append_and_merge_slash(
        self, path: str, status: int, cli: CLI
    ) -> None:
        extra_middlewares = [web.normalize_path_middleware()]

        client = await cli(extra_middlewares)
        resp = await client.get(path)
        assert resp.status == status
        assert resp.url.query == URL(path).query

    @pytest.mark.parametrize(
        "path, status",
        [
            ("/resource1/a/b", 200),
            ("/resource1/a/b/", 200),
            ("//resource2//a//b", 404),
            ("//resource2//a//b/", 200),
            ("///resource1//a//b", 200),
            ("///resource1//a//b/", 200),
            ("/////resource1/a///b", 200),
            ("/////resource1/a///b/", 200),
            ("/////resource1/a///b///", 200),
            ("/resource2/a/b", 404),
            ("//resource2//a//b", 404),
            ("//resource2//a//b/", 200),
            ("///resource2//a//b", 404),
            ("///resource2//a//b/", 200),
            ("/////resource2/a///b", 404),
            ("/////resource2/a///b/", 200),
            ("/resource1/a/b?p=1", 200),
            ("/resource1/a/b/?p=1", 200),
            ("//resource2//a//b?p=1", 404),
            ("//resource2//a//b/?p=1", 200),
            ("///resource1//a//b?p=1", 200),
            ("///resource1//a//b/?p=1", 200),
            ("/////resource1/a///b?p=1", 200),
            ("/////resource1/a///b/?p=1", 200),
            ("/resource2/a/b?p=1", 404),
            ("//resource2//a//b?p=1", 404),
            ("//resource2//a//b/?p=1", 200),
            ("///resource2//a//b?p=1", 404),
            ("///resource2//a//b/?p=1", 200),
            ("/////resource2/a///b?p=1", 404),
            ("/////resource2/a///b/?p=1", 200),
        ],
    )
    async def test_remove_and_merge_slash(
        self, path: str, status: int, cli: CLI
    ) -> None:
        extra_middlewares = [
            web.normalize_path_middleware(append_slash=False, remove_slash=True)
        ]

        client = await cli(extra_middlewares)
        resp = await client.get(path)
        assert resp.status == status
        assert resp.url.query == URL(path).query

    async def test_cannot_remove_and_add_slash(self) -> None:
        with pytest.raises(AssertionError):
            web.normalize_path_middleware(append_slash=True, remove_slash=True)

    @pytest.mark.parametrize(
        ["append_slash", "remove_slash"],
        [
            (True, False),
            (False, True),
            (False, False),
        ],
    )
    async def test_open_redirects(
        self, append_slash: bool, remove_slash: bool, aiohttp_client: AiohttpClient
    ) -> None:
        async def handle(request: web.Request) -> web.StreamResponse:
            pytest.fail(
                "Security advisory 'GHSA-v6wp-4m6f-gcjg' test handler "
                "matched unexpectedly",
                pytrace=False,
            )

        app = web.Application(
            middlewares=[
                web.normalize_path_middleware(
                    append_slash=append_slash, remove_slash=remove_slash
                )
            ]
        )
        app.add_routes([web.get("/", handle), web.get("/google.com", handle)])
        client = await aiohttp_client(app, server_kwargs={"skip_url_asserts": True})
        resp = await client.get("//google.com", allow_redirects=False)
        assert resp.status == 308
        assert resp.headers["Location"] == "/google.com"
        assert resp.url.query == URL("//google.com").query

    async def test_open_redirect_absolute_form_target(
        self, aiohttp_server: AiohttpServer
    ) -> None:
        async def handle(request: web.Request) -> web.Response:
            assert False

        app = web.Application(middlewares=[web.normalize_path_middleware()])
        app.add_routes([web.get("/google.com/", handle)])
        server = await aiohttp_server(app)

        reader, writer = await asyncio.open_connection(server.host, server.port)
        try:
            writer.write(
                b"GET http://google.com/google.com HTTP/1.1\r\n"
                b"Host: localhost\r\nConnection: close\r\n\r\n"
            )
            await writer.drain()
            head = (await reader.readuntil(b"\r\n\r\n")).decode("ascii")
        finally:
            writer.close()
            with suppress(ConnectionResetError, BrokenPipeError):
                await writer.wait_closed()

        assert head.startswith("HTTP/1.1 308 ")
        assert "\r\nLocation: /google.com/\r\n" in head


async def test_normalize_path_skips_parser_error(
    aiohttp_server: AiohttpServer,
) -> None:
    # Regression: normalize_path_middleware used to redirect SystemRoute
    # matches indiscriminately. After parser errors started flowing through
    # the middleware chain (issue #3287) the dummy "/" rel_url on a
    # malformed request would match a real "/" route and trigger 301.
    async def root(request: web.Request) -> NoReturn:
        pytest.fail("root handler should not run for a parser-errored request")

    app = web.Application(middlewares=[web.normalize_path_middleware()])
    app.router.add_get("/", root)
    server = await aiohttp_server(app)

    reader, writer = await asyncio.open_connection(server.host, server.port)
    try:
        writer.write(b"garbage\r\n\r\n")
        await writer.drain()
        head = b""
        while b"\r\n\r\n" not in head:
            chunk = await reader.read(4096)
            assert chunk, "connection closed before headers complete"
            head += chunk
    finally:
        writer.close()
        await writer.wait_closed()

    assert head.startswith(b"HTTP/1.0 400 ")


async def test_normalize_path_redirects_method_not_allowed(
    aiohttp_client: AiohttpClient,
) -> None:
    # A 405 is a routing failure (the path matched a resource, but not for this
    # method), so normalize_path_middleware should still try alternate paths:
    # POST /foo has no POST route, but POST /foo/ does.
    async def handler(request: web.Request) -> web.Response:
        assert False

    app = web.Application(middlewares=[web.normalize_path_middleware()])
    app.router.add_get("/foo", handler)
    app.router.add_post("/foo/", handler)
    client = await aiohttp_client(app)

    resp = await client.post("/foo", allow_redirects=False)
    assert resp.status == 308
    assert resp.headers["Location"] == "/foo/"


async def test_bug_3669(aiohttp_client: AiohttpClient) -> None:
    async def paymethod(request: web.Request) -> NoReturn:
        assert False

    app = web.Application()
    app.router.add_route("GET", "/paymethod", paymethod)
    app.middlewares.append(
        web.normalize_path_middleware(append_slash=False, remove_slash=True)
    )

    client = await aiohttp_client(app, server_kwargs={"skip_url_asserts": True})

    resp = await client.get("/paymethods")
    assert resp.status == 404
    assert resp.url.path != "/paymethod"


async def test_old_style_middleware(aiohttp_client: AiohttpClient) -> None:
    async def view_handler(request: web.Request) -> web.Response:
        return web.Response(body=b"OK")

    with pytest.deprecated_call(
        match=r"^Middleware decorator is deprecated since 4\.0 and its "
        r"behaviour is default, you can simply remove this decorator\.$",
    ):

        @web.middleware
        async def middleware(request: web.Request, handler: Handler) -> web.Response:
            resp = await handler(request)
            assert 200 == resp.status
            resp.set_status(201)
            assert isinstance(resp, web.Response)
            assert resp.text is not None
            resp.text = resp.text + "[old style middleware]"
            return resp

    app = web.Application(middlewares=[middleware])
    app.router.add_route("GET", "/", view_handler)
    client = await aiohttp_client(app)
    resp = await client.get("/")
    assert 201 == resp.status
    txt = await resp.text()
    assert "OK[old style middleware]" == txt


async def test_new_style_middleware_class(aiohttp_client: AiohttpClient) -> None:
    async def handler(request: web.Request) -> web.Response:
        return web.Response(body=b"OK")

    class Middleware:
        async def __call__(
            self, request: web.Request, handler: Handler
        ) -> web.Response:
            resp = await handler(request)
            assert 200 == resp.status
            resp.set_status(201)
            assert isinstance(resp, web.Response)
            assert resp.text is not None
            resp.text = resp.text + "[new style middleware]"
            return resp

    app = web.Application()
    app.middlewares.append(Middleware())
    app.router.add_route("GET", "/", handler)
    client = await aiohttp_client(app)
    resp = await client.get("/")
    assert 201 == resp.status
    txt = await resp.text()
    assert "OK[new style middleware]" == txt


async def test_new_style_middleware_method(aiohttp_client: AiohttpClient) -> None:
    async def handler(request: web.Request) -> web.Response:
        return web.Response(body=b"OK")

    class Middleware:
        async def call(self, request: web.Request, handler: Handler) -> web.Response:
            resp = await handler(request)
            assert 200 == resp.status
            resp.set_status(201)
            assert isinstance(resp, web.Response)
            assert resp.text is not None
            resp.text = resp.text + "[new style middleware]"
            return resp

    app = web.Application()
    app.middlewares.append(Middleware().call)
    app.router.add_route("GET", "/", handler)
    client = await aiohttp_client(app)
    resp = await client.get("/")
    assert 201 == resp.status
    txt = await resp.text()
    assert "OK[new style middleware]" == txt


async def test_middleware_does_not_leak(aiohttp_client: AiohttpClient) -> None:
    async def any_handler(request: web.Request) -> NoReturn:
        assert False

    class Middleware:
        async def call(
            self, request: web.Request, handler: Handler
        ) -> web.StreamResponse:
            return await handler(request)

    app = web.Application()
    app.router.add_route("POST", "/any", any_handler)
    app.middlewares.append(Middleware().call)

    client = await aiohttp_client(app)

    web_app._cached_build_middleware.cache_clear()
    for _ in range(10):
        resp = await client.get("/any")
        assert resp.status == 405
    assert web_app._cached_build_middleware.cache_info().currsize < 10
