# SPDX-License-Identifier: MIT OR Apache-2.0
# This file is dual licensed under the terms of the Apache License, Version
# 2.0, and the MIT License.  See the LICENSE file in the root of this
# repository for complete details.

from __future__ import annotations

import json
import logging
import logging.config
import os
import sys

from collections.abc import Callable, Collection
from io import StringIO
from typing import Any
from unittest.mock import patch

import pytest
import pytest_asyncio

from structlog import (
    BytesLogger,
    PrintLogger,
    ReturnLogger,
    configure,
    get_context,
    wrap_logger,
)
from structlog._config import _CONFIG
from structlog._log_levels import CRITICAL, DEBUG, NAME_TO_LEVEL, WARN
from structlog.dev import ConsoleRenderer
from structlog.exceptions import DropEvent
from structlog.processors import JSONRenderer, KeyValueRenderer
from structlog.stdlib import (
    AsyncBoundLogger,
    BoundLogger,
    ExtraAdder,
    LoggerFactory,
    PositionalArgumentsFormatter,
    ProcessorFormatter,
    _FixedFindCallerLogger,
    add_log_level,
    add_log_level_number,
    add_logger_name,
    filter_by_level,
    get_logger,
    recreate_defaults,
    render_to_log_args_and_kwargs,
    render_to_log_kwargs,
)
from structlog.testing import CapturedCall
from structlog.typing import BindableLogger, EventDict

from .additional_frame import additional_frame
from .helpers import call_recorder, stub


def build_bl(logger=None, processors=None, context=None):
    """
    Convenience function to build BoundLogger with sane defaults.
    """
    return BoundLogger(logger or ReturnLogger(), processors, {})


def return_method_name(_, method_name, __):
    """
    A final renderer that returns the name of the logging method.
    """
    return method_name


class TestLoggerFactory:
    def setup_method(self, method):
        """
        The stdlib logger factory modifies global state to fix caller
        identification.
        """
        self.original_logger = logging.getLoggerClass()

    def teardown_method(self, method):
        logging.setLoggerClass(self.original_logger)

    def test_deduces_correct_name(self):
        """
        The factory isn't called directly but from structlog._config so
        deducing has to be slightly smarter.
        """
        assert "tests.additional_frame" == (
            additional_frame(LoggerFactory()).name
        )
        assert "tests.test_stdlib" == LoggerFactory()().name

    def test_ignores_frames(self):
        """
        The name guesser walks up the frames until it reaches a frame whose
        name is not from structlog or one of the configurable other names.
        """

        # Compute the names to __main__ so it doesn't get thrown off if people
        # install plugins that alter the frames. E.g. #370
        names = set()
        f = sys._getframe()
        while f.f_globals["__name__"] != "__main__":
            names.add(f.f_globals["__name__"].split(".", 1)[0])
            f = f.f_back

        assert (
            "__main__"
            == additional_frame(
                LoggerFactory(ignore_frame_names=list(names))
            ).name
        )

    def test_deduces_correct_caller(self):
        """
        It will find the correct caller.
        """
        logger = _FixedFindCallerLogger("test")

        file_name, _line_number, func_name = logger.findCaller()[:3]

        assert file_name == os.path.realpath(__file__)
        assert func_name == "test_deduces_correct_caller"

    def test_stack_info(self):
        """
        If we ask for stack_info, it will returned.
        """
        logger = _FixedFindCallerLogger("test")
        testing, is_, fun, stack_info = logger.findCaller(stack_info=True)  # noqa: RUF059

        assert "testing, is_, fun" in stack_info

    def test_no_stack_info_by_default(self):
        """
        If we don't ask for stack_info, it won't be returned.
        """
        logger = _FixedFindCallerLogger("test")
        testing, is_, fun, stack_info = logger.findCaller()  # noqa: RUF059

        assert None is stack_info

    def test_find_caller(self, caplog):
        """
        The caller is found.
        """
        logger = LoggerFactory()()

        logger.error("Test")

        assert caplog.text.startswith(
            "ERROR    tests.test_stdlib:test_stdlib.py"
        )

    def test_sets_correct_logger(self):
        """
        Calling LoggerFactory ensures that Logger.findCaller gets patched.
        """
        LoggerFactory()

        assert logging.getLoggerClass() is _FixedFindCallerLogger

    def test_positional_argument_avoids_guessing(self):
        """
        If a positional argument is passed to the factory, it's used as the
        name instead of guessing.
        """
        lf = LoggerFactory()("foo")

        assert "foo" == lf.name


class TestFilterByLevel:
    def test_filters_lower_levels(self):
        """
        Log entries below the current level raise a DropEvent.
        """
        logger = logging.Logger(__name__)
        logger.setLevel(CRITICAL)

        with pytest.raises(DropEvent):
            filter_by_level(logger, "warn", {})

    def test_passes_higher_levels(self):
        """
        Log entries with higher levels are passed through unchanged.
        """
        logger = logging.Logger(__name__)
        logger.setLevel(WARN)
        event_dict = {"event": "test"}

        assert event_dict is filter_by_level(logger, "warn", event_dict)
        assert event_dict is filter_by_level(logger, "error", event_dict)
        assert event_dict is filter_by_level(logger, "exception", event_dict)


