import asyncio
import logging
import sys
from collections.abc import AsyncIterator, Callable, Iterator
from contextlib import asynccontextmanager
from typing import NoReturn
from unittest import mock

import pytest
from pytest_aiohttp import AiohttpClient, AiohttpServer

from aiohttp import hdrs, log, web
from aiohttp.http_exceptions import HttpProcessingError
from aiohttp.typedefs import Handler


async def test_app_ctor() -> None:
    app = web.Application()
    assert app.logger is log.web_logger


def test_app_call() -> None:
    app = web.Application()
    assert app is app()


async def test_app_register_on_finish() -> None:
    app = web.Application()
    cb1 = mock.AsyncMock(return_value=None)
    cb2 = mock.AsyncMock(return_value=None)
    app.on_cleanup.append(cb1)
    app.on_cleanup.append(cb2)
    app.freeze()
    await app.cleanup()
    cb1.assert_called_once_with(app)
    cb2.assert_called_once_with(app)


async def test_app_register_coro() -> None:
    app = web.Application()
    fut = asyncio.get_running_loop().create_future()

    async def cb(app: web.Application) -> None:
        await asyncio.sleep(0.001)
        fut.set_result(123)

    app.on_cleanup.append(cb)
    app.freeze()
    await app.cleanup()
    assert fut.done()
    assert 123 == fut.result()


def test_logging() -> None:
    logger = mock.Mock()
    app = web.Application()
    app.logger = logger
    assert app.logger is logger


async def test_on_shutdown() -> None:
    app = web.Application()
    called = False

    async def on_shutdown(app_param: web.Application) -> None:
        nonlocal called
        assert app is app_param
        called = True

    app.on_shutdown.append(on_shutdown)
    app.freeze()
    await app.shutdown()
    assert called


async def test_on_startup() -> None:
    app = web.Application()

    long_running1_called = False
    long_running2_called = False
    all_long_running_called = False

    async def long_running1(app_param: web.Application) -> None:
        nonlocal long_running1_called
        assert app is app_param
        long_running1_called = True

    async def long_running2(app_param: web.Application) -> None:
        nonlocal long_running2_called
        assert app is app_param
        long_running2_called = True

    async def on_startup_all_long_running(app_param: web.Application) -> None:
        nonlocal all_long_running_called
        assert app is app_param
        all_long_running_called = True
        await asyncio.gather(long_running1(app_param), long_running2(app_param))

    app.on_startup.append(on_startup_all_long_running)
    app.freeze()

    await app.startup()
    assert long_running1_called
    assert long_running2_called
    assert all_long_running_called


def test_appkey() -> None:
    key = web.AppKey("key", str)
    app = web.Application()
    app[key] = "value"
    assert app[key] == "value"
    assert len(app) == 1
    del app[key]
    assert len(app) == 0


def test_appkey_repr_concrete() -> None:
    key = web.AppKey("key", int)
    assert repr(key) in (
        "<AppKey(__channelexec__.key, type=int)>",  # pytest-xdist
        "<AppKey(__main__.key, type=int)>",
    )
    key2 = web.AppKey("key", web.Request)
    assert repr(key2) in (
        # pytest-xdist:
        "<AppKey(__channelexec__.key, type=aiohttp.web_request.Request)>",
        "<AppKey(__main__.key, type=aiohttp.web_request.Request)>",
    )


def test_appkey_repr_nonconcrete() -> None:
    key = web.AppKey("key", Iterator[int])
    if sys.version_info < (3, 11):
        assert repr(key) in (
            # pytest-xdist:
            "<AppKey(__channelexec__.key, type=collections.abc.Iterator)>",
            "<AppKey(__main__.key, type=collections.abc.Iterator)>",
        )
    else:
        assert repr(key) in (
            # pytest-xdist:
            "<AppKey(__channelexec__.key, type=collections.abc.Iterator[int])>",
            "<AppKey(__main__.key, type=collections.abc.Iterator[int])>",
        )


def test_appkey_repr_annotated() -> None:
    key = web.AppKey[Iterator[int]]("key")
    if sys.version_info < (3, 11):
        assert repr(key) in (
            # pytest-xdist:
            "<AppKey(__channelexec__.key, type=collections.abc.Iterator)>",
            "<AppKey(__main__.key, type=collections.abc.Iterator)>",
        )
    else:
        assert repr(key) in (
            # pytest-xdist:
            "<AppKey(__channelexec__.key, type=collections.abc.Iterator[int])>",
            "<AppKey(__main__.key, type=collections.abc.Iterator[int])>",
        )


