import sys
from io import StringIO

import pytest

import logbook

from .utils import capturing_stderr_context, missing, require_module


@require_module("jinja2")
def test_jinja_formatter(logger):
    from logbook.more import JinjaFormatter

    fmter = JinjaFormatter("{{ record.channel }}/{{ record.level_name }}")
    handler = logbook.TestHandler()
    handler.formatter = fmter
    with handler:
        logger.info("info")
    assert "testlogger/INFO" in handler.formatted_records


@missing("jinja2")
def test_missing_jinja2():
    from logbook.more import JinjaFormatter

    # check the RuntimeError is raised
    with pytest.raises(RuntimeError):
        JinjaFormatter("dummy")


def test_colorizing_support(logger):
    from logbook.more import ColorizedStderrHandler

    class TestColorizingHandler(ColorizedStderrHandler):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self._obj_stream = StringIO()

        @property
        def stream(self):
            return self._obj_stream

    with TestColorizingHandler(format_string="{record.message}") as handler:
        handler.force_color()
        logger.error("An error")
        logger.warn("A warning")
        logger.debug("A debug message")
        lines = handler.stream.getvalue().rstrip("\n").splitlines()
        assert lines == [
            "\x1b[31;01mAn error\x1b[39;49;00m",
            "\x1b[33;01mA warning\x1b[39;49;00m",
            "\x1b[37mA debug message\x1b[39;49;00m",
        ]

    with TestColorizingHandler(format_string="{record.message}") as handler:
        handler.forbid_color()
        logger.error("An error")
        logger.warn("A warning")
        logger.debug("A debug message")
        lines = handler.stream.getvalue().rstrip("\n").splitlines()
        assert lines == ["An error", "A warning", "A debug message"]


def test_tagged(default_handler):
    from logbook.more import TaggingHandler, TaggingLogger

    stream = StringIO()
    second_handler = logbook.StreamHandler(stream)

    logger = TaggingLogger("name", ["cmd"])
    handler = TaggingHandler(
        dict(
            info=default_handler,
            cmd=second_handler,
            both=[default_handler, second_handler],
        )
    )
    handler.bubble = False

    with handler:
        with capturing_stderr_context() as captured:
            logger.log("info", "info message")
            logger.log("both", "all message")
            logger.cmd("cmd message")

    stderr = captured.getvalue()

    assert "info message" in stderr
    assert "all message" in stderr
    assert "cmd message" not in stderr

    stringio = stream.getvalue()

    assert "info message" not in stringio
    assert "all message" in stringio
    assert "cmd message" in stringio


def test_tagging_logger(default_handler):
    from logbook import StderrHandler
    from logbook.more import TaggingLogger

    logger = TaggingLogger("tagged", ["a", "b"])
    handler = StderrHandler(format_string="{record.msg}|{record.extra[tags]}")

    with handler:
        with capturing_stderr_context() as captured:
            logger.a("a")
            logger.b("b")

    stderr = captured.getvalue()

    assert "a|['a']" in stderr
    assert "a|['b']" not in stderr
    assert "b|['b']" in stderr
    assert "b|['a']" not in stderr


def test_external_application_handler(tmpdir, logger):
    from logbook.more import ExternalApplicationHandler as Handler

    fn = tmpdir.join("tempfile")
    handler = Handler(
        [
            sys.executable,
            "-c",
            r"""if 1:
    f = open({tempfile}, 'w')
    try:
        f.write('{{record.message}}\n')
    finally:
        f.close()
    """.format(
                tempfile=repr(str(fn))
            ),
        ]
    )
    with handler:
        logger.error("this is a really bad idea")
    with fn.open() as rf:
        contents = rf.read().strip()
    assert contents == "this is a really bad idea"


def test_exception_handler(logger):
    from logbook.more import ExceptionHandler

    with ExceptionHandler(ValueError):
        with pytest.raises(ValueError) as caught:
            logger.info("here i am")
    assert "INFO: testlogger: here i am" in caught.value.args[0]


def test_exception_handler_specific_level(logger):
    from logbook.more import ExceptionHandler

    with logbook.TestHandler() as test_handler:
        with pytest.raises(ValueError) as caught:
            with ExceptionHandler(ValueError, level="WARNING"):
                logger.info("this is irrelevant")
                logger.warn("here i am")
        assert "WARNING: testlogger: here i am" in caught.value.args[0]
    assert "this is irrelevant" in test_handler.records[0].message


def test_dedup_handler(logger):
    from logbook.more import DedupHandler

    with logbook.TestHandler() as test_handler:
        with DedupHandler():
            logger.info("foo")
            logger.info("bar")
            logger.info("foo")
    assert 2 == len(test_handler.records)
    assert "message repeated 2 times: foo" in test_handler.records[0].message
    assert "message repeated 1 times: bar" in test_handler.records[1].message


class TestRiemannHandler:
    @require_module("riemann_client")
    def test_happy_path(self, logger):
        from logbook.more import RiemannHandler

        riemann_handler = RiemannHandler(
            "127.0.0.1", 5555, message_type="test", level=logbook.INFO
        )
        null_handler = logbook.NullHandler()
        with null_handler.applicationbound():
            with riemann_handler:
                logger.error("Something bad has happened")
                try:
                    raise RuntimeError("For example, a RuntimeError")
                except Exception as ex:
                    logger.exception(ex)
                logger.info("But now it is ok")

        q = riemann_handler.queue
        assert len(q) == 3
        error_event = q[0]
        assert error_event["state"] == "error"
        exc_event = q[1]
        assert exc_event["description"] == "For example, a RuntimeError"
        info_event = q[2]
        assert info_event["state"] == "ok"

    @require_module("riemann_client")
    def test_incorrect_type(self):
        from logbook.more import RiemannHandler

        with pytest.raises(RuntimeError):
            RiemannHandler("127.0.0.1", 5555, message_type="fancy_type")

    @require_module("riemann_client")
    def test_flush(self, logger):
        from logbook.more import RiemannHandler

        riemann_handler = RiemannHandler(
            "127.0.0.1",
            5555,
            message_type="test",
            flush_threshold=2,
            level=logbook.INFO,
        )
        null_handler = logbook.NullHandler()
        with null_handler.applicationbound():
            with riemann_handler:
                logger.info("Msg #1")
                logger.info("Msg #2")
                logger.info("Msg #3")

        q = riemann_handler.queue
        assert len(q) == 1
        assert q[0]["description"] == "Msg #3"