class TestBoundLogger:
    @pytest.mark.parametrize(
        ("method_name"),
        ["debug", "info", "warning", "error", "exception", "critical"],
    )
    def test_proxies_to_correct_method(self, method_name):
        """
        The basic proxied methods are proxied to the correct counterparts.
        """
        bl = BoundLogger(ReturnLogger(), [return_method_name], {})

        assert method_name == getattr(bl, method_name)("event")

    def test_proxies_to_correct_method_special_cases(self):
        """
        Fatal maps to critical and warn to warning.
        """
        bl = BoundLogger(ReturnLogger(), [return_method_name], {})

        assert "warning" == bl.warn("event")
        assert "critical" == bl.fatal("event")

    def test_proxies_log(self):
        """
        BoundLogger.exception.log() is proxied to the appropriate method.
        """
        bl = BoundLogger(ReturnLogger(), [return_method_name], {})

        assert "critical" == bl.log(50, "event")
        assert "debug" == bl.log(10, "event")

    def test_positional_args_proxied(self):
        """
        Positional arguments supplied are proxied as kwarg.
        """
        bl = BoundLogger(ReturnLogger(), [], {})
        _args, kwargs = bl.debug("event", "foo", bar="baz")

        assert "baz" == kwargs.get("bar")
        assert ("foo",) == kwargs.get("positional_args")

    @pytest.mark.parametrize(
        "attribute_name",
        ["name", "level", "parent", "propagate", "handlers", "disabled"],
    )
    def test_stdlib_passthrough_attributes(self, attribute_name):
        """
        stdlib logger attributes are also available in stdlib BoundLogger.
        """
        stdlib_logger = logging.getLogger("Test")
        stdlib_logger_attribute = getattr(stdlib_logger, attribute_name)
        bl = BoundLogger(stdlib_logger, [], {})
        bound_logger_attribute = getattr(bl, attribute_name)

        assert bound_logger_attribute == stdlib_logger_attribute

    @pytest.mark.parametrize(
        ("method_name", "method_args"),
        [
            ("addHandler", [None]),
            ("removeHandler", [None]),
            ("hasHandlers", None),
            ("callHandlers", [None]),
            ("handle", [None]),
            ("setLevel", [None]),
            ("getEffectiveLevel", None),
            ("isEnabledFor", [None]),
            ("findCaller", None),
            (
                "makeRecord",
                [
                    "name",
                    "debug",
                    "test_func",
                    "1",
                    "test msg",
                    ["foo"],
                    False,
                ],
            ),
            ("getChild", [None]),
        ],
    )
    def test_stdlib_passthrough_methods(self, method_name, method_args):
        """
        stdlib logger methods are also available in stdlib BoundLogger.
        """
        called_stdlib_method = [False]

        def validate(*args, **kw):
            called_stdlib_method[0] = True

        stdlib_logger = logging.getLogger("Test")
        stdlib_logger_method = getattr(stdlib_logger, method_name, None)
        if stdlib_logger_method:
            setattr(stdlib_logger, method_name, validate)
            bl = BoundLogger(stdlib_logger, [], {})
            bound_logger_method = getattr(bl, method_name)

            assert bound_logger_method is not None

            if method_args:
                bound_logger_method(*method_args)
            else:
                bound_logger_method()

            assert called_stdlib_method[0] is True

    def test_is_enabled_for(self):
        """
        is_enabled_for is a snake_case alias that delegates to the wrapped
        logger's isEnabledFor, for compatibility with FilteringBoundLogger.
        """
        stdlib_logger = logging.getLogger("test_is_enabled_for")
        stdlib_logger.setLevel(WARN)
        bl = BoundLogger(stdlib_logger, [], {})

        assert bl.is_enabled_for(WARN) is True
        assert bl.is_enabled_for(DEBUG) is False

    def test_get_effective_level(self):
        """
        get_effective_level is a snake_case alias that delegates to the wrapped
        logger's getEffectiveLevel, for compatibility with
        FilteringBoundLogger.
        """
        stdlib_logger = logging.getLogger("test_get_effective_level")
        stdlib_logger.setLevel(WARN)
        bl = BoundLogger(stdlib_logger, [], {})

        assert WARN == bl.get_effective_level()

    def test_exception_exc_info(self):
        """
        BoundLogger.exception sets exc_info=True.
        """
        bl = BoundLogger(ReturnLogger(), [], {})

        assert ((), {"exc_info": True, "event": "event"}) == bl.exception(
            "event"
        )

    def test_exception_exc_info_override(self):
        """
        If *exc_info* is password to exception, it's used.
        """
        bl = BoundLogger(ReturnLogger(), [], {})

        assert ((), {"exc_info": 42, "event": "event"}) == bl.exception(
            "event", exc_info=42
        )

    def test_proxies_bind(self):
        """
        Bind calls the correct bind.
        """
        bl = build_bl().bind(a=42)

        assert {"a": 42} == get_context(bl)

    def test_proxies_new(self):
        """
        Newcalls the correct new.
        """
        bl = build_bl().bind(a=42).new(b=23)

        assert {"b": 23} == get_context(bl)

    def test_proxies_unbind(self):
        """
        Unbind calls the correct unbind.
        """
        bl = build_bl().bind(a=42).unbind("a")

        assert {} == get_context(bl)

    def test_proxies_try_unbind(self):
        """
        try_unbind calls the correct try_unbind.
        """
        bl = build_bl().bind(a=42).try_unbind("a", "b")

        assert {} == get_context(bl)

    @pytest.mark.parametrize(
        "meth", ["debug", "info", "warning", "error", "critical"]
    )
    async def test_async_log_methods(self, meth, cl):
        """
        Async methods log async.
        """
        bl = build_bl(cl, processors=[])

        await getattr(bl, f"a{meth}")("Async!")

        assert [
            CapturedCall(method_name=meth, args=(), kwargs={"event": "Async!"})
        ] == cl.calls

    async def test_async_log_methods_special_cases(self, cl):
        """
        afatal maps to critical.
        """
        bl = build_bl(cl, processors=[])

        await bl.afatal("Async!")

        assert [
            CapturedCall(
                method_name="critical", args=(), kwargs={"event": "Async!"}
            )
        ] == cl.calls

    async def test_alog(self, cl):
        """
        Alog logs async at the correct level.
        """
        bl = build_bl(cl, processors=[])

        await bl.alog(logging.INFO, "foo %s", "bar")

        assert [
            CapturedCall(
                method_name="info",
                args=(),
                kwargs={"positional_args": ("bar",), "event": "foo %s"},
            )
        ] == cl.calls

    async def test_aexception_exc_info_true(self, cl):
        """
        aexception passes current exc_info into dispatch.
        """
        bl = build_bl(cl, processors=[])

        try:
            raise ValueError(42)
        except ValueError as e:
            await bl.aexception("oops")
            exc = e

        (cc,) = cl.calls

        assert isinstance(cc[2]["exc_info"], tuple)
        assert exc == cc[2]["exc_info"][1]

    async def test_aexception_exc_info_explicit(self, cl):
        """
        In aexception, if exc_info isn't missing or True, leave it be.
        """
        bl = build_bl(cl, processors=[])

        obj = object()

        await bl.aexception("ooops", exc_info=obj)

        assert obj is cl.calls[0].kwargs["exc_info"]