def test_app_str_keys() -> None:
    app = web.Application()
    with pytest.warns(
        UserWarning, match=r"web_advanced\.html#application-s-config"
    ) as checker:
        app["key"] = "value"
        # Check that the error is emitted at the call site (stacklevel=2)
        assert checker[0].filename == __file__
    assert app["key"] == "value"


def test_app_get() -> None:
    key = web.AppKey("key", int)
    app = web.Application()
    assert app.get(key, "foo") == "foo"
    app[key] = 5
    assert app.get(key, "foo") == 5


def test_app_freeze() -> None:
    app = web.Application()
    subapp = mock.Mock()
    subapp._middlewares = ()
    app._subapps.append(subapp)

    app.freeze()
    assert subapp.freeze.called

    app.freeze()
    assert len(subapp.freeze.call_args_list) == 1


def test_equality() -> None:
    app1 = web.Application()
    app2 = web.Application()

    assert app1 == app1
    assert app1 != app2


def test_app_run_middlewares() -> None:
    root = web.Application()
    sub = web.Application()
    root.add_subapp("/sub", sub)
    root.freeze()
    assert root._run_middlewares is False

    async def middleware(request: web.Request, handler: Handler) -> web.StreamResponse:
        assert False

    root = web.Application(middlewares=[middleware])
    sub = web.Application()
    root.add_subapp("/sub", sub)
    root.freeze()
    assert root._run_middlewares is True

    root = web.Application()
    sub = web.Application(middlewares=[middleware])
    root.add_subapp("/sub", sub)
    root.freeze()
    assert root._run_middlewares is True


def test_subapp_pre_frozen_after_adding() -> None:
    app = web.Application()
    subapp = web.Application()

    app.add_subapp("/prefix", subapp)
    assert subapp.pre_frozen
    assert not subapp.frozen


def test_app_inheritance() -> None:
    with pytest.raises(TypeError):

        class A(web.Application):  # type: ignore[misc]
            pass


def test_app_custom_attr() -> None:
    app = web.Application()
    with pytest.raises(AttributeError):
        app.custom = None  # type: ignore[attr-defined]


async def test_cleanup_ctx() -> None:
    app = web.Application()
    out = []

    def f(num: int) -> Callable[[web.Application], AsyncIterator[None]]:
        async def inner(app: web.Application) -> AsyncIterator[None]:
            out.append("pre_" + str(num))
            yield None
            out.append("post_" + str(num))

        return inner

    app.cleanup_ctx.append(f(1))
    app.cleanup_ctx.append(f(2))
    app.freeze()
    await app.startup()
    assert out == ["pre_1", "pre_2"]
    await app.cleanup()
    assert out == ["pre_1", "pre_2", "post_2", "post_1"]


async def test_cleanup_ctx_exception_on_startup() -> None:
    app = web.Application()
    out = []

    exc = Exception("fail")

    def f(
        num: int, fail: bool = False
    ) -> Callable[[web.Application], AsyncIterator[None]]:
        async def inner(app: web.Application) -> AsyncIterator[None]:
            out.append("pre_" + str(num))
            if fail:
                raise exc
            yield None
            out.append("post_" + str(num))

        return inner

    app.cleanup_ctx.append(f(1))
    app.cleanup_ctx.append(f(2, True))
    app.cleanup_ctx.append(f(3))
    app.freeze()
    with pytest.raises(Exception) as ctx:
        await app.startup()
    assert ctx.value is exc
    assert out == ["pre_1", "pre_2"]
    await app.cleanup()
    assert out == ["pre_1", "pre_2", "post_1"]


async def test_cleanup_ctx_exception_on_cleanup() -> None:
    app = web.Application()
    out = []

    exc = Exception("fail")

    def f(
        num: int, fail: bool = False
    ) -> Callable[[web.Application], AsyncIterator[None]]:
        async def inner(app: web.Application) -> AsyncIterator[None]:
            out.append("pre_" + str(num))
            yield None
            out.append("post_" + str(num))
            if fail:
                raise exc

        return inner

    app.cleanup_ctx.append(f(1))
    app.cleanup_ctx.append(f(2, True))
    app.cleanup_ctx.append(f(3))
    app.freeze()
    await app.startup()
    assert out == ["pre_1", "pre_2", "pre_3"]
    with pytest.raises(Exception) as ctx:
        await app.cleanup()
    assert ctx.value is exc
    assert out == ["pre_1", "pre_2", "pre_3", "post_3", "post_2", "post_1"]


