# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

import builtins
import json
import logging
import os
import sys
from collections.abc import Callable, Generator
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import ANY

import pytest
import structlog
import structlog._frames
import structlog.stdlib
from _pytest.capture import CaptureFixture
from _pytest.logging import LogCaptureFixture
from _pytest.monkeypatch import MonkeyPatch

import haystack.utils.jupyter
from haystack import logging as haystack_logging
from test.tracing.utils import SpyingTracer


@pytest.fixture(autouse=True)
def reset_logging_config() -> None:
    # `configure_logging` attaches its handler to Haystack's own namespaces (and may flip `propagate`), so we snapshot
    # and restore the root logger plus those namespaces to keep tests isolated.
    names = ["haystack", "haystack_integrations", "haystack_experimental"]
    root_handlers = logging.root.handlers.copy()
    snapshots = {name: _snapshot_logger(name) for name in names}
    yield
    logging.root.handlers = root_handlers
    for name, (handlers, propagate, level) in snapshots.items():
        logger = logging.getLogger(name)
        logger.handlers = handlers
        logger.propagate = propagate
        logger.setLevel(level)


def _snapshot_logger(name: str) -> tuple[list[logging.Handler], bool, int]:
    logger = logging.getLogger(name)
    return logger.handlers.copy(), logger.propagate, logger.level


@pytest.fixture()
def restore_named_loggers() -> Generator[Callable[[str], logging.Logger], None, None]:
    """Snapshot a named logger's handlers/propagate/level and restore them after the test."""
    snapshots: dict[str, tuple[list[logging.Handler], bool, int]] = {}

    def _snapshot(name: str) -> logging.Logger:
        logger = logging.getLogger(name)
        snapshots[name] = (logger.handlers.copy(), logger.propagate, logger.level)
        return logger

    yield _snapshot

    for name, (handlers, propagate, level) in snapshots.items():
        logger = logging.getLogger(name)
        logger.handlers = handlers
        logger.propagate = propagate
        logger.setLevel(level)


@pytest.fixture()
def restore_structlog_config() -> Generator[None, None, None]:
    """Snapshot the global structlog configuration and restore it after the test."""
    was_configured = structlog.is_configured()
    config = structlog.get_config()
    yield
    if was_configured:
        structlog.configure(**config)
    else:
        structlog.reset_defaults()


def _sentinel_processor(logger: object, method_name: str, event_dict: dict) -> dict:
    """A no-op processor used to detect whether an existing structlog config was left untouched."""
    return event_dict


@pytest.fixture()
def set_context_var_key() -> Generator[str, None, None]:
    structlog.contextvars.bind_contextvars(context_var="value")

    yield "context_var"

    structlog.contextvars.unbind_contextvars("context_var")


class TestSkipLoggingConfiguration:
    def test_skip_logging_configuration(
        self, monkeypatch: MonkeyPatch, capfd: CaptureFixture, caplog: LogCaptureFixture
    ) -> None:
        monkeypatch.setenv("HAYSTACK_LOGGING_IGNORE_STRUCTLOG", "true")
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # the pytest fixture caplog only captures logs being rendered from the stdlib logging module
        assert caplog.messages == ["Hello, structured logging!"]

        # Nothing should be captured by capfd since structlog is not configured
        assert capfd.readouterr().err == ""