class TestPositionalArgumentsFormatter:
    def test_formats_tuple(self):
        """
        Positional arguments as simple types are rendered.
        """
        formatter = PositionalArgumentsFormatter()
        event_dict = formatter(
            None,
            None,
            {"event": "%d %d %s", "positional_args": (1, 2, "test")},
        )

        assert "1 2 test" == event_dict["event"]
        assert "positional_args" not in event_dict

    def test_formats_dict(self):
        """
        Positional arguments as dict are rendered.
        """
        formatter = PositionalArgumentsFormatter()
        event_dict = formatter(
            None,
            None,
            {"event": "%(foo)s bar", "positional_args": ({"foo": "bar"},)},
        )

        assert "bar bar" == event_dict["event"]
        assert "positional_args" not in event_dict

    def test_positional_args_retained(self):
        """
        Positional arguments are retained if remove_positional_args
        argument is set to False.
        """
        formatter = PositionalArgumentsFormatter(remove_positional_args=False)
        positional_args = (1, 2, "test")
        event_dict = formatter(
            None,
            None,
            {"event": "%d %d %s", "positional_args": positional_args},
        )

        assert "positional_args" in event_dict
        assert positional_args == event_dict["positional_args"]

    def test_nop_no_args(self):
        """
        If no positional args are passed, nothing happens.
        """
        formatter = PositionalArgumentsFormatter()

        assert {} == formatter(None, None, {})

    def test_args_removed_if_empty(self):
        """
        If remove_positional_args is True and positional_args is (), still
        remove them.

        Regression test for https://github.com/hynek/structlog/issues/82.
        """
        formatter = PositionalArgumentsFormatter()

        assert {} == formatter(None, None, {"positional_args": ()})


class TestAddLogLevelNumber:
    @pytest.mark.parametrize(("level", "number"), NAME_TO_LEVEL.items())
    def test_log_level_number_added(self, level, number):
        """
        The log level number is added to the event dict.
        """
        event_dict = add_log_level_number(None, level, {})

        assert number == event_dict["level_number"]


class TestAddLogLevel:
    def test_log_level_added(self):
        """
        The log level is added to the event dict.
        """
        event_dict = add_log_level(None, "error", {})

        assert "error" == event_dict["level"]

    @pytest.mark.parametrize(
        ("alias", "normalized"), [("warn", "warning"), ("exception", "error")]
    )
    def test_log_level_alias_normalized(self, alias, normalized):
        """
        The normalized name of the log level is added to the event dict.
        """
        event_dict = add_log_level(None, alias, {})

        assert normalized == event_dict["level"]


@pytest.fixture(name="make_log_record")
def _make_log_record():
    """
    A LogRecord factory.
    """

    def create_log_record(**kwargs):
        defaults = {
            "name": "sample-name",
            "level": logging.INFO,
            "pathname": None,
            "lineno": None,
            "msg": "sample-message",
            "args": [],
            "exc_info": None,
        }
        defaults.update(kwargs)
        return logging.LogRecord(**defaults)

    return create_log_record


class TestAddLoggerName:
    def test_logger_name_added(self):
        """
        The logger name is added to the event dict.
        """
        name = "sample-name"
        logger = logging.getLogger(name)
        event_dict = add_logger_name(logger, None, {})

        assert name == event_dict["logger"]

    def test_logger_name_added_with_record(self, make_log_record):
        """
        The logger name is deduced from the LogRecord if provided.
        """
        name = "sample-name"
        record = make_log_record(name=name)
        event_dict = add_logger_name(None, None, {"_record": record})

        assert name == event_dict["logger"]

    def test_logger_name_added_with_bytes_logger(self):
        """
        add_logger_name works with BytesLogger that has a name attribute.
        """
        name = "sample-name"
        logger = BytesLogger(name=name)
        event_dict = add_logger_name(logger, None, {})

        assert name == event_dict["logger"]

    def test_logger_name_none_with_unnamed_bytes_logger(self):
        """
        add_logger_name works with BytesLogger without a name, returning None.
        """
        logger = BytesLogger()
        event_dict = add_logger_name(logger, None, {})

        assert event_dict["logger"] is None


def extra_dict() -> dict[str, Any]:
    """
    A dict to be passed in the `extra` parameter of the `logging` module's log
    methods.
    """
    return {
        "this": "is",
        "some": "extra values",
        "x_int": 4,
        "x_bool": True,
    }


@pytest.fixture(name="extra_dict")
def extra_dict_fixture():
    return extra_dict()