async def test_cleanup_ctx_cleanup_after_exception() -> None:
    app = web.Application()
    ctx_state = None

    async def success_ctx(app: web.Application) -> AsyncIterator[None]:
        nonlocal ctx_state
        ctx_state = "START"
        yield
        ctx_state = "CLEAN"

    async def fail_ctx(app: web.Application) -> AsyncIterator[NoReturn]:
        raise Exception()
        yield  # type: ignore[unreachable]  # pragma: no cover

    app.cleanup_ctx.append(success_ctx)
    app.cleanup_ctx.append(fail_ctx)
    runner = web.AppRunner(app)
    try:
        with pytest.raises(Exception):
            await runner.setup()
    finally:
        await runner.cleanup()

    assert ctx_state == "CLEAN"


@pytest.mark.parametrize("exc_cls", (Exception, asyncio.CancelledError))
async def test_cleanup_ctx_exception_on_cleanup_multiple(
    exc_cls: type[BaseException],
) -> None:
    app = web.Application()
    out = []

    def f(
        num: int, fail: bool = False
    ) -> Callable[[web.Application], AsyncIterator[None]]:
        async def inner(app: web.Application) -> AsyncIterator[None]:
            out.append("pre_" + str(num))
            yield None
            out.append("post_" + str(num))
            if fail:
                raise exc_cls("fail_" + str(num))

        return inner

    app.cleanup_ctx.append(f(1))
    app.cleanup_ctx.append(f(2, True))
    app.cleanup_ctx.append(f(3, True))
    app.freeze()
    await app.startup()
    assert out == ["pre_1", "pre_2", "pre_3"]
    with pytest.raises(web.CleanupError) as ctx:
        await app.cleanup()
    exc = ctx.value
    assert len(exc.exceptions) == 2
    assert str(exc.exceptions[0]) == "fail_3"
    assert str(exc.exceptions[1]) == "fail_2"
    assert out == ["pre_1", "pre_2", "pre_3", "post_3", "post_2", "post_1"]


async def test_cleanup_ctx_multiple_yields() -> None:
    app = web.Application()
    out = []

    def f(num: int) -> Callable[[web.Application], AsyncIterator[None]]:
        async def inner(app: web.Application) -> AsyncIterator[None]:
            out.append("pre_" + str(num))
            yield None
            out.append("post_" + str(num))
            yield None

        return inner

    app.cleanup_ctx.append(f(1))
    app.freeze()
    await app.startup()
    assert out == ["pre_1"]
    with pytest.raises(RuntimeError):
        await app.cleanup()
    assert out == ["pre_1", "post_1"]


async def test_cleanup_ctx_with_async_generator_and_asynccontextmanager() -> None:
    entered = []

    async def gen_ctx(app: web.Application) -> AsyncIterator[None]:
        entered.append("enter-gen")
        try:
            yield
        finally:
            entered.append("exit-gen")

    @asynccontextmanager
    async def cm_ctx(app: web.Application) -> AsyncIterator[None]:
        entered.append("enter-cm")
        try:
            yield
        finally:
            entered.append("exit-cm")

    app = web.Application()
    app.cleanup_ctx.append(gen_ctx)
    app.cleanup_ctx.append(cm_ctx)
    app.freeze()
    await app.startup()
    assert "enter-gen" in entered and "enter-cm" in entered
    await app.cleanup()
    assert "exit-gen" in entered and "exit-cm" in entered


async def test_cleanup_ctx_exception_in_cm_exit() -> None:
    app = web.Application()

    exc = RuntimeError("exit failed")

    @asynccontextmanager
    async def failing_exit_ctx(app: web.Application) -> AsyncIterator[None]:
        yield
        raise exc

    app.cleanup_ctx.append(failing_exit_ctx)
    app.freeze()
    await app.startup()
    with pytest.raises(RuntimeError) as ctx:
        await app.cleanup()
    assert ctx.value is exc


async def test_cleanup_ctx_mixed_with_exception_in_cm_exit() -> None:
    app = web.Application()
    out = []

    async def working_gen(app: web.Application) -> AsyncIterator[None]:
        out.append("pre_gen")
        yield
        out.append("post_gen")

    exc = RuntimeError("cm exit failed")

    @asynccontextmanager
    async def failing_exit_cm(app: web.Application) -> AsyncIterator[None]:
        out.append("pre_cm")
        yield
        out.append("post_cm")
        raise exc

    app.cleanup_ctx.append(working_gen)
    app.cleanup_ctx.append(failing_exit_cm)
    app.freeze()
    await app.startup()
    with pytest.raises(RuntimeError) as ctx:
        await app.cleanup()
    assert ctx.value is exc
    assert out == ["pre_gen", "pre_cm", "post_cm", "post_gen"]


