diff --git a/news/440.feat.md b/news/440.feat.md new file mode 100644 index 00000000..da14b0ed --- /dev/null +++ b/news/440.feat.md @@ -0,0 +1 @@ +Added integration with the built-in logging package. diff --git a/src/cleo/logging/__init__.py b/src/cleo/logging/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/cleo/logging/cleo_handler.py b/src/cleo/logging/cleo_handler.py new file mode 100644 index 00000000..344c3e76 --- /dev/null +++ b/src/cleo/logging/cleo_handler.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import logging + +from logging import LogRecord +from typing import TYPE_CHECKING +from typing import ClassVar +from typing import cast + +from cleo.exceptions import CleoUserError +from cleo.io.outputs.output import Verbosity +from cleo.ui.exception_trace.component import ExceptionTrace + + +if TYPE_CHECKING: + from cleo.io.outputs.output import Output + + +class CleoFilter: + def __init__(self, output: Output): + self.output = output + + @property + def current_loglevel(self) -> int: + verbosity_mapping: dict[Verbosity, int] = { + Verbosity.QUIET: logging.CRITICAL, # Nothing gets emitted to the output anyway + Verbosity.NORMAL: logging.WARNING, + Verbosity.VERBOSE: logging.INFO, + Verbosity.VERY_VERBOSE: logging.DEBUG, + Verbosity.DEBUG: logging.DEBUG, + } + return verbosity_mapping[self.output.verbosity] + + def filter(self, record: LogRecord) -> bool: + return record.levelno >= self.current_loglevel + + +class CleoHandler(logging.Handler): + """ + A handler class which writes logging records, appropriately formatted, + to a Cleo output stream. + """ + + tags: ClassVar[dict[str, str]] = { + "CRITICAL": "", + "ERROR": "", + "WARNING": "", + "DEBUG": "", + } + + def __init__(self, output: Output): + super().__init__() + self.output = output + self.addFilter(CleoFilter(output)) + + def emit(self, record: logging.LogRecord) -> None: + """ + Emit a record. + + If a formatter is specified, it is used to format the record. + The record is then written to the output with a trailing newline. If + exception information is present, it is formatted using + traceback.print_exception and appended to the stream. If the stream + has an 'encoding' attribute, it is used to determine how to do the + output to the stream. + """ + + try: + msg = self.tags.get(record.levelname, "") + self.format(record) + "" + self.output.write(msg, new_line=True) + if record.exc_info: + _type, error, traceback = record.exc_info + simple = not self.output.is_verbose() or isinstance( + error, CleoUserError + ) + error = cast(Exception, error) + trace = ExceptionTrace(error) + trace.render(self.output, simple) + + except Exception: + self.handleError(record) diff --git a/tests/fixtures/foo4_command.py b/tests/fixtures/foo4_command.py new file mode 100644 index 00000000..d53c18ed --- /dev/null +++ b/tests/fixtures/foo4_command.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import logging + +from typing import TYPE_CHECKING +from typing import ClassVar + +from cleo.commands.command import Command +from cleo.helpers import option + + +if TYPE_CHECKING: + from cleo.io.inputs.option import Option + + +_logger = logging.getLogger(__file__) + + +def log_stuff() -> None: + _logger.debug("This is an debug log record") + _logger.info("This is an info log record") + _logger.warning("This is an warning log record") + _logger.error("This is an error log record") + + +def log_exception() -> None: + try: + raise RuntimeError("This is an exception that I raised") + except RuntimeError as e: + _logger.exception(e) + + +class Foo4Command(Command): + name = "foo4" + + description = "The foo4 bar command" + + aliases: ClassVar[list[str]] = ["foo4"] + + options: ClassVar[list[Option]] = [option("exception")] + + def handle(self) -> int: + if self.option("exception"): + log_exception() + else: + log_stuff() + + return 0 diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 00000000..45123eca --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import logging + +import pytest + +from cleo.application import Application +from cleo.logging.cleo_handler import CleoHandler +from cleo.testers.application_tester import ApplicationTester +from tests.fixtures.foo4_command import Foo4Command + + +@pytest.fixture +def app() -> Application: + app = Application() + cmd = Foo4Command() + app.add(cmd) + app._default_command = cmd.name + return app + + +@pytest.fixture +def tester(app: Application) -> ApplicationTester: + app.catch_exceptions(False) + return ApplicationTester(app) + + +@pytest.fixture +def root_logger() -> logging.Logger: + root = logging.getLogger() + root.setLevel(logging.NOTSET) + return root + + +def test_cleohandler_normal( + tester: ApplicationTester, + root_logger: logging.Logger, +) -> None: + handler = CleoHandler(tester.io.output) + root_logger.addHandler(handler) + + status_code = tester.execute("") + + expected = "This is an warning log record\n" "This is an error log record\n" + + assert status_code == 0 + assert tester.io.fetch_output() == expected + + +def test_cleohandler_quiet( + tester: ApplicationTester, + root_logger: logging.Logger, +) -> None: + handler = CleoHandler(tester.io.output) + root_logger.addHandler(handler) + + status_code = tester.execute("-q") + + assert status_code == 0 + assert tester.io.fetch_output() == "" + + +def test_cleohandler_verbose( + tester: ApplicationTester, + root_logger: logging.Logger, +) -> None: + handler = CleoHandler(tester.io.output) + root_logger.addHandler(handler) + + status_code = tester.execute("-v") + + expected = ( + "This is an info log record\n" + "This is an warning log record\n" + "This is an error log record\n" + ) + + assert status_code == 0 + assert tester.io.fetch_output() == expected + + +def test_cleohandler_very_verbose( + tester: ApplicationTester, + root_logger: logging.Logger, +) -> None: + handler = CleoHandler(tester.io.output) + root_logger.addHandler(handler) + + status_code = tester.execute("-vv") + + expected = ( + "This is an debug log record\n" + "This is an info log record\n" + "This is an warning log record\n" + "This is an error log record\n" + ) + + assert status_code == 0 + assert tester.io.fetch_output() == expected + + +def test_cleohandler_exception_normal( + tester: ApplicationTester, + root_logger: logging.Logger, +) -> None: + handler = CleoHandler(tester.io.output) + root_logger.addHandler(handler) + + status_code = tester.execute("--exception") + + assert status_code == 0 + lines = tester.io.fetch_output().splitlines() + + assert len(lines) == 7 + assert lines[0] == "This is an exception that I raised" + + +def test_cleohandler_exception_verbose( + tester: ApplicationTester, + root_logger: logging.Logger, +) -> None: + handler = CleoHandler(tester.io.output) + root_logger.addHandler(handler) + + status_code = tester.execute("-v --exception") + + assert status_code == 0 + lines = tester.io.fetch_output().splitlines() + + assert len(lines) == 20 + assert lines[0] == "This is an exception that I raised" + + +def test_cleohandler_exception_very_verbose( + tester: ApplicationTester, + root_logger: logging.Logger, +) -> None: + handler = CleoHandler(tester.io.output) + root_logger.addHandler(handler) + + status_code = tester.execute("-vv --exception") + + assert status_code == 0 + lines = tester.io.fetch_output().splitlines() + + assert len(lines) == 20 + assert lines[0] == "This is an exception that I raised"