class TestExtraAdder:
    @pytest.mark.parametrize(
        ("allow", "misses"),
        [
            (None, None),
            ({}, None),
            *[({key}, None) for key in extra_dict()],
            ({"missing"}, {"missing"}),
            ({"missing", "keys"}, {"missing"}),
            ({"this", "x_int"}, None),
        ],
    )
    def test_add_extra(
        self,
        make_log_record: Callable[[], logging.LogRecord],
        extra_dict: dict[str, Any],
        allow: Collection[str] | None,
        misses: set[str] | None,
    ):
        """
        Extra attributes of a LogRecord object are added to the event dict.
        """
        record: logging.LogRecord = make_log_record()
        record.__dict__.update(extra_dict)
        event_dict = {"_record": record, "ed_key": "ed_value"}
        expected = self._copy_allowed(event_dict, extra_dict, allow)

        if allow is None:
            actual = ExtraAdder()(None, None, event_dict)
            assert expected == actual
        actual = ExtraAdder(allow)(None, None, event_dict)
        assert expected == actual
        if misses:
            assert misses.isdisjoint(expected.keys())

    def test_no_record(self):
        """
        If the event_dict has no LogRecord, do nothing.
        """
        actual = ExtraAdder()(None, None, {})

        assert {} == actual

    @pytest.mark.parametrize(
        ("allow", "misses"),
        [
            (None, None),
            ({}, None),
            *[({key}, None) for key in extra_dict()],
            ({"missing"}, {"missing"}),
            ({"missing", "keys"}, {"missing"}),
            ({"this", "x_int"}, None),
        ],
    )
    def test_add_extra_e2e(
        self,
        extra_dict: dict[str, Any],
        allow: Collection[str] | None,
        misses: set[str] | None,
    ):
        """
        Values passed in the `extra` parameter of the `logging` module's log
        methods pass through to log output.
        """
        logger = logging.Logger(sys._getframe().f_code.co_name)
        string_io = StringIO()
        handler = logging.StreamHandler(string_io)
        formatter = ProcessorFormatter(
            foreign_pre_chain=[ExtraAdder(allow)],
            processors=[JSONRenderer()],
        )
        handler.setFormatter(formatter)
        handler.setLevel(0)
        logger.addHandler(handler)
        logger.setLevel(0)
        logging.warning("allow = %s", allow)  # noqa: LOG015

        event_dict = {"event": "Some text"}
        expected = self._copy_allowed(event_dict, extra_dict, allow)

        logger.info("Some %s", "text", extra=extra_dict)
        actual = {
            key: value
            for key, value in json.loads(string_io.getvalue()).items()
            if not key.startswith("_")
        }

        assert expected == actual
        if misses:
            assert misses.isdisjoint(expected.keys())

    @classmethod
    def _copy_allowed(
        cls,
        event_dict: EventDict,
        extra_dict: dict[str, Any],
        allow: Collection[str] | None,
    ) -> EventDict:
        if allow is None:
            return {**event_dict, **extra_dict}

        return {
            **event_dict,
            **{
                key: value for key, value in extra_dict.items() if key in allow
            },
        }


@pytest.fixture(name="stdlib_logger")
def _stdlib_logger():
    logger = logging.getLogger("test_logger")
    logger.setLevel(logging.DEBUG)

    yield logger

    logging.basicConfig()


class TestRenderToLogArgsAndKwargs:
    def test_default(self, stdlib_logger: logging.Logger):
        """
        Passes `event` key from `event_dict` in the first positional argument
        and handles otherwise empty `event_dict`.
        """
        method_name = "debug"
        event = "message"
        args, kwargs = render_to_log_args_and_kwargs(
            stdlib_logger, method_name, {"event": event}
        )

        assert (event,) == args
        assert {} == kwargs

        with patch.object(stdlib_logger, "_log") as mock_log:
            getattr(stdlib_logger, method_name)(*args, **kwargs)

        mock_log.assert_called_once_with(logging.DEBUG, event, ())

    def test_pass_remaining_event_dict_as_extra(
        self, stdlib_logger: logging.Logger, event_dict: dict[str, Any]
    ):
        """
        Passes remaining `event_dict` as `extra`.
        """
        expected_extra = event_dict.copy()

        method_name = "info"
        event = "message"
        event_dict["event"] = event

        args, kwargs = render_to_log_args_and_kwargs(
            stdlib_logger, method_name, event_dict
        )

        assert (event,) == args
        assert {"extra": expected_extra} == kwargs

        with patch.object(stdlib_logger, "_log") as mock_log:
            getattr(stdlib_logger, method_name)(*args, **kwargs)

        mock_log.assert_called_once_with(
            logging.INFO, event, (), extra=expected_extra
        )

    def test_pass_positional_args_from_event_dict_as_args(
        self, stdlib_logger: logging.Logger, event_dict: dict[str, Any]
    ):
        """
        Passes items from "positional_args" key from `event_dict` as positional
        arguments.
        """
        expected_extra = event_dict.copy()

        method_name = "warning"
        event = "message: a = %s, b = %d"
        positional_args = ("foo", 123)
        event_dict["event"] = event
        event_dict["positional_args"] = positional_args

        args, kwargs = render_to_log_args_and_kwargs(
            stdlib_logger, method_name, event_dict
        )

        assert (event, *(positional_args)) == args
        assert {"extra": expected_extra} == kwargs

        with patch.object(stdlib_logger, "_log") as mock_log:
            getattr(stdlib_logger, method_name)(*args, **kwargs)

        mock_log.assert_called_once_with(
            logging.WARNING, event, positional_args, extra=expected_extra
        )

    def test_pass_kwargs_from_event_dict_as_kwargs(
        self, stdlib_logger: logging.Logger, event_dict: dict[str, Any]
    ):
        """
        Passes "exc_info", "stack_info", and "stacklevel" keys from `event_dict`
        as keyword arguments.
        """
        expected_extra = event_dict.copy()

        method_name = "info"
        event = "message"
        exc_info = True
        stack_info = False
        stacklevel = 2
        event_dict["event"] = event
        event_dict["exc_info"] = exc_info
        event_dict["stack_info"] = stack_info
        event_dict["stacklevel"] = stacklevel

        args, kwargs = render_to_log_args_and_kwargs(
            stdlib_logger, method_name, event_dict
        )

        assert (event,) == args
        assert {
            "exc_info": exc_info,
            "stack_info": stack_info,
            "stacklevel": stacklevel,
            "extra": expected_extra,
        } == kwargs

        with patch.object(stdlib_logger, "_log") as mock_log:
            getattr(stdlib_logger, method_name)(*args, **kwargs)

        mock_log.assert_called_once_with(
            logging.INFO,
            event,
            (),
            exc_info=exc_info,
            stack_info=stack_info,
            stacklevel=stacklevel,
            extra=expected_extra,
        )

    def test_integration(
        self, stdlib_logger: logging.Logger, event_dict: dict[str, Any]
    ):
        """
        `render_to_log_args_and_kwargs` with a wrapped logger calls the stdlib
        logger correctly.

        Reserved stdlib keyword arguments are in `logging.Logger._log`.
        https://github.com/python/cpython/blob/60403a5409ff2c3f3b07dd2ca91a7a3e096839c7/Lib/logging/__init__.py#L1640
        """
        event = "message: a = %s, b = %d"
        arg_1 = "foo"
        arg_2 = 123
        exc_info = False
        stack_info = True
        stacklevel = 3

        struct_logger = wrap_logger(
            stdlib_logger,
            processors=[render_to_log_args_and_kwargs],
            wrapper_class=BoundLogger,
        )

        with patch.object(stdlib_logger, "_log") as mock_log:
            struct_logger.info(
                event,
                arg_1,
                arg_2,
                exc_info=exc_info,
                stack_info=stack_info,
                stacklevel=stacklevel,
                **event_dict,
            )

        mock_log.assert_called_once_with(
            logging.INFO,
            event,
            (arg_1, arg_2),
            exc_info=exc_info,
            stack_info=stack_info,
            stacklevel=stacklevel,
            extra=event_dict,
        )


