398 lines
15 KiB
Python
398 lines
15 KiB
Python
import logging
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Optional, Dict, Any, Union
|
||
from datetime import datetime
|
||
import json
|
||
import traceback
|
||
import inspect
|
||
|
||
# Цвета для консольного вывода (только если поддерживается)
|
||
from colorama import init, Fore, Back, Style
|
||
init(autoreset=True)
|
||
|
||
class TestLogger:
|
||
"""Кастомный логгер для тестов с поддержкой Allure и консольного вывода"""
|
||
|
||
# Уровни логирования с цветами
|
||
LEVEL_COLORS = {
|
||
'DEBUG': Fore.CYAN,
|
||
'INFO': Fore.GREEN,
|
||
'WARNING': Fore.YELLOW,
|
||
'ERROR': Fore.RED,
|
||
'CRITICAL': Fore.RED + Back.WHITE,
|
||
}
|
||
|
||
# Уровни логирования для Allure
|
||
ALLURE_LEVELS = {
|
||
'DEBUG': 'debug',
|
||
'INFO': 'info',
|
||
'WARNING': 'warning',
|
||
'ERROR': 'error',
|
||
'CRITICAL': 'critical',
|
||
}
|
||
|
||
_instances = {} # Кэш инстансов логгеров
|
||
|
||
def __init__(self, name: str, level: str = 'INFO',
|
||
log_to_file: bool = True,
|
||
log_to_console: bool = True,
|
||
log_to_allure: bool = True):
|
||
"""
|
||
Инициализация логгера
|
||
|
||
Args:
|
||
name: Имя логгера (обычно __name__)
|
||
level: Уровень логирования (DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
||
log_to_file: Логировать в файл
|
||
log_to_console: Логировать в консоль
|
||
log_to_allure: Логировать в Allure отчет
|
||
"""
|
||
self.name = name
|
||
self.log_to_file = log_to_file
|
||
self.log_to_console = log_to_console
|
||
self.log_to_allure = log_to_allure
|
||
|
||
# Создаем стандартный логгер
|
||
self.logger = logging.getLogger(name)
|
||
self.logger.setLevel(getattr(logging, level.upper()))
|
||
|
||
# Удаляем существующие обработчики
|
||
self.logger.handlers.clear()
|
||
|
||
# Форматтеры
|
||
self._setup_formatters()
|
||
|
||
# Обработчики
|
||
self._setup_handlers()
|
||
|
||
# Для хранения контекста теста
|
||
self.test_context = {}
|
||
|
||
def _setup_formatters(self):
|
||
"""Настройка форматтеров"""
|
||
# Подробный формат для файла
|
||
self.file_formatter = logging.Formatter(
|
||
fmt='%(asctime)s.%(msecs)03d | %(levelname)-8s | %(name)s | %(filename)s:%(lineno)d | %(message)s',
|
||
datefmt='%Y-%m-%d %H:%M:%S'
|
||
)
|
||
|
||
# Простой формат для консоли
|
||
self.console_formatter = logging.Formatter(
|
||
fmt='%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s',
|
||
datefmt='%H:%M:%S'
|
||
)
|
||
|
||
# JSON формат для машинной обработки
|
||
self.json_formatter = JSONFormatter()
|
||
|
||
def _setup_handlers(self):
|
||
"""Настройка обработчиков"""
|
||
# Консольный обработчик
|
||
if self.log_to_console:
|
||
console_handler = logging.StreamHandler(sys.stdout)
|
||
console_handler.setLevel(self.logger.level)
|
||
console_handler.setFormatter(self.console_formatter)
|
||
|
||
console_handler.setFormatter(ColorFormatter())
|
||
|
||
self.logger.addHandler(console_handler)
|
||
|
||
# Файловый обработчик
|
||
if self.log_to_file:
|
||
self._setup_file_handler()
|
||
|
||
def _setup_file_handler(self):
|
||
"""Настройка файлового обработчика"""
|
||
from config.settings import settings
|
||
|
||
log_file = Path(settings.LOG_FILE)
|
||
log_file.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
file_handler = logging.FileHandler(log_file, encoding='utf-8')
|
||
file_handler.setLevel(self.logger.level)
|
||
file_handler.setFormatter(self.file_formatter)
|
||
self.logger.addHandler(file_handler)
|
||
|
||
# JSON лог для машинной обработки
|
||
json_log_file = log_file.parent / f"{log_file.stem}_json{log_file.suffix}"
|
||
json_handler = logging.FileHandler(json_log_file, encoding='utf-8')
|
||
json_handler.setLevel(self.logger.level)
|
||
json_handler.setFormatter(self.json_formatter)
|
||
self.logger.addHandler(json_handler)
|
||
|
||
def set_test_context(self, **kwargs):
|
||
"""Установка контекста теста (имя теста, тестовые данные и т.д.)"""
|
||
self.test_context.update(kwargs)
|
||
|
||
def clear_test_context(self):
|
||
"""Очистка контекста теста"""
|
||
self.test_context.clear()
|
||
|
||
def debug(self, message: str, **kwargs):
|
||
"""Логирование на уровне DEBUG"""
|
||
self._log('DEBUG', message, **kwargs)
|
||
|
||
def info(self, message: str, **kwargs):
|
||
"""Логирование на уровне INFO"""
|
||
self._log('INFO', message, **kwargs)
|
||
|
||
def warning(self, message: str, **kwargs):
|
||
"""Логирование на уровне WARNING"""
|
||
self._log('WARNING', message, **kwargs)
|
||
|
||
def error(self, message: str, exception: Optional[Exception] = None, **kwargs):
|
||
"""Логирование на уровне ERROR"""
|
||
if exception:
|
||
message = f"{message} | Exception: {exception} | Traceback: {traceback.format_exc()}"
|
||
self._log('ERROR', message, **kwargs)
|
||
|
||
def critical(self, message: str, **kwargs):
|
||
"""Логирование на уровне CRITICAL"""
|
||
self._log('CRITICAL', message, **kwargs)
|
||
|
||
def _log(self, level: str, message: str, **kwargs):
|
||
"""Внутренний метод логирования"""
|
||
# Добавляем контекст к сообщению
|
||
if self.test_context:
|
||
context_str = ' | '.join(f"{k}: {v}" for k, v in self.test_context.items())
|
||
message = f"{message} | Context: {context_str}"
|
||
|
||
# Добавляем дополнительные поля
|
||
if kwargs:
|
||
extra_fields = ' | '.join(f"{k}: {v}" for k, v in kwargs.items())
|
||
message = f"{message} | {extra_fields}"
|
||
|
||
# Получаем информацию о вызывающем коде
|
||
curr_frame = inspect.currentframe()
|
||
lineno = None
|
||
func_name = None
|
||
filename = None
|
||
|
||
if curr_frame and curr_frame.f_back:
|
||
frame = curr_frame.f_back.f_back
|
||
if frame:
|
||
filename = frame.f_code.co_filename
|
||
lineno = frame.f_lineno
|
||
func_name = frame.f_code.co_name
|
||
|
||
extra = {
|
||
'filename': Path(filename).name if filename else '-',
|
||
'lineno': lineno,
|
||
'func_name': func_name,
|
||
'test_context': self.test_context.copy(),
|
||
'timestamp': datetime.now().isoformat(),
|
||
}
|
||
|
||
# Логируем через стандартный логгер
|
||
log_method = getattr(self.logger, level.lower())
|
||
log_method(message, extra=extra)
|
||
|
||
# Логируем в Allure
|
||
if self.log_to_allure:
|
||
self._log_to_allure(level, message)
|
||
|
||
def _log_to_allure(self, level: str, message: str):
|
||
"""Логирование в Allure отчет"""
|
||
try:
|
||
import allure
|
||
|
||
allure_level = self.ALLURE_LEVELS.get(level, 'info')
|
||
getattr(allure.step, allure_level)(message)
|
||
except ImportError:
|
||
pass # Allure не установлен
|
||
except Exception as e:
|
||
self.logger.warning(f"Failed to log to Allure: {e}")
|
||
|
||
def log_api_request(self, method: str, url: str,
|
||
headers: Optional[dict[str, Any]] = None,
|
||
body: Any = None,
|
||
response: Any = None):
|
||
"""Логирование API запроса"""
|
||
log_data = {
|
||
'type': 'API_REQUEST',
|
||
'method': method,
|
||
'url': url,
|
||
'timestamp': datetime.now().isoformat(),
|
||
}
|
||
|
||
if headers:
|
||
log_data['headers'] = {k: v for k, v in headers.items() if k.lower() != 'authorization'}
|
||
|
||
if body:
|
||
log_data['body'] = str(body)[:500] + ('...' if len(str(body)) > 500 else '')
|
||
|
||
if response:
|
||
log_data['response_status'] = getattr(response, 'status_code', None)
|
||
log_data['response_body'] = str(getattr(response, 'text', ''))[:500] + ('...' if len(str(getattr(response, 'text', ''))) > 500 else '')
|
||
|
||
self.debug(f"API Request: {method} {url}", **log_data)
|
||
|
||
def log_ui_action(self, action: str, element: Optional[str] = None,
|
||
value: Optional[str] = None, success: bool = True):
|
||
"""Логирование UI действий"""
|
||
log_data = {
|
||
'type': 'UI_ACTION',
|
||
'action': action,
|
||
'element': element,
|
||
'value': value,
|
||
'success': success,
|
||
'timestamp': datetime.now().isoformat(),
|
||
}
|
||
|
||
level = 'INFO' if success else 'ERROR'
|
||
message = f"UI Action: {action}"
|
||
if element:
|
||
message += f" on {element}"
|
||
if value:
|
||
message += f" with value: {value}"
|
||
|
||
self._log(level, message, **log_data)
|
||
|
||
def log_test_start(self, test_name: str, test_params: Optional[dict] = None):
|
||
"""Логирование начала теста"""
|
||
self.set_test_context(test_name=test_name)
|
||
|
||
log_data = {
|
||
'type': 'TEST_START',
|
||
'test_name': test_name,
|
||
'timestamp': datetime.now().isoformat(),
|
||
}
|
||
|
||
if test_params:
|
||
log_data['parameters'] = test_params
|
||
|
||
self.info(f"🚀 Starting test: {test_name}", **log_data)
|
||
|
||
def log_test_end(self, test_name: str, status: str,
|
||
duration: float = None, error: str = None):
|
||
"""Логирование окончания теста"""
|
||
log_data = {
|
||
'type': 'TEST_END',
|
||
'test_name': test_name,
|
||
'status': status,
|
||
'timestamp': datetime.now().isoformat(),
|
||
}
|
||
|
||
if duration is not None:
|
||
log_data['duration_seconds'] = round(duration, 2)
|
||
|
||
if error:
|
||
log_data['error'] = error
|
||
|
||
emoji = '✅' if status == 'PASSED' else '❌' if status == 'FAILED' else '⚠️'
|
||
self.info(f"{emoji} Test {status}: {test_name} "
|
||
f"(Duration: {duration:.2f}s)" if duration else "", **log_data)
|
||
|
||
self.clear_test_context()
|
||
|
||
def log_screenshot(self, screenshot_path: str, description: str = "Screenshot"):
|
||
"""Логирование скриншота"""
|
||
log_data = {
|
||
'type': 'SCREENSHOT',
|
||
'path': screenshot_path,
|
||
'description': description,
|
||
'timestamp': datetime.now().isoformat(),
|
||
}
|
||
|
||
self.info(f"📸 Screenshot saved: {description} -> {screenshot_path}", **log_data)
|
||
|
||
# Прикрепляем к Allure
|
||
if self.log_to_allure:
|
||
try:
|
||
import allure
|
||
if Path(screenshot_path).exists():
|
||
with open(screenshot_path, 'rb') as f:
|
||
allure.attach(
|
||
f.read(),
|
||
name=description,
|
||
attachment_type=allure.attachment_type.PNG
|
||
)
|
||
except Exception as e:
|
||
self.warning(f"Failed to attach screenshot to Allure: {e}")
|
||
|
||
@classmethod
|
||
def get_logger(cls, name: Optional[str] = None, **kwargs) -> 'TestLogger':
|
||
"""Фабричный метод для получения логгера (singleton pattern)"""
|
||
if name is None:
|
||
# Получаем имя вызывающего модуля
|
||
curr_frame = inspect.currentframe()
|
||
if curr_frame:
|
||
frame = curr_frame.f_back
|
||
module = inspect.getmodule(frame)
|
||
name = module.__name__ if module else '__main__'
|
||
|
||
if name not in cls._instances and name:
|
||
cls._instances[name] = cls(name, **kwargs)
|
||
|
||
return cls._instances[name]
|
||
|
||
|
||
class ColorFormatter(logging.Formatter):
|
||
"""Форматтер с цветами для консоли"""
|
||
|
||
FORMATS = {
|
||
logging.DEBUG: Fore.CYAN + '%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s' + Style.RESET_ALL,
|
||
logging.INFO: Fore.GREEN + '%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s' + Style.RESET_ALL,
|
||
logging.WARNING: Fore.YELLOW + '%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s' + Style.RESET_ALL,
|
||
logging.ERROR: Fore.RED + '%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s' + Style.RESET_ALL,
|
||
logging.CRITICAL: Fore.RED + Back.WHITE + '%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s' + Style.RESET_ALL,
|
||
}
|
||
|
||
def format(self, record):
|
||
log_fmt = self.FORMATS.get(record.levelno, self.FORMATS[logging.INFO])
|
||
formatter = logging.Formatter(log_fmt, datefmt='%H:%M:%S')
|
||
return formatter.format(record)
|
||
|
||
|
||
class JSONFormatter(logging.Formatter):
|
||
"""Форматтер для JSON логов (удобно для машинной обработки)"""
|
||
|
||
def format(self, record):
|
||
log_record = {
|
||
'timestamp': datetime.now().isoformat(),
|
||
'level': record.levelname,
|
||
'logger': record.name,
|
||
'message': record.getMessage(),
|
||
'module': record.module,
|
||
'function': record.funcName,
|
||
'line': record.lineno,
|
||
'thread': record.threadName,
|
||
'process': record.processName,
|
||
}
|
||
|
||
if not record:
|
||
return json.dumps(log_record, ensure_ascii=False)
|
||
|
||
# Добавляем дополнительные поля из extra
|
||
if hasattr(record, 'test_context'):
|
||
log_record['test_context'] = record.test_context
|
||
|
||
if hasattr(record, 'filename'):
|
||
log_record['filename'] = record.filename
|
||
|
||
# Добавляем exception info если есть
|
||
if record.exc_info:
|
||
log_record['exception'] = {
|
||
'type': record.exc_info[0].__name__ if record.exc_info[0] else 'Unknown exception type',
|
||
'message': str(record.exc_info[1]),
|
||
'traceback': self.formatException(record.exc_info),
|
||
}
|
||
|
||
return json.dumps(log_record, ensure_ascii=False)
|
||
|
||
|
||
# Глобальный логгер по умолчанию
|
||
def get_logger(name: Optional[str] = None, **kwargs) -> TestLogger:
|
||
"""
|
||
Утилита для получения логгера
|
||
|
||
Args:
|
||
name: Имя логгера (обычно __name__)
|
||
**kwargs: Дополнительные параметры для TestLogger
|
||
|
||
Returns:
|
||
Экземпляр TestLogger
|
||
"""
|
||
return TestLogger.get_logger(name, **kwargs)
|