diff --git a/scrapy/utils/log.py b/scrapy/utils/log.py index 60ac44833..b1f1c24d5 100644 --- a/scrapy/utils/log.py +++ b/scrapy/utils/log.py @@ -1,287 +1,287 @@ -from __future__ import annotations - -import logging -import pprint -import sys -from collections.abc import MutableMapping -from logging.config import dictConfig -from typing import TYPE_CHECKING, Any, cast - -from twisted.internet import asyncioreactor -from twisted.python import log as twisted_log -from twisted.python.failure import Failure - -import scrapy -from scrapy.settings import Settings -from scrapy.utils.versions import get_versions - -if TYPE_CHECKING: - from types import TracebackType - - from scrapy.crawler import Crawler - from scrapy.logformatter import LogFormatterResult - - -logger = logging.getLogger(__name__) - - -def failure_to_exc_info( - failure: Failure, -) -> tuple[type[BaseException], BaseException, TracebackType | None] | None: - """Extract exc_info from Failure instances""" - if isinstance(failure, Failure): - assert failure.type - assert failure.value - return ( - failure.type, - failure.value, - cast("TracebackType | None", failure.getTracebackObject()), - ) - return None - - -class TopLevelFormatter(logging.Filter): - """Keep only top level loggers' name (direct children from root) from - records. - - This filter will replace Scrapy loggers' names with 'scrapy'. This mimics - the old Scrapy log behaviour and helps shortening long names. - - Since it can't be set for just one logger (it won't propagate for its - children), it's going to be set in the root handler, with a parametrized - ``loggers`` list where it should act. - """ - - def __init__(self, loggers: list[str] | None = None): - super().__init__() - self.loggers: list[str] = loggers or [] - - def filter(self, record: logging.LogRecord) -> bool: - if any(record.name.startswith(logger + ".") for logger in self.loggers): - record.name = record.name.split(".", 1)[0] - return True - - -DEFAULT_LOGGING = { - "version": 1, - "disable_existing_loggers": False, - "loggers": { - "filelock": { - "level": "ERROR", - }, - "hpack": { - "level": "ERROR", - }, - "httpcore": { - "level": "ERROR", - }, - "httpx": { - "level": "WARNING", - }, - "scrapy": { - "level": "DEBUG", - }, - "twisted": { - "level": "ERROR", - }, - }, -} - - -def configure_logging( - settings: Settings | dict[str, Any] | None = None, - install_root_handler: bool = True, -) -> None: - """ - Initialize logging defaults for Scrapy. - - :param settings: settings used to create and configure a handler for the - root logger (default: None). - :type settings: dict, :class:`~scrapy.settings.Settings` object or ``None`` - - :param install_root_handler: whether to install root logging handler - (default: True) - :type install_root_handler: bool - - This function does: - - - Route warnings and twisted logging through Python standard logging - - Assign DEBUG and ERROR level to Scrapy and Twisted loggers respectively - - Route stdout to log if LOG_STDOUT setting is True - - When ``install_root_handler`` is True (default), this function also - creates a handler for the root logger according to given settings - (see :ref:`topics-logging-settings`). You can override default options - using ``settings`` argument. When ``settings`` is empty or None, defaults - are used. - """ - if not sys.warnoptions: - # Route warnings through python logging - logging.captureWarnings(True) - - observer = twisted_log.PythonLoggingObserver("twisted") - observer.start() - - dictConfig(DEFAULT_LOGGING) - - if isinstance(settings, dict) or settings is None: - settings = Settings(settings) - - if settings.getbool("LOG_STDOUT"): - sys.stdout = StreamLogger(logging.getLogger("stdout")) - - if install_root_handler: - install_scrapy_root_handler(settings) - - -_scrapy_root_handler: logging.Handler | None = None - - -def install_scrapy_root_handler(settings: Settings) -> None: - global _scrapy_root_handler # noqa: PLW0603 - - _uninstall_scrapy_root_handler() - logging.root.setLevel(logging.NOTSET) - _scrapy_root_handler = _get_handler(settings) - logging.root.addHandler(_scrapy_root_handler) - - -def _uninstall_scrapy_root_handler() -> None: - global _scrapy_root_handler # noqa: PLW0603 - - if _scrapy_root_handler is None: - return - - if _scrapy_root_handler in logging.root.handlers: - logging.root.removeHandler(_scrapy_root_handler) - _scrapy_root_handler.close() - _scrapy_root_handler = None - - -def get_scrapy_root_handler() -> logging.Handler | None: - return _scrapy_root_handler - - -def _get_handler(settings: Settings) -> logging.Handler: - """Return a log handler object according to settings""" - filename = settings.get("LOG_FILE") - handler: logging.Handler - if filename: - mode = "a" if settings.getbool("LOG_FILE_APPEND") else "w" - encoding = settings.get("LOG_ENCODING") - handler = logging.FileHandler(filename, mode=mode, encoding=encoding) - elif settings.getbool("LOG_ENABLED"): - handler = logging.StreamHandler() - else: - handler = logging.NullHandler() - - formatter = logging.Formatter( - fmt=settings.get("LOG_FORMAT"), datefmt=settings.get("LOG_DATEFORMAT") - ) - handler.setFormatter(formatter) - handler.setLevel(settings.get("LOG_LEVEL")) - if settings.getbool("LOG_SHORT_NAMES"): - handler.addFilter(TopLevelFormatter(["scrapy"])) - return handler - - -def log_scrapy_info(settings: Settings) -> None: - logger.info( - "Scrapy %(version)s started (bot: %(bot)s)", - {"version": scrapy.__version__, "bot": settings["BOT_NAME"]}, - ) - software: list[str] = settings.getlist("LOG_VERSIONS") - if not software: - return - versions = pprint.pformat(dict(get_versions(software)), sort_dicts=False) - logger.info(f"Versions:\n{versions}") - - -def log_reactor_info() -> None: - from twisted.internet import reactor - - logger.debug("Using reactor: %s.%s", reactor.__module__, reactor.__class__.__name__) - if isinstance(reactor, asyncioreactor.AsyncioSelectorReactor): - logger.debug( - "Using asyncio event loop: %s.%s", - reactor._asyncioEventloop.__module__, - reactor._asyncioEventloop.__class__.__name__, - ) - - -class StreamLogger: - """Fake file-like stream object that redirects writes to a logger instance - - Taken from: - https://www.electricmonk.nl/log/2011/08/14/redirect-stdout-and-stderr-to-a-logger-in-python/ - """ - - def __init__(self, logger: logging.Logger, log_level: int = logging.INFO): - self.logger: logging.Logger = logger - self.log_level: int = log_level - self.linebuf: str = "" - - def write(self, buf: str) -> None: - for line in buf.rstrip().splitlines(): - self.logger.log(self.log_level, line.rstrip()) - - def flush(self) -> None: - for h in self.logger.handlers: - h.flush() - - -class LogCounterHandler(logging.Handler): - """Record log levels count into a crawler stats""" - - def __init__(self, crawler: Crawler, *args: Any, **kwargs: Any): - super().__init__(*args, **kwargs) - self.crawler: Crawler = crawler - - def emit(self, record: logging.LogRecord) -> None: - record_crawler = getattr(record, "crawler", None) - if record_crawler is not None and record_crawler is not self.crawler: - return - - record_spider = getattr(record, "spider", None) - record_spider_crawler = getattr(record_spider, "crawler", None) - if ( - record_spider_crawler is not None - and record_spider_crawler is not self.crawler - ): - return - - sname = f"log_count/{record.levelname}" - assert self.crawler.stats - self.crawler.stats.inc_value(sname) - - -def logformatter_adapter( - logkws: LogFormatterResult, -) -> tuple[int, str, dict[str, Any] | tuple[Any, ...]]: - """ - Helper that takes the dictionary output from the methods in LogFormatter - and adapts it into a tuple of positional arguments for logger.log calls, - handling backward compatibility as well. - """ - - level = logkws.get("level", logging.INFO) - message = logkws.get("msg") or "" - # NOTE: This also handles 'args' being an empty dict, that case doesn't - # play well in logger.log calls - args = cast("dict[str, Any]", logkws) if not logkws.get("args") else logkws["args"] - - return (level, message, args) - - -# LoggerAdapter is only parameterized since Python 3.11 -class SpiderLoggerAdapter(logging.LoggerAdapter): # type: ignore[type-arg] - def process( - self, msg: str, kwargs: MutableMapping[str, Any] - ) -> tuple[str, MutableMapping[str, Any]]: - """Method that augments logging with additional 'extra' data""" - if isinstance(kwargs.get("extra"), MutableMapping): - kwargs["extra"].update(self.extra) - else: - kwargs["extra"] = self.extra - - return msg, kwargs +from __future__ import annotations + +import logging +import pprint +import sys +from collections.abc import MutableMapping +from logging.config import dictConfig +from typing import TYPE_CHECKING, Any, cast + +from twisted.internet import asyncioreactor +from twisted.python import log as twisted_log +from twisted.python.failure import Failure + +import scrapy +from scrapy.settings import Settings +from scrapy.utils.versions import get_versions + +if TYPE_CHECKING: + from types import TracebackType + + from scrapy.crawler import Crawler + from scrapy.logformatter import LogFormatterResult + + +logger = logging.getLogger(__name__) + + +def failure_to_exc_info( + failure: Failure, +) -> tuple[type[BaseException], BaseException, TracebackType | None] | None: + """Extract exc_info from Failure instances""" + if isinstance(failure, Failure): + assert failure.type + assert failure.value + return ( + failure.type, + failure.value, + cast("TracebackType | None", failure.getTracebackObject()), + ) + return None + + +class TopLevelFormatter(logging.Filter): + """Keep only top level loggers' name (direct children from root) from + records. + + This filter will replace Scrapy loggers' names with 'scrapy'. This mimics + the old Scrapy log behaviour and helps shortening long names. + + Since it can't be set for just one logger (it won't propagate for its + children), it's going to be set in the root handler, with a parametrized + ``loggers`` list where it should act. + """ + + def __init__(self, loggers: list[str] | None = None): + super().__init__() + self.loggers: list[str] = loggers or [] + + def filter(self, record: logging.LogRecord) -> bool: + if any(record.name.startswith(logger + ".") for logger in self.loggers): + record.name = record.name.split(".", 1)[0] + return True + + +DEFAULT_LOGGING = { + "version": 1, + "disable_existing_loggers": False, + "loggers": { + "filelock": { + "level": "ERROR", + }, + "hpack": { + "level": "ERROR", + }, + "httpcore": { + "level": "ERROR", + }, + "httpx": { + "level": "WARNING", + }, + "scrapy": { + "level": "DEBUG", + }, + "twisted": { + "level": "ERROR", + }, + }, +} + + +def configure_logging( + settings: Settings | dict[str, Any] | None = None, + install_root_handler: bool = True, +) -> None: + """ + Initialize logging defaults for Scrapy. + + :param settings: settings used to create and configure a handler for the + root logger (default: None). + :type settings: dict, :class:`~scrapy.settings.Settings` object or ``None`` + + :param install_root_handler: whether to install root logging handler + (default: True) + :type install_root_handler: bool + + This function does: + + - Route warnings and twisted logging through Python standard logging + - Assign DEBUG and ERROR level to Scrapy and Twisted loggers respectively + - Route stdout to log if LOG_STDOUT setting is True + + When ``install_root_handler`` is True (default), this function also + creates a handler for the root logger according to given settings + (see :ref:`topics-logging-settings`). You can override default options + using ``settings`` argument. When ``settings`` is empty or None, defaults + are used. + """ + if not sys.warnoptions: + # Route warnings through python logging + logging.captureWarnings(True) + + observer = twisted_log.PythonLoggingObserver("twisted") + observer.start() + + dictConfig(DEFAULT_LOGGING) + + if isinstance(settings, dict) or settings is None: + settings = Settings(settings) + + if settings.getbool("LOG_STDOUT"): + sys.stdout = StreamLogger(logging.getLogger("stdout")) + + if install_root_handler: + install_scrapy_root_handler(settings) + + +_scrapy_root_handler: logging.Handler | None = None + + +def install_scrapy_root_handler(settings: Settings) -> None: + global _scrapy_root_handler # noqa: PLW0603 + + _uninstall_scrapy_root_handler() + logging.root.setLevel(logging.NOTSET) + _scrapy_root_handler = _get_handler(settings) + logging.root.addHandler(_scrapy_root_handler) + + +def _uninstall_scrapy_root_handler() -> None: + global _scrapy_root_handler # noqa: PLW0603 + + if _scrapy_root_handler is None: + return + + if _scrapy_root_handler in logging.root.handlers: + logging.root.removeHandler(_scrapy_root_handler) + _scrapy_root_handler.close() + _scrapy_root_handler = None + + +def get_scrapy_root_handler() -> logging.Handler | None: + return _scrapy_root_handler + + +def _get_handler(settings: Settings) -> logging.Handler: + """Return a log handler object according to settings""" + filename = settings.get("LOG_FILE") + handler: logging.Handler + if filename: + mode = "a" if settings.getbool("LOG_FILE_APPEND") else "w" + encoding = settings.get("LOG_ENCODING") + handler = logging.FileHandler(filename, mode=mode, encoding=encoding) + elif settings.getbool("LOG_ENABLED"): + handler = logging.StreamHandler() + else: + handler = logging.NullHandler() + + formatter = logging.Formatter( + fmt=settings.get("LOG_FORMAT"), datefmt=settings.get("LOG_DATEFORMAT") + ) + handler.setFormatter(formatter) + handler.setLevel(settings.get("LOG_LEVEL")) + if settings.getbool("LOG_SHORT_NAMES"): + handler.addFilter(TopLevelFormatter(["scrapy"])) + return handler + + +def log_scrapy_info(settings: Settings) -> None: + logger.info( + "Scrapy %(version)s started (bot: %(bot)s)", + {"version": scrapy.__version__, "bot": settings["BOT_NAME"]}, + ) + software: list[str] = settings.getlist("LOG_VERSIONS") + if not software: + return + versions = pprint.pformat(dict(get_versions(software)), sort_dicts=False) + logger.info(f"Versions:\n{versions}") + + +def log_reactor_info() -> None: + from twisted.internet import reactor + + logger.debug("Using reactor: %s.%s", reactor.__module__, reactor.__class__.__name__) + if isinstance(reactor, asyncioreactor.AsyncioSelectorReactor): + logger.debug( + "Using asyncio event loop: %s.%s", + reactor._asyncioEventloop.__module__, + reactor._asyncioEventloop.__class__.__name__, + ) + + +class StreamLogger: + """Fake file-like stream object that redirects writes to a logger instance + + Taken from: + https://www.electricmonk.nl/log/2011/08/14/redirect-stdout-and-stderr-to-a-logger-in-python/ + """ + + def __init__(self, logger: logging.Logger, log_level: int = logging.INFO): + self.logger: logging.Logger = logger + self.log_level: int = log_level + self.linebuf: str = "" + + def write(self, buf: str) -> None: + for line in buf.rstrip().splitlines(): + self.logger.log(self.log_level, line.rstrip()) + + def flush(self) -> None: + for h in self.logger.handlers: + h.flush() + + +class LogCounterHandler(logging.Handler): + """Record log levels count into a crawler stats""" + + def __init__(self, crawler: Crawler, *args: Any, **kwargs: Any): + super().__init__(*args, **kwargs) + self.crawler: Crawler = crawler + + def emit(self, record: logging.LogRecord) -> None: + record_crawler = getattr(record, "crawler", None) + if record_crawler is not None and record_crawler is not self.crawler: + return + + record_spider = getattr(record, "spider", None) + record_spider_crawler = getattr(record_spider, "crawler", None) + if ( + record_spider_crawler is not None + and record_spider_crawler is not self.crawler + ): + return + + sname = f"log_count/{record.levelname}" + assert self.crawler.stats + self.crawler.stats.inc_value(sname) + + +def logformatter_adapter( + logkws: LogFormatterResult, +) -> tuple[int, str, dict[str, Any] | tuple[Any, ...]]: + """ + Helper that takes the dictionary output from the methods in LogFormatter + and adapts it into a tuple of positional arguments for logger.log calls, + handling backward compatibility as well. + """ + + level = logkws.get("level", logging.INFO) + message = logkws.get("msg") or "" + # NOTE: This also handles 'args' being an empty dict, that case doesn't + # play well in logger.log calls + args = cast("dict[str, Any]", logkws) if not logkws.get("args") else logkws["args"] + + return (level, message, args) + + +# LoggerAdapter is only parameterized since Python 3.11 +class SpiderLoggerAdapter(logging.LoggerAdapter): # type: ignore[type-arg] + def process( + self, msg: str, kwargs: MutableMapping[str, Any] + ) -> tuple[str, MutableMapping[str, Any]]: + """Method that augments logging with additional 'extra' data""" + if isinstance(kwargs.get("extra"), MutableMapping): + kwargs["extra"].update(self.extra) + else: + kwargs["extra"] = self.extra + + return msg, kwargs diff --git a/tests/test_utils_log.py b/tests/test_utils_log.py index 04866fa02..2be98606e 100644 --- a/tests/test_utils_log.py +++ b/tests/test_utils_log.py @@ -1,327 +1,327 @@ -from __future__ import annotations - -import json -import logging -import re -import sys -from io import StringIO -from typing import TYPE_CHECKING, Any - -import pytest -from testfixtures import LogCapture -from twisted.python.failure import Failure - -from scrapy.utils.log import ( - LogCounterHandler, - SpiderLoggerAdapter, - StreamLogger, - TopLevelFormatter, - failure_to_exc_info, -) -from scrapy.utils.test import get_crawler -from tests.spiders import LogSpider - -if TYPE_CHECKING: - from collections.abc import Generator, Mapping, MutableMapping - - from scrapy.crawler import Crawler - - -class TestFailureToExcInfo: - def test_failure(self): - try: - 0 / 0 - except ZeroDivisionError: - exc_info = sys.exc_info() - failure = Failure() - - assert exc_info == failure_to_exc_info(failure) - - def test_non_failure(self): - assert failure_to_exc_info("test") is None - - -class TestTopLevelFormatter: - def test_top_level_logger(self, caplog: pytest.LogCaptureFixture) -> None: - caplog.handler.addFilter(TopLevelFormatter(["test"])) - logger = logging.getLogger("test") - logger.warning("test log msg") - assert ("test", logging.WARNING, "test log msg") in caplog.record_tuples - - def test_children_logger(self, caplog: pytest.LogCaptureFixture) -> None: - caplog.handler.addFilter(TopLevelFormatter(["test"])) - logger = logging.getLogger("test.test1") - logger.warning("test log msg") - assert ("test", logging.WARNING, "test log msg") in caplog.record_tuples - - def test_overlapping_name_logger(self, caplog: pytest.LogCaptureFixture) -> None: - caplog.handler.addFilter(TopLevelFormatter(["test"])) - logger = logging.getLogger("test2") - logger.warning("test log msg") - assert ("test2", logging.WARNING, "test log msg") in caplog.record_tuples - - def test_different_name_logger(self, caplog: pytest.LogCaptureFixture) -> None: - caplog.handler.addFilter(TopLevelFormatter(["test"])) - logger = logging.getLogger("different") - logger.warning("test log msg") - assert ("different", logging.WARNING, "test log msg") in caplog.record_tuples - - -class TestLogCounterHandler: - @pytest.fixture - def crawler(self) -> Crawler: - settings = {"LOG_LEVEL": "WARNING"} - return get_crawler(settings_dict=settings) - - @pytest.fixture - def logger(self, crawler: Crawler) -> Generator[logging.Logger]: - logger = logging.getLogger("test") - logger.setLevel(logging.DEBUG) - logger.propagate = False - handler = LogCounterHandler(crawler, level=crawler.settings.get("LOG_LEVEL")) - logger.addHandler(handler) - try: - yield logger - finally: - logger.propagate = True - logger.setLevel(logging.NOTSET) - logger.removeHandler(handler) - - def test_init(self, crawler: Crawler, logger: logging.Logger) -> None: - assert crawler.stats - assert crawler.stats.get_value("log_count/DEBUG") is None - assert crawler.stats.get_value("log_count/INFO") is None - assert crawler.stats.get_value("log_count/WARNING") is None - assert crawler.stats.get_value("log_count/ERROR") is None - assert crawler.stats.get_value("log_count/CRITICAL") is None - - def test_accepted_level(self, crawler: Crawler, logger: logging.Logger) -> None: - logger.error("test log msg") - assert crawler.stats - assert crawler.stats.get_value("log_count/ERROR") == 1 - - def test_other_crawler(self, crawler: Crawler, logger: logging.Logger) -> None: - other_crawler = get_crawler(settings_dict={"LOG_LEVEL": "WARNING"}) - logger.error("test log msg", extra={"crawler": other_crawler}) - assert crawler.stats - assert crawler.stats.get_value("log_count/ERROR") is None - - def test_other_spider(self, crawler: Crawler, logger: logging.Logger) -> None: - other_crawler = get_crawler(settings_dict={"LOG_LEVEL": "WARNING"}) - other_spider = other_crawler._create_spider(name="other") - logger.error("test log msg", extra={"spider": other_spider}) - assert crawler.stats - assert crawler.stats.get_value("log_count/ERROR") is None - - def test_filtered_out_level(self, crawler: Crawler, logger: logging.Logger) -> None: - logger.debug("test log msg") - assert crawler.stats - assert crawler.stats.get_value("log_count/DEBUG") is None - - -class TestStreamLogger: - def test_redirect(self): - logger = logging.getLogger("test") - logger.setLevel(logging.WARNING) - old_stdout = sys.stdout - sys.stdout = StreamLogger(logger, logging.ERROR) - - with LogCapture() as log: - print("test log msg") - log.check(("test", "ERROR", "test log msg")) - - sys.stdout = old_stdout - - -@pytest.mark.parametrize( - ("base_extra", "log_extra", "expected_extra"), - [ - ( - {"spider": "test"}, - {"extra": {"log_extra": "info"}}, - {"extra": {"log_extra": "info", "spider": "test"}}, - ), - ( - {"spider": "test"}, - {"extra": None}, - {"extra": {"spider": "test"}}, - ), - ( - {"spider": "test"}, - {"extra": {"spider": "test2"}}, - {"extra": {"spider": "test"}}, - ), - ], -) -def test_spider_logger_adapter_process( - base_extra: Mapping[str, Any], - log_extra: MutableMapping[str, Any], - expected_extra: dict[str, Any], -) -> None: - logger = logging.getLogger("test") - spider_logger_adapter = SpiderLoggerAdapter(logger, base_extra) - - log_message = "test_log_message" - result_message, result_kwargs = spider_logger_adapter.process( - log_message, log_extra - ) - - assert result_message == log_message - assert result_kwargs == expected_extra - - -class TestLogging: - @pytest.fixture - def log_stream(self) -> StringIO: - return StringIO() - - @pytest.fixture - def spider(self) -> LogSpider: - return LogSpider() - - @pytest.fixture(autouse=True) - def logger(self, log_stream: StringIO) -> Generator[logging.Logger]: - handler = logging.StreamHandler(log_stream) - logger = logging.getLogger("log_spider") - logger.addHandler(handler) - logger.setLevel(logging.DEBUG) - - yield logger - - logger.removeHandler(handler) - - def test_debug_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Foo message" - spider.log_debug(log_message) - log_contents = log_stream.getvalue() - - assert log_contents == f"{log_message}\n" - - def test_info_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Bar message" - spider.log_info(log_message) - log_contents = log_stream.getvalue() - - assert log_contents == f"{log_message}\n" - - def test_warning_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Baz message" - spider.log_warning(log_message) - log_contents = log_stream.getvalue() - - assert log_contents == f"{log_message}\n" - - def test_error_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Foo bar message" - spider.log_error(log_message) - log_contents = log_stream.getvalue() - - assert log_contents == f"{log_message}\n" - - def test_critical_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Foo bar baz message" - spider.log_critical(log_message) - log_contents = log_stream.getvalue() - - assert log_contents == f"{log_message}\n" - - -class TestLoggingWithExtra: - regex_pattern = re.compile(r"^]+>$") - - @pytest.fixture - def log_stream(self) -> StringIO: - return StringIO() - - @pytest.fixture - def spider(self) -> LogSpider: - return LogSpider() - - @pytest.fixture(autouse=True) - def logger(self, log_stream: StringIO) -> Generator[logging.Logger]: - handler = logging.StreamHandler(log_stream) - formatter = logging.Formatter( - '{"levelname": "%(levelname)s", "message": "%(message)s", "spider": "%(spider)s", "important_info": "%(important_info)s"}' - ) - handler.setFormatter(formatter) - logger = logging.getLogger("log_spider") - logger.addHandler(handler) - logger.setLevel(logging.DEBUG) - - yield logger - - logger.removeHandler(handler) - - def test_debug_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Foo message" - extra = {"important_info": "foo"} - spider.log_debug(log_message, extra) - log_contents_str = log_stream.getvalue() - log_contents = json.loads(log_contents_str) - - assert log_contents["levelname"] == "DEBUG" - assert log_contents["message"] == log_message - assert self.regex_pattern.match(log_contents["spider"]) - assert log_contents["important_info"] == extra["important_info"] - - def test_info_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Bar message" - extra = {"important_info": "bar"} - spider.log_info(log_message, extra) - log_contents_str = log_stream.getvalue() - log_contents = json.loads(log_contents_str) - - assert log_contents["levelname"] == "INFO" - assert log_contents["message"] == log_message - assert self.regex_pattern.match(log_contents["spider"]) - assert log_contents["important_info"] == extra["important_info"] - - def test_warning_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Baz message" - extra = {"important_info": "baz"} - spider.log_warning(log_message, extra) - log_contents_str = log_stream.getvalue() - log_contents = json.loads(log_contents_str) - - assert log_contents["levelname"] == "WARNING" - assert log_contents["message"] == log_message - assert self.regex_pattern.match(log_contents["spider"]) - assert log_contents["important_info"] == extra["important_info"] - - def test_error_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Foo bar message" - extra = {"important_info": "foo bar"} - spider.log_error(log_message, extra) - log_contents_str = log_stream.getvalue() - log_contents = json.loads(log_contents_str) - - assert log_contents["levelname"] == "ERROR" - assert log_contents["message"] == log_message - assert self.regex_pattern.match(log_contents["spider"]) - assert log_contents["important_info"] == extra["important_info"] - - def test_critical_logging(self, log_stream: StringIO, spider: LogSpider) -> None: - log_message = "Foo bar baz message" - extra = {"important_info": "foo bar baz"} - spider.log_critical(log_message, extra) - log_contents_str = log_stream.getvalue() - log_contents = json.loads(log_contents_str) - - assert log_contents["levelname"] == "CRITICAL" - assert log_contents["message"] == log_message - assert self.regex_pattern.match(log_contents["spider"]) - assert log_contents["important_info"] == extra["important_info"] - - def test_overwrite_spider_extra( - self, log_stream: StringIO, spider: LogSpider - ) -> None: - log_message = "Foo message" - extra = {"important_info": "foo", "spider": "shouldn't change"} - spider.log_error(log_message, extra) - log_contents_str = log_stream.getvalue() - log_contents = json.loads(log_contents_str) - - assert log_contents["levelname"] == "ERROR" - assert log_contents["message"] == log_message - assert self.regex_pattern.match(log_contents["spider"]) - assert log_contents["important_info"] == extra["important_info"] +from __future__ import annotations + +import json +import logging +import re +import sys +from io import StringIO +from typing import TYPE_CHECKING, Any + +import pytest +from testfixtures import LogCapture +from twisted.python.failure import Failure + +from scrapy.utils.log import ( + LogCounterHandler, + SpiderLoggerAdapter, + StreamLogger, + TopLevelFormatter, + failure_to_exc_info, +) +from scrapy.utils.test import get_crawler +from tests.spiders import LogSpider + +if TYPE_CHECKING: + from collections.abc import Generator, Mapping, MutableMapping + + from scrapy.crawler import Crawler + + +class TestFailureToExcInfo: + def test_failure(self): + try: + 0 / 0 + except ZeroDivisionError: + exc_info = sys.exc_info() + failure = Failure() + + assert exc_info == failure_to_exc_info(failure) + + def test_non_failure(self): + assert failure_to_exc_info("test") is None + + +class TestTopLevelFormatter: + def test_top_level_logger(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.handler.addFilter(TopLevelFormatter(["test"])) + logger = logging.getLogger("test") + logger.warning("test log msg") + assert ("test", logging.WARNING, "test log msg") in caplog.record_tuples + + def test_children_logger(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.handler.addFilter(TopLevelFormatter(["test"])) + logger = logging.getLogger("test.test1") + logger.warning("test log msg") + assert ("test", logging.WARNING, "test log msg") in caplog.record_tuples + + def test_overlapping_name_logger(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.handler.addFilter(TopLevelFormatter(["test"])) + logger = logging.getLogger("test2") + logger.warning("test log msg") + assert ("test2", logging.WARNING, "test log msg") in caplog.record_tuples + + def test_different_name_logger(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.handler.addFilter(TopLevelFormatter(["test"])) + logger = logging.getLogger("different") + logger.warning("test log msg") + assert ("different", logging.WARNING, "test log msg") in caplog.record_tuples + + +class TestLogCounterHandler: + @pytest.fixture + def crawler(self) -> Crawler: + settings = {"LOG_LEVEL": "WARNING"} + return get_crawler(settings_dict=settings) + + @pytest.fixture + def logger(self, crawler: Crawler) -> Generator[logging.Logger]: + logger = logging.getLogger("test") + logger.setLevel(logging.DEBUG) + logger.propagate = False + handler = LogCounterHandler(crawler, level=crawler.settings.get("LOG_LEVEL")) + logger.addHandler(handler) + try: + yield logger + finally: + logger.propagate = True + logger.setLevel(logging.NOTSET) + logger.removeHandler(handler) + + def test_init(self, crawler: Crawler, logger: logging.Logger) -> None: + assert crawler.stats + assert crawler.stats.get_value("log_count/DEBUG") is None + assert crawler.stats.get_value("log_count/INFO") is None + assert crawler.stats.get_value("log_count/WARNING") is None + assert crawler.stats.get_value("log_count/ERROR") is None + assert crawler.stats.get_value("log_count/CRITICAL") is None + + def test_accepted_level(self, crawler: Crawler, logger: logging.Logger) -> None: + logger.error("test log msg") + assert crawler.stats + assert crawler.stats.get_value("log_count/ERROR") == 1 + + def test_other_crawler(self, crawler: Crawler, logger: logging.Logger) -> None: + other_crawler = get_crawler(settings_dict={"LOG_LEVEL": "WARNING"}) + logger.error("test log msg", extra={"crawler": other_crawler}) + assert crawler.stats + assert crawler.stats.get_value("log_count/ERROR") is None + + def test_other_spider(self, crawler: Crawler, logger: logging.Logger) -> None: + other_crawler = get_crawler(settings_dict={"LOG_LEVEL": "WARNING"}) + other_spider = other_crawler._create_spider(name="other") + logger.error("test log msg", extra={"spider": other_spider}) + assert crawler.stats + assert crawler.stats.get_value("log_count/ERROR") is None + + def test_filtered_out_level(self, crawler: Crawler, logger: logging.Logger) -> None: + logger.debug("test log msg") + assert crawler.stats + assert crawler.stats.get_value("log_count/DEBUG") is None + + +class TestStreamLogger: + def test_redirect(self): + logger = logging.getLogger("test") + logger.setLevel(logging.WARNING) + old_stdout = sys.stdout + sys.stdout = StreamLogger(logger, logging.ERROR) + + with LogCapture() as log: + print("test log msg") + log.check(("test", "ERROR", "test log msg")) + + sys.stdout = old_stdout + + +@pytest.mark.parametrize( + ("base_extra", "log_extra", "expected_extra"), + [ + ( + {"spider": "test"}, + {"extra": {"log_extra": "info"}}, + {"extra": {"log_extra": "info", "spider": "test"}}, + ), + ( + {"spider": "test"}, + {"extra": None}, + {"extra": {"spider": "test"}}, + ), + ( + {"spider": "test"}, + {"extra": {"spider": "test2"}}, + {"extra": {"spider": "test"}}, + ), + ], +) +def test_spider_logger_adapter_process( + base_extra: Mapping[str, Any], + log_extra: MutableMapping[str, Any], + expected_extra: dict[str, Any], +) -> None: + logger = logging.getLogger("test") + spider_logger_adapter = SpiderLoggerAdapter(logger, base_extra) + + log_message = "test_log_message" + result_message, result_kwargs = spider_logger_adapter.process( + log_message, log_extra + ) + + assert result_message == log_message + assert result_kwargs == expected_extra + + +class TestLogging: + @pytest.fixture + def log_stream(self) -> StringIO: + return StringIO() + + @pytest.fixture + def spider(self) -> LogSpider: + return LogSpider() + + @pytest.fixture(autouse=True) + def logger(self, log_stream: StringIO) -> Generator[logging.Logger]: + handler = logging.StreamHandler(log_stream) + logger = logging.getLogger("log_spider") + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + + yield logger + + logger.removeHandler(handler) + + def test_debug_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Foo message" + spider.log_debug(log_message) + log_contents = log_stream.getvalue() + + assert log_contents == f"{log_message}\n" + + def test_info_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Bar message" + spider.log_info(log_message) + log_contents = log_stream.getvalue() + + assert log_contents == f"{log_message}\n" + + def test_warning_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Baz message" + spider.log_warning(log_message) + log_contents = log_stream.getvalue() + + assert log_contents == f"{log_message}\n" + + def test_error_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Foo bar message" + spider.log_error(log_message) + log_contents = log_stream.getvalue() + + assert log_contents == f"{log_message}\n" + + def test_critical_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Foo bar baz message" + spider.log_critical(log_message) + log_contents = log_stream.getvalue() + + assert log_contents == f"{log_message}\n" + + +class TestLoggingWithExtra: + regex_pattern = re.compile(r"^]+>$") + + @pytest.fixture + def log_stream(self) -> StringIO: + return StringIO() + + @pytest.fixture + def spider(self) -> LogSpider: + return LogSpider() + + @pytest.fixture(autouse=True) + def logger(self, log_stream: StringIO) -> Generator[logging.Logger]: + handler = logging.StreamHandler(log_stream) + formatter = logging.Formatter( + '{"levelname": "%(levelname)s", "message": "%(message)s", "spider": "%(spider)s", "important_info": "%(important_info)s"}' + ) + handler.setFormatter(formatter) + logger = logging.getLogger("log_spider") + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + + yield logger + + logger.removeHandler(handler) + + def test_debug_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Foo message" + extra = {"important_info": "foo"} + spider.log_debug(log_message, extra) + log_contents_str = log_stream.getvalue() + log_contents = json.loads(log_contents_str) + + assert log_contents["levelname"] == "DEBUG" + assert log_contents["message"] == log_message + assert self.regex_pattern.match(log_contents["spider"]) + assert log_contents["important_info"] == extra["important_info"] + + def test_info_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Bar message" + extra = {"important_info": "bar"} + spider.log_info(log_message, extra) + log_contents_str = log_stream.getvalue() + log_contents = json.loads(log_contents_str) + + assert log_contents["levelname"] == "INFO" + assert log_contents["message"] == log_message + assert self.regex_pattern.match(log_contents["spider"]) + assert log_contents["important_info"] == extra["important_info"] + + def test_warning_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Baz message" + extra = {"important_info": "baz"} + spider.log_warning(log_message, extra) + log_contents_str = log_stream.getvalue() + log_contents = json.loads(log_contents_str) + + assert log_contents["levelname"] == "WARNING" + assert log_contents["message"] == log_message + assert self.regex_pattern.match(log_contents["spider"]) + assert log_contents["important_info"] == extra["important_info"] + + def test_error_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Foo bar message" + extra = {"important_info": "foo bar"} + spider.log_error(log_message, extra) + log_contents_str = log_stream.getvalue() + log_contents = json.loads(log_contents_str) + + assert log_contents["levelname"] == "ERROR" + assert log_contents["message"] == log_message + assert self.regex_pattern.match(log_contents["spider"]) + assert log_contents["important_info"] == extra["important_info"] + + def test_critical_logging(self, log_stream: StringIO, spider: LogSpider) -> None: + log_message = "Foo bar baz message" + extra = {"important_info": "foo bar baz"} + spider.log_critical(log_message, extra) + log_contents_str = log_stream.getvalue() + log_contents = json.loads(log_contents_str) + + assert log_contents["levelname"] == "CRITICAL" + assert log_contents["message"] == log_message + assert self.regex_pattern.match(log_contents["spider"]) + assert log_contents["important_info"] == extra["important_info"] + + def test_overwrite_spider_extra( + self, log_stream: StringIO, spider: LogSpider + ) -> None: + log_message = "Foo message" + extra = {"important_info": "foo", "spider": "shouldn't change"} + spider.log_error(log_message, extra) + log_contents_str = log_stream.getvalue() + log_contents = json.loads(log_contents_str) + + assert log_contents["levelname"] == "ERROR" + assert log_contents["message"] == log_message + assert self.regex_pattern.match(log_contents["spider"]) + assert log_contents["important_info"] == extra["important_info"]