class TestRenderToLogKwargs:
    def test_default(self, stdlib_logger):
        """
        Translates `event` to `msg` and handles otherwise empty `event_dict`s.
        """
        d = render_to_log_kwargs(None, None, {"event": "message"})

        assert {"msg": "message", "extra": {}} == d

        # now check stdlib logger likes those kwargs
        with patch.object(stdlib_logger, "_log") as mock_log:
            stdlib_logger.info(**d)

        mock_log.assert_called_once_with(logging.INFO, "message", (), extra={})

    def test_add_extra_event_dict(self, event_dict, stdlib_logger):
        """
        Adds all remaining data from `event_dict` into `extra`.
        """
        event_dict["event"] = "message"
        d = render_to_log_kwargs(None, None, event_dict)

        assert {"msg": "message", "extra": event_dict} == d

        # now check stdlib logger likes those kwargs
        with patch.object(stdlib_logger, "_log") as mock_log:
            stdlib_logger.info(**d)

        mock_log.assert_called_once_with(
            logging.INFO, "message", (), extra=event_dict
        )

    def test_handles_special_kw(self, event_dict, stdlib_logger):
        """
        "exc_info", "stack_info", and "stacklevel" aren't passed as extras.

        Cf. https://github.com/hynek/structlog/issues/424
        """
        del event_dict["a"]  # needs a repr
        event_dict["event"] = "message"

        event_dict["exc_info"] = True
        event_dict["stack_info"] = False
        event_dict["stacklevel"] = 1
        event_dict["stackLevel"] = 1  # not a reserved kw

        d = render_to_log_kwargs(None, None, event_dict)
        expected = {
            "msg": "message",
            "exc_info": True,
            "stack_info": False,
            "stacklevel": 1,
            "extra": {
                "b": [3, 4],
                "x": 7,
                "y": "test",
                "z": (1, 2),
                "stackLevel": 1,
            },
        }

        assert expected == d

        # now check stdlib logger likes those kwargs
        with patch.object(stdlib_logger, "_log") as mock_log:
            stdlib_logger.info(**d)

        expected.pop("msg")
        mock_log.assert_called_once_with(
            logging.INFO, "message", (), **expected
        )

    def test_integration_special_kw(self, event_dict, stdlib_logger):
        """
        render_to_log_kwargs with a wrapped logger calls the stdlib logger
        correctly

        reserved stdlib keywords are in logging.Logger._log
        https://github.com/python/cpython/blob/ae7b17673f29efe17b416cbcfbf43b5b3ff5977c/Lib/logging/__init__.py#L1632
        """
        expected = {
            "msg": "message",
            "exc_info": True,
            "stack_info": False,
            "stacklevel": 1,
            "extra": {**event_dict},
        }

        event_dict["exc_info"] = True
        event_dict["stack_info"] = False
        event_dict["stacklevel"] = 1

        struct_logger = wrap_logger(
            stdlib_logger,
            processors=[render_to_log_kwargs],
        )

        # now check struct logger passes those kwargs to stdlib
        with patch.object(stdlib_logger, "_log") as mock_log:
            struct_logger.info("message", **event_dict)

        expected.pop("msg")
        mock_log.assert_called_once_with(
            logging.INFO, "message", (), **expected
        )


@pytest.fixture(name="configure_for_processor_formatter")
def _configure_for_processor_formatter():
    """
    Configure structlog to use ProcessorFormatter.

    Reset logging setting after the test (structlog is reset automatically
    before all tests).
    """
    configure(
        processors=[add_log_level, ProcessorFormatter.wrap_for_formatter],
        logger_factory=LoggerFactory(),
        wrapper_class=BoundLogger,
    )

    yield

    logging.basicConfig()


