import datetime
import logging
import platform
import sys
from contextvars import ContextVar
from typing import NoReturn
from unittest import mock

import pytest
from pytest_aiohttp import AiohttpClient, AiohttpRawServer, AiohttpServer

import aiohttp
from aiohttp import web
from aiohttp.abc import AbstractAccessLogger, AbstractAsyncAccessLogger
from aiohttp.test_utils import make_mocked_request
from aiohttp.typedefs import Handler
from aiohttp.web_log import AccessLogger
from aiohttp.web_response import Response

if sys.version_info >= (3, 11):
    from typing import Self
else:
    from typing import Any as Self

IS_PYPY = platform.python_implementation() == "PyPy"


def test_access_logger_format() -> None:
    log_format = '%T "%{ETag}o" %X {X} %%P'
    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, log_format)
    expected = '%s "%s" %%X {X} %%%s'
    assert expected == access_logger._log_format


@pytest.mark.skipif(
    IS_PYPY,
    reason="""
    Because of patching :py:class:`datetime.datetime`, under PyPy it
    fails in :py:func:`isinstance` call in
    :py:meth:`datetime.datetime.__sub__` (called from
    :py:meth:`aiohttp.AccessLogger._format_t`):

    *** TypeError: isinstance() arg 2 must be a class, type, or tuple of classes and types

    (Pdb) from datetime import datetime
    (Pdb) isinstance(now, datetime)
    *** TypeError: isinstance() arg 2 must be a class, type, or tuple of classes and types
    (Pdb) datetime.__class__
    <class 'unittest.mock.MagicMock'>
    (Pdb) isinstance(now, datetime.__class__)
    False

    Ref: https://bitbucket.org/pypy/pypy/issues/1187/call-to-isinstance-in-__sub__-self-other
    Ref: https://github.com/celery/celery/issues/811
    Ref: https://stackoverflow.com/a/46102240/595220
    """,
)
@pytest.mark.parametrize(
    "log_format,expected,extra",
    [
        (
            "%t",
            "[01/Jan/1843:00:29:56 +0800]",
            {"request_start_time": "[01/Jan/1843:00:29:56 +0800]"},
        ),
        (
            '%a %t %P %r %s %b %T %Tf %D "%{H1}i" "%{H2}i"',
            (
                "127.0.0.2 [01/Jan/1843:00:29:56 +0800] <42> "
                'GET /path HTTP/1.1 200 42 3 3.141593 3141593 "a" "b"'
            ),
            {
                "first_request_line": "GET /path HTTP/1.1",
                "process_id": "<42>",
                "remote_address": "127.0.0.2",
                "request_start_time": "[01/Jan/1843:00:29:56 +0800]",
                "request_time": "3",
                "request_time_frac": "3.141593",
                "request_time_micro": "3141593",
                "response_size": 42,
                "response_status": 200,
                "request_header": {"H1": "a", "H2": "b"},
            },
        ),
    ],
)
def test_access_logger_atoms(
    monkeypatch: pytest.MonkeyPatch,
    log_format: str,
    expected: str,
    extra: dict[str, object],
) -> None:
    class PatchedDatetime(datetime.datetime):
        @classmethod
        def now(cls, tz: datetime.tzinfo | None = None) -> Self:
            assert tz is not None
            # Simulate: real UTC time is 1842-12-31 16:30, convert to local tz
            utc = datetime.datetime(1842, 12, 31, 16, 30, tzinfo=datetime.timezone.utc)
            local = utc.astimezone(tz)
            return cls(
                local.year,
                local.month,
                local.day,
                local.hour,
                local.minute,
                local.second,
                tzinfo=tz,
            )

    monkeypatch.setattr("datetime.datetime", PatchedDatetime)
    # Mock localtime to return CST (+0800 = 28800 seconds)
    mock_localtime = mock.Mock()
    mock_localtime.return_value.tm_gmtoff = 28800
    monkeypatch.setattr("aiohttp.web_log.time_mod.localtime", mock_localtime)
    # Clear cached timezone so it gets rebuilt with our mock
    AccessLogger._cached_tz = None
    AccessLogger._cached_tz_expires = 0.0
    monkeypatch.setattr("os.getpid", lambda: 42)
    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, log_format)
    request = mock.Mock(
        headers={"H1": "a", "H2": "b"},
        method="GET",
        path_qs="/path",
        version=aiohttp.HttpVersion(1, 1),
        remote="127.0.0.2",
    )
    response = mock.Mock(headers={}, body_length=42, status=200)
    access_logger.log(request, response, 3.1415926)
    assert not mock_logger.exception.called, mock_logger.exception.call_args

    mock_logger.info.assert_called_with(expected, extra=extra)