async def test_subapp_chained_config_dict_visibility(
    aiohttp_client: AiohttpClient,
) -> None:
    key1 = web.AppKey("key1", str)
    key2 = web.AppKey("key2", str)

    async def main_handler(request: web.Request) -> web.Response:
        assert request.config_dict[key1] == "val1"
        assert key2 not in request.config_dict
        return web.Response(status=200)

    root = web.Application()
    root[key1] = "val1"
    root.add_routes([web.get("/", main_handler)])

    async def sub_handler(request: web.Request) -> web.Response:
        assert request.config_dict[key1] == "val1"
        assert request.config_dict[key2] == "val2"
        return web.Response(status=201)

    sub = web.Application()
    sub[key2] = "val2"
    sub.add_routes([web.get("/", sub_handler)])
    root.add_subapp("/sub", sub)

    client = await aiohttp_client(root)

    resp = await client.get("/")
    assert resp.status == 200
    resp = await client.get("/sub/")
    assert resp.status == 201


async def test_subapp_chained_config_dict_overriding(
    aiohttp_client: AiohttpClient,
) -> None:
    key = web.AppKey("key", str)

    async def main_handler(request: web.Request) -> web.Response:
        assert request.config_dict[key] == "val1"
        return web.Response(status=200)

    root = web.Application()
    root[key] = "val1"
    root.add_routes([web.get("/", main_handler)])

    async def sub_handler(request: web.Request) -> web.Response:
        assert request.config_dict[key] == "val2"
        return web.Response(status=201)

    sub = web.Application()
    sub[key] = "val2"
    sub.add_routes([web.get("/", sub_handler)])
    root.add_subapp("/sub", sub)

    client = await aiohttp_client(root)

    resp = await client.get("/")
    assert resp.status == 200
    resp = await client.get("/sub/")
    assert resp.status == 201


async def test_subapp_on_startup(aiohttp_client: AiohttpClient) -> None:
    subapp = web.Application()
    startup = web.AppKey("startup", bool)
    cleanup = web.AppKey("cleanup", bool)

    startup_called = False

    async def on_startup(app: web.Application) -> None:
        nonlocal startup_called
        startup_called = True
        app[startup] = True

    subapp.on_startup.append(on_startup)

    ctx_pre_called = False
    ctx_post_called = False

    async def cleanup_ctx(app: web.Application) -> AsyncIterator[None]:
        nonlocal ctx_pre_called, ctx_post_called
        ctx_pre_called = True
        app[cleanup] = True
        yield None
        ctx_post_called = True

    subapp.cleanup_ctx.append(cleanup_ctx)

    shutdown_called = False

    async def on_shutdown(app: web.Application) -> None:
        nonlocal shutdown_called
        shutdown_called = True

    subapp.on_shutdown.append(on_shutdown)

    cleanup_called = False

    async def on_cleanup(app: web.Application) -> None:
        nonlocal cleanup_called
        cleanup_called = True

    subapp.on_cleanup.append(on_cleanup)

    app = web.Application()

    app.add_subapp("/subapp", subapp)

    assert not startup_called
    assert not ctx_pre_called
    assert not ctx_post_called
    assert not shutdown_called
    assert not cleanup_called

    assert subapp.on_startup.frozen
    assert subapp.cleanup_ctx.frozen
    assert subapp.on_shutdown.frozen
    assert subapp.on_cleanup.frozen
    assert subapp.router.frozen

    client = await aiohttp_client(app)

    assert startup_called
    assert ctx_pre_called  # type: ignore[unreachable]
    assert not ctx_post_called
    assert not shutdown_called
    assert not cleanup_called

    await client.close()

    assert startup_called
    assert ctx_pre_called
    assert ctx_post_called
    assert shutdown_called
    assert cleanup_called


@pytest.mark.filterwarnings(r"ignore:.*web\.AppKey:UserWarning")
def test_app_iter() -> None:
    app = web.Application()
    b = web.AppKey("b", str)
    c = web.AppKey("c", str)
    app["a"] = "0"
    app[b] = "1"
    app[c] = "2"
    app["d"] = "4"
    assert sorted(list(app)) == [b, c, "a", "d"]


def test_app_forbid_nonslot_attr() -> None:
    app = web.Application()
    with pytest.raises(AttributeError):
        app.unknow_attr  # type: ignore[attr-defined]
    with pytest.raises(AttributeError):
        app.unknow_attr = 1  # type: ignore[attr-defined]


def test_forbid_changing_frozen_app() -> None:
    app = web.Application()
    app.freeze()
    with pytest.raises(RuntimeError):
        app["key"] = "value"


def test_app_boolean() -> None:
    app = web.Application()
    assert app