def configure_logging(
    pre_chain,
    logger=None,
    pass_foreign_args=False,
    renderer=ConsoleRenderer(colors=False),  # noqa: B008
):
    """
    Configure logging to use ProcessorFormatter.

    Return a list that is filled with event dicts form calls.
    """
    event_dicts = []

    def capture(_, __, ed):
        event_dicts.append(ed.copy())

        return ed

    logging.config.dictConfig(
        {
            "version": 1,
            "disable_existing_loggers": False,
            "formatters": {
                "plain": {
                    "()": ProcessorFormatter,
                    "processors": [
                        capture,
                        ProcessorFormatter.remove_processors_meta,
                        renderer,
                    ],
                    "foreign_pre_chain": pre_chain,
                    "format": "%(message)s [in %(funcName)s]",
                    "logger": logger,
                    "pass_foreign_args": pass_foreign_args,
                }
            },
            "handlers": {
                "default": {
                    "level": "DEBUG",
                    "class": "logging.StreamHandler",
                    "formatter": "plain",
                }
            },
            "loggers": {
                "": {
                    "handlers": ["default"],
                    "level": "DEBUG",
                    "propagate": True,
                }
            },
        }
    )

    return event_dicts


@pytest.mark.usefixtures("configure_for_processor_formatter")
class TestProcessorFormatter:
    """
    These are all integration tests because they're all about integration.
    """

    def test_foreign_delegate(self, capsys):
        """
        If foreign_pre_chain is None, non-structlog log entries are delegated
        to logging. The processor chain's event dict is invoked with
        `_from_structlog=False`
        """
        calls = configure_logging(None)

        logging.getLogger().warning("foo")

        assert ("", "foo [in test_foreign_delegate]\n") == capsys.readouterr()
        assert calls[0]["_from_structlog"] is False
        assert isinstance(calls[0]["_record"], logging.LogRecord)

    def test_clears_args(self, capsys):
        """
        We render our log records before sending it back to logging.  Therefore
        we must clear `LogRecord.args` otherwise the user gets an
        `TypeError: not all arguments converted during string formatting.` if
        they use positional formatting in stdlib logging.
        """
        configure_logging(None)

        logging.getLogger().warning("hello %s.", "world")

        assert (
            "",
            "hello world. [in test_clears_args]\n",
        ) == capsys.readouterr()

    def test_pass_foreign_args_true_sets_positional_args_key(self):
        """
        If `pass_foreign_args` is `True` we set the `positional_args` key in
        the `event_dict` before clearing args.
        """
        test_processor = call_recorder(lambda _, __, event_dict: event_dict)
        configure_logging((test_processor,), pass_foreign_args=True)

        positional_args = {"foo": "bar"}
        logging.getLogger().info("okay %(foo)s", positional_args)

        event_dict = test_processor.call_args_list[0].args[2]

        assert "positional_args" in event_dict
        assert positional_args == event_dict["positional_args"]

    def test_log_dict(self, capsys):
        """
        dicts can be logged with std library loggers.
        """
        configure_logging(None)

        logging.getLogger().warning({"foo": "bar"})

        assert (
            "",
            "{'foo': 'bar'} [in test_log_dict]\n",
        ) == capsys.readouterr()

    def test_foreign_pre_chain(self, capsys):
        """
        If foreign_pre_chain is an iterable, it's used to pre-process
        non-structlog log entries.
        """
        configure_logging([add_log_level])

        logging.getLogger().warning("foo")

        assert (
            "",
            "[warning  ] foo [in test_foreign_pre_chain]\n",
        ) == capsys.readouterr()

    def test_foreign_pre_chain_add_logger_name(self, capsys):
        """
        foreign_pre_chain works with add_logger_name processor.
        """
        configure_logging((add_logger_name,))

        logging.getLogger("sample-name").warning("foo")

        assert (
            "",
            (
                "foo                            [sample-name] [in test_foreign_pr"
                "e_chain_add_logger_name]\n"
            ),
        ) == capsys.readouterr()

    def test_foreign_chain_can_pass_dictionaries_without_excepting(
        self, capsys
    ):
        """
        If a foreign logger passes a dictionary to a logging function,
        check we correctly identify that it did not come from structlog.
        """
        configure_logging(None)
        configure(
            processors=[ProcessorFormatter.wrap_for_formatter],
            logger_factory=LoggerFactory(),
            wrapper_class=BoundLogger,
        )

        logging.getLogger().warning({"foo": "bar"})

        assert (
            "",
            (
                "{'foo': 'bar'} [in "
                "test_foreign_chain_can_pass_dictionaries_without_excepting]\n"
            ),
        ) == capsys.readouterr()

    def test_foreign_pre_chain_gets_exc_info(self):
        """
        If non-structlog record contains exc_info, foreign_pre_chain functions
        have access to it.
        """
        test_processor = call_recorder(lambda _, __, event_dict: event_dict)
        configure_logging((test_processor,), renderer=KeyValueRenderer())

        try:
            raise RuntimeError("oh no")
        except Exception:
            logging.getLogger().exception("okay")

        event_dict = test_processor.call_args_list[0].args[2]

        assert "exc_info" in event_dict
        assert isinstance(event_dict["exc_info"], tuple)

    def test_foreign_pre_chain_sys_exc_info(self):
        """
        If a foreign_pre_chain function accesses sys.exc_info(),
        ProcessorFormatter should not have changed it.
        """

        class MyError(Exception):
            pass

        def add_excinfo(logger, log_method, event_dict):
            event_dict["exc_info"] = sys.exc_info()
            return event_dict

        test_processor = call_recorder(lambda _, __, event_dict: event_dict)
        configure_logging(
            (add_excinfo, test_processor), renderer=KeyValueRenderer()
        )

        try:
            raise MyError("oh no")
        except Exception:
            logging.getLogger().error("okay")

        event_dict = test_processor.call_args_list[0].args[2]

        assert MyError is event_dict["exc_info"][0]

    def test_other_handlers_get_original_record(self):
        """
        Logging handlers that come after the handler with ProcessorFormatter
        should receive original, unmodified record.
        """
        configure_logging(None)

        handler1 = logging.StreamHandler()
        handler1.setFormatter(ProcessorFormatter(JSONRenderer()))
        handler2 = stub(
            handle=call_recorder(lambda record: None),
            level=logging.INFO,
        )
        logger = logging.getLogger()
        logger.addHandler(handler1)
        logger.addHandler(handler2)

        logger.info("meh")

        assert 1 == len(handler2.handle.call_args_list)

        handler2_record = handler2.handle.call_args_list[0].args[0]

        assert "meh" == handler2_record.msg

    @pytest.mark.parametrize("keep", [True, False])
    def test_formatter_unsets_exc_info(self, capsys, keep):
        """
        Stack traces doesn't get printed outside of the json document when
        keep_exc_info are set to False but preserved if set to True.
        """
        configure_logging(None)
        logger = logging.getLogger()

        def format_exc_info_fake(logger, name, event_dict):
            del event_dict["exc_info"]
            event_dict["exception"] = "Exception!"
            return event_dict

        formatter = ProcessorFormatter(
            processor=JSONRenderer(),
            keep_stack_info=keep,
            keep_exc_info=keep,
            foreign_pre_chain=[format_exc_info_fake],
        )
        logger.handlers[0].setFormatter(formatter)

        try:
            raise RuntimeError("oh no")
        except Exception:
            logging.getLogger().exception("seen worse")

        out, err = capsys.readouterr()

        assert "" == out

        if keep is False:
            assert (
                '{"event": "seen worse", "exception": "Exception!"}\n'
            ) == err
        else:
            assert "Traceback (most recent call last):" in err

    @pytest.mark.parametrize("keep", [True, False])
    def test_formatter_unsets_stack_info(self, capsys, keep):
        """
        Stack traces doesn't get printed outside of the json document when
        keep_stack_info are set to False but preserved if set to True.
        """
        configure_logging(None)
        logger = logging.getLogger()

        formatter = ProcessorFormatter(
            processor=JSONRenderer(),
            keep_stack_info=keep,
            keep_exc_info=keep,
            foreign_pre_chain=[],
        )
        logger.handlers[0].setFormatter(formatter)

        logging.getLogger().warning("have a stack trace", stack_info=True)

        out, err = capsys.readouterr()

        assert "" == out

        if keep is False:
            assert 1 == err.count("Stack (most recent call last):")
        else:
            assert 2 == err.count("Stack (most recent call last):")

    def test_native(self, capsys):
        """
        If the log entry comes from structlog, it's unpackaged and processed.
        """
        eds = configure_logging(None)

        get_logger().warning("foo")

        assert (
            "",
            "[warning  ] foo [in test_native]\n",
        ) == capsys.readouterr()
        assert eds[0]["_from_structlog"] is True
        assert isinstance(eds[0]["_record"], logging.LogRecord)

    def test_native_logger(self, capsys):
        """
        If the log entry comes from structlog, it's unpackaged and processed.
        """
        logger = logging.getLogger()
        eds = configure_logging(None, logger=logger)

        get_logger().warning("foo")

        assert (
            "",
            "[warning  ] foo [in test_native_logger]\n",
        ) == capsys.readouterr()
        assert eds[0]["_from_structlog"] is True
        assert isinstance(eds[0]["_record"], logging.LogRecord)

    def test_foreign_pre_chain_filter_by_level(self, capsys):
        """
        foreign_pre_chain works with filter_by_level processor.
        """
        logger = logging.getLogger()
        configure_logging([filter_by_level], logger=logger)
        configure(
            processors=[ProcessorFormatter.wrap_for_formatter],
            logger_factory=LoggerFactory(),
            wrapper_class=BoundLogger,
        )

        logger.warning("foo")

        assert (
            "",
            "foo [in test_foreign_pre_chain_filter_by_level]\n",
        ) == capsys.readouterr()

    def test_processor_and_processors(self):
        """
        Passing both processor and processors raises a TypeError.
        """
        with pytest.raises(TypeError, match="mutually exclusive"):
            ProcessorFormatter(processor=1, processors=[1])

    def test_no_renderer(self):
        """
        Passing neither processor nor processors raises a TypeError.
        """
        with pytest.raises(TypeError, match="must be passed"):
            ProcessorFormatter()

    def test_remove_processors_meta(self):
        """
        remove_processors_meta removes _record and _from_structlog. And only
        them.
        """
        assert {"foo": "bar"} == ProcessorFormatter.remove_processors_meta(
            None,
            None,
            {"foo": "bar", "_record": "foo", "_from_structlog": True},
        )

    def test_non_string_message_warning(self):
        """
        A warning is raised if the last processor in
        ProcessorFormatter.processors doesn't return a string.
        """
        configure_logging(None)
        logger = logging.getLogger()

        formatter = ProcessorFormatter(
            processors=[lambda *args, **kwargs: {"foo": "bar"}],
        )
        logger.handlers[0].setFormatter(formatter)

        with pytest.warns(
            RuntimeWarning,
            match="The last processor in ProcessorFormatter.processors must return a string",
        ):
            logger.info("baz")

    def test_logrecord_exc_info(self):
        """
        LogRecord.exc_info is set consistently for structlog and non-structlog
        log records.
        """
        configure_logging(None)

        # This doesn't test ProcessorFormatter itself directly, but it's
        # relevant to setups where ProcessorFormatter is used, i.e. where
        # handlers will receive LogRecord objects that come from both structlog
        # and non-structlog loggers.

        records: dict[str, logging.LogRecord] = {}

        class DummyHandler(logging.Handler):
            def emit(self, record):
                # Don't do anything; just store the record in the records dict
                # by its message, so we can assert things about it.
                if isinstance(record.msg, dict):
                    records[record.msg["event"]] = record
                else:
                    records[record.msg] = record

        stdlib_logger = logging.getLogger()
        structlog_logger = get_logger()

        # It doesn't matter which logger we add the handler to here.
        stdlib_logger.addHandler(DummyHandler())

        try:
            raise Exception("foo")
        except Exception:
            stdlib_logger.exception("bar")
            structlog_logger.exception("baz")

        stdlib_record = records.pop("bar")

        assert "bar" == stdlib_record.msg
        assert stdlib_record.exc_info
        assert Exception is stdlib_record.exc_info[0]
        assert ("foo",) == stdlib_record.exc_info[1].args

        structlog_record = records.pop("baz")

        assert "baz" == structlog_record.msg["event"]
        assert True is structlog_record.msg["exc_info"]
        assert structlog_record.exc_info
        assert Exception is structlog_record.exc_info[0]
        assert ("foo",) == structlog_record.exc_info[1].args

        assert not records

    def test_use_get_message_false(self):
        """
        If use_get_message_is False, the event is obtained using
        str(record.msg) instead of calling record.getMessage. That means
        positional formatting is not performed.
        """
        event_dicts = []

        def capture(_, __, ed):
            event_dicts.append(ed.copy())

            return str(ed)

        proc = ProcessorFormatter(processors=[capture], use_get_message=False)

        record = logging.LogRecord(
            "foo",
            logging.INFO,
            "path.py",
            42,
            "le msg: %s",
            ("keep separate",),
            None,
        )

        assert proc.format(record)
        assert "le msg: %s" == event_dicts[0]["event"]