@pytest.mark.skipif(
    IS_PYPY,
    reason="PyPy has issues with patching datetime.datetime",
)
def test_access_logger_dst_timezone(monkeypatch: pytest.MonkeyPatch) -> None:
    """Test that _format_t uses the current local UTC offset, not a cached one.

    This ensures timestamps are correct during DST transitions. The old
    implementation used time.timezone which is a constant and doesn't
    reflect DST changes.
    """
    # Simulate a timezone that observes DST (e.g., US Eastern)
    # During EST: UTC-5 (-18000s), during EDT: UTC-4 (-14400s)
    gmtoff_est = -18000  # UTC-5
    gmtoff_edt = -14400  # UTC-4

    class PatchedDatetime(datetime.datetime):
        @classmethod
        def now(cls, tz: datetime.tzinfo | None = None) -> Self:
            assert tz is not None
            # Simulate: real UTC time is 07:00, convert to local tz
            utc = datetime.datetime(2024, 3, 10, 7, 0, 0, tzinfo=datetime.timezone.utc)
            local = utc.astimezone(tz)
            return cls(
                local.year,
                local.month,
                local.day,
                local.hour,
                local.minute,
                local.second,
                tzinfo=tz,
            )

    monkeypatch.setattr("datetime.datetime", PatchedDatetime)
    mock_localtime = mock.Mock()
    mock_localtime.return_value.tm_gmtoff = gmtoff_est
    monkeypatch.setattr("aiohttp.web_log.time_mod.localtime", mock_localtime)
    # Force cache refresh
    AccessLogger._cached_tz = None
    AccessLogger._cached_tz_expires = 0.0

    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, "%t")
    request = mock.Mock(
        headers={}, method="GET", path_qs="/", version=(1, 1), remote="127.0.0.1"
    )
    response = mock.Mock(headers={}, body_length=0, status=200)

    # During EST (UTC-5): time is 07:00-05:00 = 02:00 EST
    access_logger.log(request, response, 0.0)
    call1 = mock_logger.info.call_args[0][0]
    assert "-0500" in call1, f"Expected EST offset in {call1}"

    mock_logger.reset_mock()

    # Switch to EDT (UTC-4): force cache invalidation
    mock_localtime.return_value.tm_gmtoff = gmtoff_edt
    AccessLogger._cached_tz = None
    AccessLogger._cached_tz_expires = 0.0
    access_logger.log(request, response, 0.0)
    call2 = mock_logger.info.call_args[0][0]
    assert "-0400" in call2, f"Expected EDT offset in {call2}"

    # Verify the hour changed too (02:00 -> 03:00)
    assert "02:00:00 -0500" in call1
    assert "03:00:00 -0400" in call2

    # Verify cached tz works too
    assert access_logger._cached_tz is not None
    with mock.patch(
        "aiohttp.web_log.time_mod.time",
        return_value=access_logger._cached_tz_expires - 1,
    ):
        access_logger.log(request, response, 0.0)
    call3 = mock_logger.info.call_args[0][0]
    assert "-0400" in call3, f"Expected EDT offset in {call3}"


