57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
|
import logging
|
||
|
|
import threading
|
||
|
|
from contextlib import AbstractContextManager, ContextDecorator
|
||
|
|
from functools import wraps
|
||
|
|
from types import TracebackType
|
||
|
|
from typing import Any, Callable
|
||
|
|
|
||
|
|
from frostfs_testlib.reporter.interfaces import ReporterHandler
|
||
|
|
|
||
|
|
|
||
|
|
class StepsLogger(ReporterHandler):
|
||
|
|
"""Handler that prints steps to log."""
|
||
|
|
|
||
|
|
def step(self, name: str) -> AbstractContextManager | ContextDecorator:
|
||
|
|
return StepLoggerContext(name)
|
||
|
|
|
||
|
|
def step_decorator(self, name: str) -> Callable:
|
||
|
|
return StepLoggerContext(name)
|
||
|
|
|
||
|
|
def attach(self, body: Any, file_name: str) -> None:
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class StepLoggerContext(AbstractContextManager):
|
||
|
|
INDENT = {}
|
||
|
|
|
||
|
|
def __init__(self, title: str):
|
||
|
|
self.title = title
|
||
|
|
self.logger = logging.getLogger("NeoLogger")
|
||
|
|
self.thread = threading.get_ident()
|
||
|
|
if self.thread not in StepLoggerContext.INDENT:
|
||
|
|
StepLoggerContext.INDENT[self.thread] = 1
|
||
|
|
|
||
|
|
def __enter__(self) -> Any:
|
||
|
|
indent = ">" * StepLoggerContext.INDENT[self.thread]
|
||
|
|
self.logger.info(f"[{self.thread}] {indent} {self.title}")
|
||
|
|
StepLoggerContext.INDENT[self.thread] += 1
|
||
|
|
|
||
|
|
def __exit__(
|
||
|
|
self,
|
||
|
|
__exc_type: type[BaseException] | None,
|
||
|
|
__exc_value: BaseException | None,
|
||
|
|
__traceback: TracebackType | None,
|
||
|
|
) -> bool | None:
|
||
|
|
|
||
|
|
StepLoggerContext.INDENT[self.thread] -= 1
|
||
|
|
indent = "<" * StepLoggerContext.INDENT[self.thread]
|
||
|
|
self.logger.info(f"[{self.thread}] {indent} {self.title}")
|
||
|
|
|
||
|
|
def __call__(self, func):
|
||
|
|
@wraps(func)
|
||
|
|
def impl(*a, **kw):
|
||
|
|
with self:
|
||
|
|
return func(*a, **kw)
|
||
|
|
|
||
|
|
return impl
|