class TestStructuredLoggingConsoleRendering:
    def test_log_filtering_when_using_debug(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=False)

        logger = logging.getLogger("haystack.test_logging")
        logger.setLevel(logging.INFO)
        logger.debug("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err
        assert output == ""

    def test_log_filtering_when_using_debug_and_log_level_is_debug(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=False)

        logger = logging.getLogger("haystack.test_logging")
        logger.setLevel(logging.DEBUG)

        logger.debug("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err
        assert "Hello, structured logging" in output
        assert "{" not in output, "Seems JSON rendering is enabled when it should not be"

    def test_console_rendered_structured_log_even_if_no_tty_but_python_config(
        self, capfd: CaptureFixture, monkeypatch: MonkeyPatch
    ) -> None:
        monkeypatch.setattr(sys.stderr, "isatty", lambda: False)

        haystack_logging.configure_logging(use_json=False)

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err

        assert "Hello, structured logging!" in output
        assert "{" not in output, "Seems JSON rendering is enabled when it should not be"

    def test_console_rendered_structured_log_if_in_ipython(
        self, capfd: CaptureFixture, monkeypatch: MonkeyPatch
    ) -> None:
        monkeypatch.setattr(builtins, "__IPYTHON__", "true", raising=False)

        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err

        assert "Hello, structured logging!" in output
        assert "{" not in output, "Seems JSON rendering is enabled when it should not be"

    def test_console_rendered_structured_log_even_in_jupyter(
        self, capfd: CaptureFixture, monkeypatch: MonkeyPatch
    ) -> None:
        monkeypatch.setattr(haystack.utils.jupyter, haystack.utils.jupyter.is_in_jupyter.__name__, lambda: True)

        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err

        assert "Hello, structured logging!" in output
        assert "{" not in output, "Seems JSON rendering is enabled when it should not be"

    def test_console_rendered_structured_log_even_if_no_tty_but_forced_through_env(
        self, capfd: CaptureFixture, monkeypatch: MonkeyPatch
    ) -> None:
        monkeypatch.setenv("HAYSTACK_LOGGING_USE_JSON", "false")

        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err

        assert "Hello, structured logging!" in output
        assert "{" not in output, "Seems JSON rendering is enabled when it should not be"

    def test_console_rendered_structured_log(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err

        # Only check for the minute to be a bit more robust
        today = datetime.now(tz=timezone.utc).isoformat(timespec="minutes").replace("+00:00", "")
        assert today in output

        log_level = "warning"
        assert log_level in output

        assert "Hello, structured logging!" in output

        assert "key1" in output
        assert "value1" in output

    def test_logging_exceptions(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")

        def function_that_raises_and_adds_to_stack_trace():
            raise ValueError("This is an error")

        try:
            function_that_raises_and_adds_to_stack_trace()
        except ValueError:
            logger.exception("An error happened")

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err

        assert "An error happened" in output

    def test_logging_of_contextvars(self, capfd: CaptureFixture, set_context_var_key: str) -> None:
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err

        assert set_context_var_key in output


class TestStructuredLoggingJSONRendering:
    def test_logging_as_json_if_not_atty(self, capfd: CaptureFixture, monkeypatch: MonkeyPatch) -> None:
        monkeypatch.setattr(sys.stderr, "isatty", lambda: False)
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err
        parsed_output = json.loads(output)  # should not raise an error

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key1": "value1",
            "key2": "value2",
            "level": "warning",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_logging_as_json(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err
        parsed_output = json.loads(output)  # should not raise an error

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key1": "value1",
            "key2": "value2",
            "level": "warning",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_logging_as_json_enabling_via_env(self, capfd: CaptureFixture, monkeypatch: MonkeyPatch) -> None:
        monkeypatch.setenv("HAYSTACK_LOGGING_USE_JSON", "true")
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err
        parsed_output = json.loads(output)  # should not raise an error

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key1": "value1",
            "key2": "value2",
            "level": "warning",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_logging_of_contextvars(
        self, capfd: CaptureFixture, monkeypatch: MonkeyPatch, set_context_var_key: str
    ) -> None:
        monkeypatch.setenv("HAYSTACK_LOGGING_USE_JSON", "true")
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err
        parsed_output = json.loads(output)  # should not raise an error

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key1": "value1",
            "key2": "value2",
            set_context_var_key: "value",
            "level": "warning",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_logging_exceptions_json(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = logging.getLogger("haystack.test_logging")

        def function_that_raises_and_adds_to_stack_trace():
            my_local_variable = "my_local_variable"  # noqa: F841
            raise ValueError("This is an error")

        try:
            function_that_raises_and_adds_to_stack_trace()
        except ValueError:
            logger.exception("An error happened ")

        # Use `capfd` to capture the output of the final structlog rendering result
        output = capfd.readouterr().err
        parsed_output = json.loads(output)
        assert parsed_output == {
            "event": "An error happened ",
            "level": "error",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
            "exception": [
                {
                    "exc_notes": [],
                    "exc_type": "ValueError",
                    "exc_value": "This is an error",
                    "exceptions": [],
                    "syntax_error": None,
                    "is_cause": False,
                    "is_group": False,
                    "frames": [
                        {
                            "filename": str(Path.cwd() / "test" / "test_logging.py"),
                            "lineno": ANY,  # otherwise the test breaks if you add a line :-)
                            "name": "test_logging_exceptions_json",
                        },
                        {
                            "filename": str(Path.cwd() / "test" / "test_logging.py"),
                            "lineno": ANY,  # otherwise the test breaks if you add a line :-)
                            "name": "function_that_raises_and_adds_to_stack_trace",
                        },
                    ],
                }
            ],
        }


class TestLogTraceCorrelation:
    def test_trace_log_correlation_python_logs_with_console_rendering(
        self, spying_tracer: SpyingTracer, capfd: CaptureFixture
    ) -> None:
        haystack_logging.configure_logging(use_json=False)

        with spying_tracer.trace("test-operation"):
            logger = logging.getLogger("haystack.test_logging")
            logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        output = capfd.readouterr().err
        assert "trace_id" not in output

    def test_trace_log_correlation_python_logs(self, spying_tracer: SpyingTracer, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        with spying_tracer.trace("test-operation") as span:
            logger = logging.getLogger("haystack.test_logging")
            logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        output = capfd.readouterr().err
        parsed_output = json.loads(output)

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key1": "value1",
            "key2": "value2",
            "level": "warning",
            "timestamp": ANY,
            "trace_id": span.trace_id,
            "span_id": span.span_id,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_trace_log_correlation_no_span(self, spying_tracer: SpyingTracer, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = logging.getLogger("haystack.test_logging")

        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        output = capfd.readouterr().err
        parsed_output = json.loads(output)

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key1": "value1",
            "key2": "value2",
            "level": "warning",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_trace_log_correlation_no_tracer(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = logging.getLogger("haystack.test_logging")

        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        output = capfd.readouterr().err
        parsed_output = json.loads(output)

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key1": "value1",
            "key2": "value2",
            "level": "warning",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }


class TestCompositeLogger:
    def test_correct_stack_level_with_stdlib_rendering(
        self, monkeypatch: MonkeyPatch, capfd: CaptureFixture, caplog: LogCaptureFixture
    ) -> None:
        monkeypatch.setenv("HAYSTACK_LOGGING_IGNORE_STRUCTLOG", "true")
        haystack_logging.configure_logging()

        logger = logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        # the pytest fixture caplog only captures logs being rendered from the stdlib logging module
        assert caplog.messages == ["Hello, structured logging!"]
        assert caplog.records[0].name == "haystack.test_logging"

        # Nothing should be captured by capfd since structlog is not configured
        assert capfd.readouterr().err == ""

    def test_correct_stack_level_with_consoler_rendering(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=False)

        logger = haystack_logging.getLogger("haystack.test_logging")
        logger.warning("Hello, structured logging!", extra={"key1": "value1", "key2": "value2"})

        output = capfd.readouterr().err
        assert "haystack.test_logging" in output

    @pytest.mark.parametrize(
        "method, expected_level",
        [
            ("debug", "debug"),
            ("info", "info"),
            ("warning", "warning"),
            ("error", "error"),
            ("fatal", "critical"),
            ("exception", "error"),
            ("critical", "critical"),
        ],
    )
    def test_various_levels(self, capfd: LogCaptureFixture, method: str, expected_level: str) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = haystack_logging.getLogger("haystack.test_logging")

        logger.setLevel(logging.DEBUG)

        getattr(logger, method)("Hello, structured {key}!", key="logging", key1="value1", key2="value2")

        output = capfd.readouterr().err
        parsed_output = json.loads(output)  # should not raise an error

        assert parsed_output == {
            "event": "Hello, structured logging!",
            "key": "logging",
            "key1": "value1",
            "key2": "value2",
            "level": expected_level,
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_log(self, capfd: LogCaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = haystack_logging.getLogger("haystack.test_logging")
        logger.setLevel(logging.DEBUG)

        logger.log(logging.DEBUG, "Hello, structured '{key}'!", key="logging", key1="value1", key2="value2")

        output = capfd.readouterr().err
        parsed_output = json.loads(output)

        assert parsed_output == {
            "event": "Hello, structured 'logging'!",
            "key": "logging",
            "key1": "value1",
            "key2": "value2",
            "level": "debug",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_log_json_content(self, capfd: LogCaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = haystack_logging.getLogger("haystack.test_logging")
        logger.setLevel(logging.DEBUG)

        logger.log(logging.DEBUG, 'Hello, structured: {"key": "value"}', key="logging", key1="value1", key2="value2")

        output = capfd.readouterr().err
        parsed_output = json.loads(output)

        assert parsed_output == {
            "event": 'Hello, structured: {"key": "value"}',
            "key": "logging",
            "key1": "value1",
            "key2": "value2",
            "level": "debug",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    def test_log_with_string_cast(self, capfd: LogCaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = haystack_logging.getLogger("haystack.test_logging")
        logger.setLevel(logging.DEBUG)

        logger.log(logging.DEBUG, "Hello, structured '{key}'!", key=LogCaptureFixture, key1="value1", key2="value2")

        output = capfd.readouterr().err
        parsed_output = json.loads(output)

        assert parsed_output == {
            "event": "Hello, structured '<class '_pytest.logging.LogCaptureFixture'>'!",
            "key": "<class '_pytest.logging.LogCaptureFixture'>",
            "key1": "value1",
            "key2": "value2",
            "level": "debug",
            "timestamp": ANY,
            "lineno": ANY,
            "module": "haystack.test_logging",
        }

    @pytest.mark.parametrize(
        "method, expected_level",
        [
            ("debug", "debug"),
            ("info", "info"),
            ("warning", "warning"),
            ("error", "error"),
            ("fatal", "critical"),
            ("exception", "exception"),
            ("critical", "critical"),
        ],
    )
    def test_haystack_logger_with_positional_args(self, method: str, expected_level: str) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = haystack_logging.getLogger("haystack.test_logging")
        logger.setLevel(logging.DEBUG)

        with pytest.raises(TypeError):
            getattr(logger, method)("Hello, structured logging %s!", "logging")

    @pytest.mark.parametrize(
        "method, expected_level",
        [
            ("debug", "debug"),
            ("info", "info"),
            ("warning", "warning"),
            ("error", "error"),
            ("fatal", "critical"),
            ("exception", "exception"),
            ("critical", "critical"),
        ],
    )
    def test_haystack_logger_with_old_interpolation(self, method: str, expected_level: str) -> None:
        haystack_logging.configure_logging(use_json=True)

        logger = haystack_logging.getLogger("haystack.test_logging")
        logger.setLevel(logging.DEBUG)

        # does not raise - hence we need to check this separately
        getattr(logger, method)("Hello, structured logging %s!", key="logging")

    def test_that_haystack_logger_is_used(self) -> None:
        """Forces the usage of the Haystack logger instead of the standard library logger."""
        allowed_list = [
            Path("haystack") / "logging.py",
            # Deliberately uses the stdlib logger: tenacity's before_log/after_log call it with positional args, which
            # the Haystack logger rejects.
            Path("haystack") / "utils" / "requests_utils.py",
        ]
        for root, _, files in os.walk("haystack"):
            for file in files:
                path = Path(root) / file

                if not path.suffix.endswith(".py"):
                    continue

                if path in allowed_list:
                    continue

                content = path.read_text(encoding="utf-8")

                # that looks like somebody is using our standard logger
                if " logging.getLogger" in content:
                    haystack_logger_in_content = " haystack import logging" in content or ", logging" in content
                    assert haystack_logger_in_content, (
                        f"{path} doesn't use the Haystack logger. Please use the Haystack logger instead of the "
                        f"standard library logger and add plenty of keyword arguments."
                    )


class TestLoggingScope:
    """
    Haystack is a library that usually runs next to other services in the same process (e.g. a web server, or another
    app's logging setup). These tests pin down that `configure_logging` only touches Haystack's own loggers and does
    not reformat or hijack the logs of everything else in the process.
    """

    def test_handler_is_attached_to_haystack_namespaces_and_not_root(self) -> None:
        haystack_logging.configure_logging(use_json=True)

        # Haystack's own namespace and the ones used by integration and experimental packages get the handler.
        for name in ["haystack", "haystack_integrations", "haystack_experimental"]:
            logger = logging.getLogger(name)
            assert any(getattr(h, "name", None) == "HaystackLoggingHandler" for h in logger.handlers)

        # The root logger - shared by every other library/service in the process - is left untouched.
        assert not any(getattr(h, "name", None) == "HaystackLoggingHandler" for h in logging.root.handlers)

    def test_integrations_logs_are_formatted_by_haystack(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        logging.getLogger("haystack_integrations.components.demo").warning("a log line from an integration")

        output = capfd.readouterr().err
        assert json.loads(output)["event"] == "a log line from an integration"

    def test_other_services_logs_are_not_reformatted_by_haystack(
        self, capfd: CaptureFixture, restore_named_loggers: Callable[[str], logging.Logger]
    ) -> None:
        # Stand-in for another service that configured its own plain-text logging (e.g. uvicorn) before Haystack runs.
        other_service = restore_named_loggers("some_other_service")
        other_service.handlers = [logging.StreamHandler()]
        other_service.setLevel(logging.INFO)
        other_service.propagate = False

        haystack_logging.configure_logging(use_json=True)

        other_service.info("a log line from another service")
        logging.getLogger("haystack.test_logging").warning("a log line from haystack")

        lines = [line for line in capfd.readouterr().err.splitlines() if line.strip()]

        # Haystack formats its own record as JSON ...
        haystack_json = [json.loads(line) for line in lines if line.startswith("{") and "from haystack" in line]
        assert any(record["event"] == "a log line from haystack" for record in haystack_json)

        # ... but the other service's record stays exactly as that service rendered it - plain text, not Haystack JSON.
        plain_lines = [line for line in lines if "from another service" in line]
        assert plain_lines
        for line in plain_lines:
            with pytest.raises(json.JSONDecodeError):
                json.loads(line)

    def test_legacy_root_behavior_still_available_via_empty_logger_name(self, capfd: CaptureFixture) -> None:
        # Opt back into the old behavior of formatting *every* record in the process.
        haystack_logging.configure_logging(use_json=True, logger_name="")

        assert any(getattr(h, "name", None) == "HaystackLoggingHandler" for h in logging.root.handlers)

        # A non-Haystack logger is now formatted by Haystack because the handler sits on the root logger.
        logging.getLogger("some_other_service").warning("formatted by haystack")
        output = capfd.readouterr().err
        assert json.loads(output)["event"] == "formatted by haystack"

    def test_switching_to_root_logger_does_not_duplicate_haystack_logs(self, capfd: CaptureFixture) -> None:
        # Mirror what `import haystack` does: install the handler on the Haystack namespaces.
        haystack_logging.configure_logging(use_json=True)
        # Then opt into root formatting. The namespace handlers must be removed - otherwise a `haystack.*` record is
        # emitted twice (once by the namespace handler, once by the root handler via propagation).
        haystack_logging.configure_logging(use_json=True, logger_name="")

        # The Haystack namespaces no longer carry our handler.
        for name in ["haystack", "haystack_integrations", "haystack_experimental"]:
            handlers = logging.getLogger(name).handlers
            assert not any(getattr(h, "name", None) == "HaystackLoggingHandler" for h in handlers)

        logging.getLogger("haystack.demo").warning("dup check")
        emitted = [line for line in capfd.readouterr().err.splitlines() if "dup check" in line]
        assert len(emitted) == 1

    def test_propagate_is_true_by_default(self, caplog: LogCaptureFixture) -> None:
        # The default keeps records flowing to ancestor loggers, so tooling that captures via the root logger (such as
        # pytest's `caplog`) keeps working.
        haystack_logging.configure_logging(use_json=True)
        assert logging.getLogger("haystack").propagate is True

        with caplog.at_level(logging.WARNING, logger="haystack.test_logging"):
            logging.getLogger("haystack.test_logging").warning("captured via propagation")
        assert "captured via propagation" in caplog.text

    def test_propagate_false_stops_records_from_reaching_root(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True, propagate=False)
        assert logging.getLogger("haystack").propagate is False

        # A plain handler on the root logger should NOT see Haystack's records when propagation is disabled.
        root_handler = logging.StreamHandler()
        root_handler.setFormatter(logging.Formatter("ROOT | %(message)s"))
        logging.root.addHandler(root_handler)
        try:
            logging.getLogger("haystack.test_logging").warning("haystack owns this line")
        finally:
            logging.root.removeHandler(root_handler)

        output = capfd.readouterr().err
        # Formatted once by Haystack (JSON), never by the root handler.
        assert "ROOT |" not in output
        assert json.loads(output)["event"] == "haystack owns this line"


class TestDynamicLogLevel:
    """
    `configure_logging` runs at import time (in `haystack/__init__.py`), long before an application sets its desired
    log level. These tests pin down that a log level set *after* `configure_logging` is still respected, instead of
    being frozen to whatever the root level happened to be at import time.
    """

    def test_structlog_native_logger_respects_level_lowered_after_configure(self, capfd: CaptureFixture) -> None:
        # Mirror the real ordering: configure while the level is high ...
        logging.getLogger("haystack").setLevel(logging.WARNING)
        haystack_logging.configure_logging(use_json=True)

        # ... then have the application lower the level afterwards.
        logging.getLogger("haystack").setLevel(logging.DEBUG)

        structlog.get_logger("haystack.native_dynamic_level").debug("debug emitted after lowering the level")

        assert "debug emitted after lowering the level" in capfd.readouterr().err

    def test_structlog_native_logger_still_filters_below_level(self, capfd: CaptureFixture) -> None:
        logging.getLogger("haystack").setLevel(logging.INFO)
        haystack_logging.configure_logging(use_json=True)

        structlog.get_logger("haystack.native_filtered_level").debug("debug below the configured level")

        assert "debug below the configured level" not in capfd.readouterr().err


class TestStructlogConfigIsPreserved:
    """
    `structlog.configure` writes to a single process-global configuration. These tests pin down that merely importing
    Haystack (which calls `configure_logging(configure_structlog=False)`) does not overwrite a structlog configuration
    that the host application already set up, while an explicit call still takes over.
    """

    def test_without_configure_structlog_existing_config_is_preserved(self, restore_structlog_config: None) -> None:
        # Stand-in for the host application configuring structlog before Haystack is imported/configured.
        structlog.reset_defaults()
        structlog.configure(processors=[_sentinel_processor])
        haystack_logger = logging.getLogger("haystack")
        haystack_logger.handlers = []

        haystack_logging.configure_logging(configure_structlog=False)

        # The application's structlog configuration is left untouched ...
        assert structlog.get_config()["processors"] == [_sentinel_processor]
        # ... but we still install our scoped handler so Haystack's own logs are formatted (independently of import
        # ordering relative to the application's structlog setup).
        assert any(getattr(h, "name", None) == "HaystackLoggingHandler" for h in haystack_logger.handlers)

    def test_configure_structlog_takes_over_existing_config(self, restore_structlog_config: None) -> None:
        structlog.reset_defaults()
        structlog.configure(processors=[_sentinel_processor])
        haystack_logger = logging.getLogger("haystack")
        haystack_logger.handlers = []

        haystack_logging.configure_logging(use_json=True, configure_structlog=True)

        assert structlog.get_config()["processors"] != [_sentinel_processor]
        assert any(getattr(h, "name", None) == "HaystackLoggingHandler" for h in haystack_logger.handlers)

    def test_installs_handler_without_configuring_structlog(self, restore_structlog_config: None) -> None:
        # The real import-time situation: nobody configured structlog yet. We install our scoped handler so Haystack's
        # own logs are formatted, but we must NOT touch the process-global structlog configuration (which would
        # reformat the host application's own native structlog loggers).
        structlog.reset_defaults()
        haystack_logger = logging.getLogger("haystack")
        haystack_logger.handlers = []
        assert not structlog.is_configured()

        haystack_logging.configure_logging(configure_structlog=False)

        # Our scoped handler is installed ...
        assert any(getattr(h, "name", None) == "HaystackLoggingHandler" for h in haystack_logger.handlers)
        # ... but the global structlog configuration is left untouched.
        assert not structlog.is_configured()


class TestGetLoggerIsIdempotent:
    """
    `logging.getLogger(name)` returns a process-wide singleton. `haystack.logging.getLogger` patches that shared
    object in place, so calling it more than once for the same name (different modules, re-imports, ...) must not wrap
    the already-wrapped methods again. The user-visible symptom of re-wrapping is that the message is run through
    `str.format` once per wrap, so a field value that itself contains `{...}` gets re-interpolated.
    """

    def test_repeated_get_logger_interpolates_the_message_exactly_once(self, capfd: CaptureFixture) -> None:
        haystack_logging.configure_logging(use_json=True)

        # Two modules grabbing the same logger name is the realistic trigger for re-wrapping.
        haystack_logging.getLogger("haystack.idempotency_test")
        logger = haystack_logging.getLogger("haystack.idempotency_test")
        logger.setLevel(logging.INFO)

        # `a`'s value contains a `{b}` placeholder. With a single interpolation it must be left as-is; a second
        # interpolation would expand it using `b` and leak "SECRET" into the message.
        logger.info("Hello {a}", a="{b}", b="SECRET")

        parsed_output = json.loads(capfd.readouterr().err)
        assert parsed_output["event"] == "Hello {b}"
        assert "SECRET" not in parsed_output["event"]

    def test_repeated_get_logger_does_not_rewrap_methods(self) -> None:
        haystack_logging.getLogger("haystack.idempotency_identity_test")
        # Capture the patched methods after the first call, before the second one runs.
        patched = logging.getLogger("haystack.idempotency_identity_test")
        debug_after_first = patched.debug
        make_record_after_first = patched.makeRecord

        haystack_logging.getLogger("haystack.idempotency_identity_test")

        # The second call must leave the already-patched methods in place, not wrap a fresh layer on top.
        assert patched.debug is debug_after_first
        assert patched.makeRecord is make_record_after_first


class TestFindCallerMatchesStructlog:
    """
    `_patch_structlog_call_information` mirrors structlog's `_FixedFindCallerLogger.findCaller`, only adding
    `haystack.logging` to the ignored frames. structlog itself does not guard the frame lookup, so neither do we: any
    error must propagate as-is instead of being swallowed and printed to stdout.
    """

    def test_find_caller_does_not_print_or_mask_errors(self, capsys: CaptureFixture, monkeypatch: MonkeyPatch) -> None:
        # Force the frame lookup to fail. It is imported inside `_patch_structlog_call_information`, so we patch the
        # module attribute before patching the logger.
        def boom(*args: object, **kwargs: object) -> tuple:
            raise RuntimeError("frame lookup failed")

        monkeypatch.setattr(structlog._frames, "_find_first_app_frame_and_name", boom)

        logger = structlog.stdlib._FixedFindCallerLogger("haystack.find_caller_test")
        haystack_logging._patch_structlog_call_information(logger)

        # The original error must propagate (not be masked by a NameError on an unbound `f`) ...
        with pytest.raises(RuntimeError, match="frame lookup failed"):
            logger.findCaller()

        # ... and nothing must be written to stdout.
        assert capsys.readouterr().out == ""