@pytest_asyncio.fixture(name="abl")
async def _abl(cl):
    return AsyncBoundLogger(cl, context={}, processors=[])


class TestAsyncBoundLogger:
    def test_sync_bl(self, abl, cl):
        """
        AsyncBoungLogger.sync_bl works outside of loops.
        """
        abl.sync_bl.info("test")

        assert [
            CapturedCall(method_name="info", args=(), kwargs={"event": "test"})
        ] == cl.calls

    @pytest.mark.asyncio
    async def test_protocol(self, abl):
        """
        AsyncBoundLogger is a proper BindableLogger.
        """
        assert isinstance(abl, BindableLogger)

    @pytest.mark.asyncio
    async def test_correct_levels(self, abl, cl, stdlib_log_method):
        """
        The proxy methods call the correct upstream methods.
        """
        await getattr(abl.bind(foo="bar"), stdlib_log_method)("42")

        aliases = {"warn": "warning"}

        expect = aliases.get(stdlib_log_method, stdlib_log_method)

        assert expect == cl.calls[0].method_name

    @pytest.mark.asyncio
    async def test_correct_level_fatal(self, abl, cl):
        """
        fatal, that I have no idea why we support, maps to critical.
        """
        await abl.bind(foo="bar").fatal("42")

        assert "critical" == cl.calls[0].method_name

    @pytest.mark.asyncio
    async def test_log_method(self, abl, cl):
        """
        The `log` method is proxied too.
        """
        await abl.bind(foo="bar").log(logging.ERROR, "42")

        assert "error" == cl.calls[0].method_name

    @pytest.mark.asyncio
    async def test_exception(self, abl, cl):
        """
        `exception` makes sure 'exc_info" is set, if it's not set already.
        """
        try:
            raise ValueError("omg")
        except ValueError:
            await abl.exception("oops")

        ei = cl.calls[0].kwargs["exc_info"]

        assert ValueError is ei[0]
        assert ("omg",) == ei[1].args

    @pytest.mark.asyncio
    async def test_exception_do_not_overwrite(self, abl, cl):
        """
        `exception` leaves exc_info be, if it's set.
        """
        o1 = object()
        o2 = object()
        o3 = object()

        try:
            raise ValueError("omg")
        except ValueError:
            await abl.exception("oops", exc_info=(o1, o2, o3))

        ei = cl.calls[0].kwargs["exc_info"]
        assert (o1, o2, o3) == ei

    @pytest.mark.asyncio
    async def test_bind_unbind(self, cl):
        """
        new/bind/unbind/try_unbind are correctly propagated.
        """
        l1 = AsyncBoundLogger(cl, context={}, processors=[])

        l2 = l1.bind(x=42)

        assert l1 is not l2
        assert l1.sync_bl is not l2.sync_bl
        assert {} == l1._context
        assert {"x": 42} == l2._context

        l3 = l2.new(y=23)

        assert l2 is not l3
        assert l2.sync_bl is not l3.sync_bl
        assert {"y": 23} == l3._context

        l4 = l3.unbind("y")

        assert {} == l4._context
        assert l3 is not l4

        # N.B. x isn't bound anymore.
        l5 = l4.try_unbind("x")

        assert {} == l5._context
        assert l4 is not l5

    @pytest.mark.asyncio
    async def test_integration(self, capsys):
        """
        Configure and log an actual entry.
        """

        configure(
            processors=[add_log_level, JSONRenderer()],
            logger_factory=PrintLogger,
            wrapper_class=AsyncBoundLogger,
            cache_logger_on_first_use=True,
        )

        logger = get_logger()

        await logger.bind(foo="bar").info("baz", x="42")

        assert {
            "foo": "bar",
            "x": "42",
            "event": "baz",
            "level": "info",
        } == json.loads(capsys.readouterr().out)


@pytest.mark.parametrize("log_level", [None, 45])
def test_recreate_defaults(log_level):
    """
    Recreate defaults configures structlog and -- if asked -- logging.
    """
    logging.basicConfig(
        stream=sys.stderr,
        level=1,
        force=True,
    )

    recreate_defaults(log_level=log_level)

    assert BoundLogger is _CONFIG.default_wrapper_class
    assert dict is _CONFIG.default_context_class
    assert isinstance(_CONFIG.logger_factory, LoggerFactory)

    log = get_logger().bind()
    if log_level is not None:
        assert log_level == log.getEffectiveLevel()
    else:
        assert 1 == log.getEffectiveLevel()