async def _read_response(
    reader: asyncio.StreamReader,
) -> tuple[bytes, dict[str, str], bytes]:
    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
    header_block, _, body = head.partition(b"\r\n\r\n")
    status_line, *header_lines = header_block.split(b"\r\n")
    headers = {}
    for line in header_lines:
        name, _, value = line.partition(b":")
        headers[name.decode().strip().lower()] = value.decode().strip()
    return status_line, headers, body


@asynccontextmanager
async def _raw_request(
    host: str, port: int, payload: bytes
) -> AsyncIterator[asyncio.StreamReader]:
    reader, writer = await asyncio.open_connection(host, port)
    try:
        writer.write(payload)
        await writer.drain()
        yield reader
    finally:
        writer.close()
        await writer.wait_closed()


async def test_parser_error_passes_through_middleware(
    aiohttp_server: AiohttpServer,
) -> None:
    async def server_header(
        request: web.Request, handler: Handler
    ) -> web.StreamResponse:
        try:
            return await handler(request)
        except web.HTTPException as exc:
            exc.headers[hdrs.SERVER] = "custom"
            raise

    app = web.Application(middlewares=[server_header])
    server = await aiohttp_server(app)

    async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader:
        status_line, headers, _ = await _read_response(reader)
        assert await reader.read() == b""

    assert status_line.startswith(b"HTTP/1.0 400 ")
    assert headers["server"] == "custom"


async def test_parser_error_middleware_can_catch(
    aiohttp_server: AiohttpServer,
) -> None:
    captured: list[BaseException | None] = []

    async def catcher(request: web.Request, handler: Handler) -> web.StreamResponse:
        try:
            return await handler(request)
        except web.HTTPBadRequest as exc:
            captured.append(exc.__cause__)
            return web.Response(status=418, text="teapot")

    app = web.Application(middlewares=[catcher])
    server = await aiohttp_server(app)

    async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader:
        status_line, _, body = await _read_response(reader)
        body += await reader.read()

    assert status_line.startswith(b"HTTP/1.0 418 ")
    assert body.endswith(b"teapot")
    assert len(captured) == 1
    assert isinstance(captured[0], HttpProcessingError)


async def test_parser_error_match_info_exposes_http_exception(
    aiohttp_server: AiohttpServer,
) -> None:
    seen: list[web.HTTPException | None] = []

    async def inspector(request: web.Request, handler: Handler) -> web.StreamResponse:
        seen.append(request.match_info.http_exception)
        return await handler(request)

    app = web.Application(middlewares=[inspector])
    server = await aiohttp_server(app)

    async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader:
        await _read_response(reader)

    assert len(seen) == 1
    assert isinstance(seen[0], web.HTTPBadRequest)


async def test_parser_error_middleware_suppresses_log(
    aiohttp_server: AiohttpServer,
) -> None:
    async def swallow(request: web.Request, handler: Handler) -> web.StreamResponse:
        try:
            return await handler(request)
        except web.HTTPBadRequest:
            return web.Response(status=204)

    logger = mock.create_autospec(logging.Logger, spec_set=True, instance=True)
    app = web.Application(middlewares=[swallow])
    server = await aiohttp_server(app, logger=logger)

    async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader:
        status_line, _, _ = await _read_response(reader)

    assert status_line.startswith(b"HTTP/1.0 204 ")
    logger.exception.assert_not_called()
    logger.debug.assert_not_called()
    logger.warning.assert_not_called()


async def test_parser_error_propagated_is_logged(
    aiohttp_server: AiohttpServer,
) -> None:
    logger = mock.create_autospec(logging.Logger, spec_set=True, instance=True)
    app = web.Application()
    server = await aiohttp_server(app, logger=logger)

    async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader:
        await _read_response(reader)

    # garbage triggers BadHttpMethod; logged at warning level by default.
    assert logger.warning.called
    assert "Error handling request" in logger.warning.call_args.args[0]


async def test_parser_error_on_response_prepare_fires(
    aiohttp_server: AiohttpServer,
) -> None:
    fired = False

    async def hook(request: web.Request, response: web.StreamResponse) -> None:
        nonlocal fired
        fired = True
        response.headers[hdrs.SERVER] = "from-signal"

    app = web.Application()
    app.on_response_prepare.append(hook)
    server = await aiohttp_server(app)

    async with _raw_request(server.host, server.port, b"garbage\r\n\r\n") as reader:
        status_line, headers, _ = await _read_response(reader)

    assert status_line.startswith(b"HTTP/1.0 400 ")
    assert fired
    assert headers["server"] == "from-signal"
