mirror of https://github.com/scrapy/scrapy.git
Normalize line endings in log counter fix
Signed-off-by: Mridankan Mandal <xerontitan90@gmail.com>
This commit is contained in:
Mridankan Mandal
parent
88a7fc40e7
commit
b3cd077b59
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"^<LogSpider\s'log_spider'\sat\s[^>]+>$")
|
||||
|
||||
@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"^<LogSpider\s'log_spider'\sat\s[^>]+>$")
|
||||
|
||||
@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"]
|
||||
|
|
|
|||
Loading…
Reference in New Issue