def test_access_logger_dicts() -> None:
    log_format = "%{User-Agent}i %{Content-Length}o %{None}i"
    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, log_format)
    request = mock.Mock(
        headers={"User-Agent": "Mock/1.0"}, version=(1, 1), remote="127.0.0.2"
    )
    response = mock.Mock(headers={"Content-Length": 123})
    access_logger.log(request, response, 0.0)
    assert not mock_logger.error.called
    expected = "Mock/1.0 123 -"
    extra = {
        "request_header": {"User-Agent": "Mock/1.0", "None": "-"},
        "response_header": {"Content-Length": 123},
    }

    mock_logger.info.assert_called_with(expected, extra=extra)


def test_access_logger_unix_socket() -> None:
    log_format = "|%a|"
    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, log_format)
    request = mock.Mock(headers={"User-Agent": "Mock/1.0"}, version=(1, 1), remote="")
    response = mock.Mock()
    access_logger.log(request, response, 0.0)
    assert not mock_logger.error.called
    expected = "||"
    mock_logger.info.assert_called_with(expected, extra={"remote_address": ""})


def test_logger_no_message() -> None:
    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, "%r %{content-type}i")
    extra_dict = {
        "first_request_line": "GET / HTTP/1.1",
        "request_header": {"content-type": "-"},
    }

    access_logger.log(make_mocked_request("GET", "/"), web.Response(), 0.0)
    mock_logger.info.assert_called_with("GET / HTTP/1.1 -", extra=extra_dict)


def test_logger_internal_error() -> None:
    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, "%D")
    access_logger.log(make_mocked_request("GET", "/"), web.Response(), "invalid")  # type: ignore[arg-type]
    mock_logger.exception.assert_called_with("Error in logging")


def test_logger_no_transport() -> None:
    mock_logger = mock.Mock()
    access_logger = AccessLogger(mock_logger, "%a")
    access_logger.log(make_mocked_request("GET", "/"), web.Response(), 0.0)
    mock_logger.info.assert_called_with("-", extra={"remote_address": "-"})


def test_logger_abc() -> None:
    class Logger(AbstractAccessLogger):
        def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            1 / 0

    mock_logger = mock.Mock()
    access_logger: AbstractAccessLogger = Logger(mock_logger, "")

    with pytest.raises(ZeroDivisionError):
        access_logger.log(make_mocked_request("GET", "/"), web.Response(), 0.0)

    class Logger2(AbstractAccessLogger):
        def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            self.logger.info(
                self.log_format.format(request=request, response=response, time=time)
            )

    mock_logger = mock.Mock()
    access_logger = Logger2(mock_logger, "{request} {response} {time}")
    access_logger.log("request", "response", 1)  # type: ignore[arg-type]
    mock_logger.info.assert_called_with("request response 1")


async def test_exc_info_context(
    aiohttp_raw_server: AiohttpRawServer, aiohttp_client: AiohttpClient
) -> None:
    exc_msg = None

    class Logger(AbstractAccessLogger):
        def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            nonlocal exc_msg
            exc_msg = "{0.__name__}: {1}".format(*sys.exc_info())

    async def handler(request: web.BaseRequest) -> NoReturn:
        raise RuntimeError("intentional runtime error")

    logger = mock.Mock()
    server = await aiohttp_raw_server(handler, access_log_class=Logger, logger=logger)
    cli = await aiohttp_client(server)  # type: ignore[var-annotated]
    resp = await cli.get("/path/to", headers={"Accept": "text/html"})
    assert resp.status == 500
    assert exc_msg == "RuntimeError: intentional runtime error"


async def test_async_logger(
    aiohttp_raw_server: AiohttpRawServer, aiohttp_client: AiohttpClient
) -> None:
    msg = None

    class Logger(AbstractAsyncAccessLogger):
        async def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            nonlocal msg
            msg = f"{request.path}: {response.status}"

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

    logger = mock.Mock()
    server = await aiohttp_raw_server(handler, access_log_class=Logger, logger=logger)
    cli = await aiohttp_client(server)  # type: ignore[var-annotated]
    resp = await cli.get("/path/to", headers={"Accept": "text/html"})
    assert resp.status == 200
    assert msg == "/path/to: 200"


async def test_contextvars_logger(
    aiohttp_server: AiohttpServer, aiohttp_client: AiohttpClient
) -> None:
    VAR = ContextVar[str]("VAR")

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    async def middleware(request: web.Request, handler: Handler) -> web.StreamResponse:
        VAR.set("uuid")
        return await handler(request)

    msg = None

    class Logger(AbstractAccessLogger):
        def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            nonlocal msg
            msg = f"contextvars: {VAR.get()}"

    app = web.Application(middlewares=[middleware])
    app.router.add_get("/", handler)
    server = await aiohttp_server(app, access_log_class=Logger)
    client = await aiohttp_client(server)  # type: ignore[var-annotated]
    resp = await client.get("/")
    assert 200 == resp.status
    assert msg == "contextvars: uuid"


def test_access_logger_feeds_logger(caplog: pytest.LogCaptureFixture) -> None:
    """Test that the logger still works."""
    mock_logger = logging.getLogger("test.aiohttp.log")
    mock_logger.setLevel(logging.INFO)
    access_logger = AccessLogger(mock_logger, "%b")
    access_logger.log(
        mock.Mock(name="mock_request"), mock.Mock(name="mock_response"), 42
    )
    assert "mock_response" in caplog.text


async def test_logger_does_not_log_when_not_enabled(
    aiohttp_server: AiohttpServer,
    aiohttp_client: AiohttpClient,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test logger does nothing when not enabled."""

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    class Logger(AbstractAccessLogger):

        def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            self.logger.critical("This should not be logged")  # pragma: no cover

        @property
        def enabled(self) -> bool:
            return False

    app = web.Application()
    app.router.add_get("/", handler)
    server = await aiohttp_server(app, access_log_class=Logger)
    client = await aiohttp_client(server)  # type: ignore[var-annotated]
    resp = await client.get("/")
    assert 200 == resp.status
    assert "This should not be logged" not in caplog.text


async def test_logger_set_to_none(
    aiohttp_server: AiohttpServer,
    aiohttp_client: AiohttpClient,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test logger does nothing when access_log is set to None."""

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    class Logger(AbstractAccessLogger):

        def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            self.logger.critical("This should not be logged")  # pragma: no cover

    app = web.Application()
    app.router.add_get("/", handler)
    server = await aiohttp_server(app, access_log=None, access_log_class=Logger)
    client = await aiohttp_client(server)  # type: ignore[var-annotated]
    resp = await client.get("/")
    assert 200 == resp.status
    assert "This should not be logged" not in caplog.text


async def test_logger_does_not_log_when_enabled_post_init(
    aiohttp_server: AiohttpServer,
    aiohttp_client: AiohttpClient,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test logger does nothing when not enabled even if enabled post init."""

    async def handler(request: web.Request) -> web.Response:
        return web.Response()

    enabled = False

    class Logger(AbstractAccessLogger):

        def log(
            self, request: web.BaseRequest, response: web.StreamResponse, time: float
        ) -> None:
            self.logger.critical("This should not be logged")  # pragma: no cover

        @property
        def enabled(self) -> bool:
            """Check if logger is enabled."""
            # Avoid formatting the log line if it will not be emitted.
            return enabled

    app = web.Application()
    app.router.add_get("/", handler)
    server = await aiohttp_server(app, access_log_class=Logger)
    client = await aiohttp_client(server)  # type: ignore[var-annotated]
    resp = await client.get("/")
    assert 200 == resp.status
    assert "This should not be logged" not in caplog.text
    assert not server.handler.connections[0]._force_close

    # mock enabling logging post-init
    enabled = True
    resp = await client.get("/")
    assert 200 == resp.status
    assert "This should not be logged" not in caplog.text
    assert not server.handler.connections[0]._force_close
