https://pastein.ru/t/JFT
Загрузка данных
constants
STATUS_DESELECTED = "deselected"
STATUS_NO_DATA = "no_data"
ALLURE_RESULTS_ARCHIVE_SUFFIX = ".tar.gz"
# Gzip: сигнатура в начале файла (magic bytes) - первые байты, по которым
# распознаётся формат. Для gzip это 0x1F 0x8B.
GZIP_MAGIC_BYTE_1 = 0x1F
GZIP_MAGIC_BYTE_2 = 0x8B
GZIP_HEADER_SIZE = 2
GZIP_MAGIC = bytes([GZIP_MAGIC_BYTE_1, GZIP_MAGIC_BYTE_2])
helpers\archive_utils.py
"""
Валидация и безопасная распаковка tar.gz архивов allure-results.
"""
from __future__ import annotations
import io
import os
import tarfile
from dataclasses import dataclass
from typing import Optional, Sequence
import constants as const
from helpers.allure_utils import extract_stand_from_environment_file
class UploadValidationError(Exception):
"""Ошибка валидации загружаемых файлов до создания записей в БД."""
def __init__(self, message: str, code: str = "upload_validation_failed") -> None:
self.message = message
self.code = code
super().__init__(message)
@dataclass
class TarArchiveInfo:
"""Результат успешной валидации tar.gz архива."""
payload_file_count: int
detected_stand: Optional[str]
def is_gzip_payload(data: bytes) -> bool:
"""
Проверяет, что файл начинается с gzip-сигнатуры (magic bytes).
"""
if len(data) < const.GZIP_HEADER_SIZE:
return False
header = data[:const.GZIP_HEADER_SIZE]
return header == const.GZIP_MAGIC
def _unsafe_path_error_message(member_name: str) -> str:
return (
f"Архив не принят: внутри есть файл '{member_name}' с некорректным путём. "
"Соберите allure-results.tar.gz так, чтобы файлы лежали в одной папке "
"без выхода 'вверх' по директориям."
)
def _unsafe_link_error_message(member_name: str) -> str:
return (
f"Архив не принят: внутри есть ссылка '{member_name}'. "
"В архиве должны быть только обычные файлы allure-results, без ссылок."
)
def _unsafe_special_file_error_message(member_name: str) -> str:
return (
f"Архив не принят: внутри есть специальный файл '{member_name}'. "
"В архиве должны быть только обычные файлы allure-results."
)
def is_allure_payload_filename(name: str) -> bool:
"""Проверяет, что имя файла соответствует allure result/container."""
basename = os.path.basename(name.replace("\\", "/"))
return (
basename.endswith(const.RESULT_NAMING)
or basename.endswith(const.CONTAINER_NAMING)
)
def _normalize_member_path(path: str) -> str:
return path.replace("\\", "/").lstrip("/")
def _is_safe_member_path(destination_dir: str, member_name: str) -> bool:
normalized = _normalize_member_path(member_name)
if normalized.startswith("/") or normalized.startswith("../"):
return False
abs_destination = os.path.abspath(destination_dir)
member_target = os.path.abspath(os.path.join(destination_dir, normalized))
return member_target == abs_destination or member_target.startswith(
abs_destination + os.sep
)
def _reject_unsafe_tar_members(tar: tarfile.TarFile, destination_dir: str) -> None:
for member in tar.getmembers():
if not _is_safe_member_path(destination_dir, member.name):
raise UploadValidationError(
_unsafe_path_error_message(member.name),
code="unsafe_archive_path",
)
if member.issym() or member.islnk():
raise UploadValidationError(
_unsafe_link_error_message(member.name),
code="unsafe_archive_member",
)
if member.ischr() or member.isblk() or member.isfifo():
raise UploadValidationError(
_unsafe_special_file_error_message(member.name),
code="unsafe_archive_member",
)
def _collect_payload_paths(tar: tarfile.TarFile) -> list[str]:
payload_paths: list[str] = []
for member in tar.getmembers():
if not member.isfile():
continue
normalized = _normalize_member_path(member.name)
if is_allure_payload_filename(normalized):
payload_paths.append(normalized)
return payload_paths
def _validate_payload_directory_layout(payload_paths: Sequence[str]) -> None:
"""
Проверяет, что файлы allure-results можно однозначно сопоставить с одной директорией.
Логика соответствует _resolve_allure_results_dir в testrun_helpers.
"""
if not payload_paths:
raise UploadValidationError(
"Архив не содержит файлов allure-results "
"(*-result.json или *-container.json).",
code="missing_allure_results",
)
root_payload_files = [
path for path in payload_paths if "/" not in path and "\\" not in path
]
if root_payload_files:
nested_paths = [
path for path in payload_paths if "/" in path or "\\" in path
]
if nested_paths:
raise UploadValidationError(
"Архив содержит файлы allure-results в корне и во вложенных "
"директориях; невозможно однозначно определить источник отчёта.",
code="ambiguous_allure_results",
)
return
payload_directories: set[str] = set()
for path in payload_paths:
parent = os.path.dirname(path.replace("\\", "/"))
if parent:
payload_directories.add(parent)
if len(payload_directories) != 1:
raise UploadValidationError(
"Архив содержит файлы allure-results в нескольких директориях; "
"невозможно однозначно определить источник отчёта.",
code="ambiguous_allure_results",
)
def _extract_stand_from_tar(tar: tarfile.TarFile) -> Optional[str]:
for member in tar.getmembers():
if not member.isfile():
continue
if os.path.basename(member.name.replace("\\", "/")) != "environment.properties":
continue
fileobj = tar.extractfile(member)
if not fileobj:
continue
raw = fileobj.read()
stand = extract_stand_from_environment_file(raw)
if stand:
return stand.strip()
return None
def open_validated_tar_gz(archive_bytes: bytes) -> tarfile.TarFile:
"""
Проверяет, что байты - gzip+tar, и открывает tar для чтения.
Вызывающий код должен закрыть TarFile.
"""
if not archive_bytes:
raise UploadValidationError("Архив пустой.", code="empty_archive")
if not is_gzip_payload(archive_bytes):
raise UploadValidationError(
"Файл не является корректным gzip-архивом (ожидается tar.gz).",
code="invalid_gzip",
)
try:
tar = tarfile.open(fileobj=io.BytesIO(archive_bytes), mode="r:gz")
except tarfile.TarError as error:
raise UploadValidationError(
f"Архив не является корректным tar.gz: {error}",
code="invalid_tar_gz",
) from error
try:
members = tar.getmembers()
if not members:
tar.close()
raise UploadValidationError(
"Архив tar.gz не содержит файлов.",
code="empty_tar_archive",
)
except tarfile.TarError as error:
tar.close()
raise UploadValidationError(
f"Не удалось прочитать содержимое tar.gz: {error}",
code="invalid_tar_gz",
) from error
return tar
def validate_tar_gz_archive(archive_bytes: bytes) -> TarArchiveInfo:
"""Полная валидация архива allure-results до записи в БД/MinIO."""
tar = open_validated_tar_gz(archive_bytes)
try:
validation_root = os.path.abspath(os.path.join(os.getcwd(), ".archive_validation"))
_reject_unsafe_tar_members(tar, validation_root)
payload_paths = _collect_payload_paths(tar)
_validate_payload_directory_layout(payload_paths)
detected_stand = _extract_stand_from_tar(tar)
return TarArchiveInfo(
payload_file_count=len(payload_paths),
detected_stand=detected_stand,
)
finally:
tar.close()
def safe_extract_tar_gz_bytes(archive_bytes: bytes, destination_dir: str) -> None:
"""
Безопасно распаковывает tar.gz в destination_dir.
Предполагается, что архив уже прошёл validate_tar_gz_archive.
"""
tar = open_validated_tar_gz(archive_bytes)
try:
os.makedirs(destination_dir, exist_ok=True)
_reject_unsafe_tar_members(tar, destination_dir)
extract_filter = getattr(tarfile, "data_filter", None)
if extract_filter is not None:
tar.extractall(destination_dir, filter="data")
return
for member in tar.getmembers():
if not member.isfile():
continue
if not _is_safe_member_path(destination_dir, member.name):
raise UploadValidationError(
_unsafe_path_error_message(member.name),
code="unsafe_archive_path",
)
tar.extract(member, path=destination_dir)
finally:
tar.close()
helpers\testrun_helpers.py
import datetime
import io
import json
import os
import shutil
import subprocess
import tempfile
from typing import Any, Dict, List, Literal, Optional, Sequence, Tuple
from zoneinfo import ZoneInfo
import flask
from flask import abort
from minio import S3Error
from sqlalchemy import inspect
from sqlalchemy.exc import DatabaseError
from werkzeug.datastructures import FileStorage
import constants as const
from app import db
from app.clients import MinioClient
from app.models import TestResult
from helpers.allure_utils import extract_stand_from_environment_file
from helpers.archive_utils import (
UploadValidationError,
is_allure_payload_filename,
open_validated_tar_gz,
safe_extract_tar_gz_bytes,
validate_tar_gz_archive,
)
from logger import init_logger
RunStatusSignal = Literal["none", "fail", "broken"]
minio_client = MinioClient()
logger = init_logger()
def validate_upload_files(files: Sequence[FileStorage]) -> None:
"""
Валидирует загружаемые файлы до создания записей в БД и загрузки в MinIO.
"""
archive_filenames: list[str] = []
loose_payload_count = 0
for file in files:
filename = file.filename or ""
if not allowed_file(filename):
raise UploadValidationError(
f"Недопустимый файл: {filename}",
code="invalid_file_type",
)
file_content = _read_file_content(file)
file.seek(0)
if _is_allure_results_archive(filename):
validate_tar_gz_archive(file_content)
archive_filenames.append(filename)
elif is_allure_payload_filename(filename):
loose_payload_count += 1
if archive_filenames:
if len(archive_filenames) > 1:
raise UploadValidationError(
"Ожидается один архив allure-results.tar.gz.",
code="multiple_archives",
)
if len(files) > len(archive_filenames):
raise UploadValidationError(
"Смешанная загрузка архива и отдельных файлов не поддерживается.",
code="mixed_upload",
)
return
if loose_payload_count == 0:
raise UploadValidationError(
"Не найдены файлы allure-results (*-result.json или *-container.json).",
code="missing_allure_results",
)
def get_request_files() -> List[FileStorage]:
"""Возвращает список валидных файлов из запроса."""
files = flask.request.files.getlist("files")
valid_files = [file for file in files if file and file.filename]
if not valid_files:
logger.error("Необходимо загрузить хотя бы один файл")
flask.abort(400, description="Необходимо загрузить хотя бы один файл")
return valid_files
def create_temp_test_result() -> TestResult:
"""Создает временную запись TestResult или завершает запрос с ошибкой."""
try:
new_result = create_temporary_test_result()
logger.info("Создана новая временная запись о запуске автотестов")
return new_result
except DatabaseError as error_msg:
db.session.rollback()
logger.exception("Ошибка при создании записи в базе данных")
flask.abort(500, description=str(error_msg))
def extract_test_run_info(files: Sequence[FileStorage]):
"""Анализирует файлы и возвращает информацию о тестране."""
try:
test_run_info = check_all_tests_passed_run(files)
if not test_run_info:
logger.error("Не удалось извлечь параметры тестрана")
flask.abort(400, description="Ошибка анализа файлов тестрана")
return test_run_info
except Exception as error_msg:
logger.exception("Неизвестная ошибка при анализе тестрана")
flask.abort(500, description=str(error_msg))
def upload_all_files(
run_name: str, files: Sequence[FileStorage]
) -> Tuple[List[str], List[str]]:
"""Загружает файлы и разделяет их на успешные/ошибочные."""
success_files: List[str] = []
error_files: List[str] = []
minio_client.ensure_bucket_exists(const.ALLURE_RESULTS_BUCKET_NAME)
for file in files:
filename = file.filename or "unknown"
if not allowed_file(filename):
logger.error("Недопустимый файл: %s", filename)
error_files.append(filename)
continue
try:
uploaded = process_and_upload_file(run_name, file)
success_files.append(uploaded)
except (DatabaseError, OSError, ValueError) as file_error:
logger.exception("Ошибка обработки файла %s: %s", filename, file_error)
db.session.rollback()
error_files.append(filename)
return success_files, error_files
def get_existing_run_or_abort(result_id: int) -> TestResult:
"""Возвращает TestResult или завершает запрос, если запись недоступна."""
testrun = TestResult.query.get(result_id)
log_and_abort(result_id, testrun)
return testrun
def _validate_upload_file(file: FileStorage) -> str:
"""Убедиться, что объект файла пригоден для дальнейшей обработки."""
if not file or not file.filename:
logger.error("Загружен пустой или поврежденный файл")
raise ValueError("Файл отсутствует или поврежден.")
return file.filename
def _read_file_content(file: FileStorage) -> bytes:
"""Считывает и валидирует содержимое файла."""
file.seek(0)
content = file.read()
if not content:
raise ValueError(f"Файл {file.filename} пустой и не будет загружен.")
return content
def _extract_stand_value(filename: str, file_content: bytes) -> Optional[str]:
"""Пытается извлечь stand из environment.properties."""
if filename != "environment.properties":
return None
try:
content_text = file_content.decode("utf-8", errors="ignore")
except Exception:
logger.exception(
"Не удалось декодировать environment.properties для извлечения stand"
)
return None
stand = extract_stand_from_environment_file(content_text) or None
return stand.strip() if stand else None
def _is_allure_results_archive(filename: str) -> bool:
"""Проверяет, что загружен архив allure-results.tar.gz."""
return filename.lower().endswith(const.ALLURE_RESULTS_ARCHIVE_SUFFIX)
def _extract_stand_from_archive(archive_bytes: bytes) -> Optional[str]:
"""Извлекает stand из environment.properties внутри tar.gz архива."""
tar = open_validated_tar_gz(archive_bytes)
try:
for member in tar.getmembers():
if not member.isfile():
continue
if os.path.basename(member.name.replace("\\", "/")) != "environment.properties":
continue
fileobj = tar.extractfile(member)
if not fileobj:
continue
raw = fileobj.read()
return _extract_stand_value("environment.properties", raw)
finally:
tar.close()
return None
def _persist_detected_stand(run_name: str, detected_stand: str) -> None:
"""Сохраняет значение stand в TestResult, если запись существует."""
try:
test_result: Optional[TestResult] = TestResult.query.filter_by(
run_name=run_name, is_deleted=False
).first()
if not test_result:
logger.warning(
"Не удалось найти TestResult для run_name=%s, stand=%s не сохранён",
run_name,
detected_stand,
)
return
test_result.stand = detected_stand
db.session.add(test_result)
db.session.commit()
logger.info(
"Сохранили stand='%s' для run=%s в TestResult(id=%s)",
detected_stand,
run_name,
test_result.id,
)
except Exception:
logger.exception(
"Ошибка при сохранении stand=%s для run=%s в базе данных",
detected_stand,
run_name,
)
def _upload_file_to_minio(run_name: str, filename: str, file_content: bytes) -> None:
"""Загружает файл в MinIO."""
file_path = f"{run_name}/{filename}"
file_stream = io.BytesIO(file_content)
minio_client.ensure_bucket_exists(const.ALLURE_RESULTS_BUCKET_NAME)
minio_client.put_object(
bucket_name=const.ALLURE_RESULTS_BUCKET_NAME,
file_path=file_path,
file_stream=file_stream,
content_length=len(file_content),
)
def _read_allure_json_from_archive(file: FileStorage) -> list[tuple[str, dict]]:
"""
Считывает JSON-объекты result/container из tar.gz для анализа статуса/времени.
"""
file.seek(0)
archive_bytes = file.read()
if not archive_bytes:
raise ValueError("Архив пустой.")
collected: list[tuple[str, dict]] = []
tar = open_validated_tar_gz(archive_bytes)
try:
for member in tar.getmembers():
if not member.isfile():
continue
basename = os.path.basename(member.name.replace("\\", "/"))
if not (
basename.endswith(const.RESULT_NAMING)
or basename.endswith(const.CONTAINER_NAMING)
):
continue
fileobj = tar.extractfile(member)
if not fileobj:
continue
raw = fileobj.read()
try:
parsed = json.loads(raw.decode(const.ENCODING))
except (UnicodeDecodeError, json.JSONDecodeError):
logger.warning("Некорректный JSON в архиве: %s", member.name)
collected.append((basename, {}))
continue
collected.append((basename, parsed))
finally:
tar.close()
return collected
def _safe_int(value: Optional[int]) -> Optional[int]:
"""Безопасно преобразует значение к int."""
try:
return int(value) if value is not None else None
except (TypeError, ValueError):
return None
def _normalize_status_value(status: Optional[str]) -> Optional[str]:
if status is None:
return None
normalized = str(status).strip().lower()
return normalized or None
def _status_signal_from_value(status: Optional[str]) -> RunStatusSignal:
"""
Возвращает значимость статуса для итогового прогона:
broken > fail > none (passed/skipped/отсутствует).
"""
normalized = _normalize_status_value(status)
if normalized is None:
return "none"
if normalized == const.STATUS_BROKEN:
return const.STATUS_BROKEN
if normalized in {const.STATUS_FAIL, "failed"}:
return const.STATUS_FAIL
if normalized in {
const.STATUS_PASS,
const.STATUS_SKIPPED,
const.STATUS_DESELECTED,
}:
return "none"
# Неизвестные статусы считаем неуспешными, чтобы не показывать ложный "passed".
return const.STATUS_FAIL
def _merge_status_signals(
current: RunStatusSignal, candidate: RunStatusSignal
) -> RunStatusSignal:
if current == const.STATUS_BROKEN or candidate == const.STATUS_BROKEN:
return const.STATUS_BROKEN
if current == const.STATUS_FAIL or candidate == const.STATUS_FAIL:
return const.STATUS_FAIL
return "none"
def _collect_steps_status_signal(steps: Optional[Sequence[dict]]) -> RunStatusSignal:
"""Рекурсивно собирает самый приоритетный статус из шагов."""
if not steps:
return "none"
signal: RunStatusSignal = "none"
for step in steps:
signal = _merge_status_signals(
signal, _status_signal_from_value(step.get(const.STATUS_KEY))
)
signal = _merge_status_signals(
signal, _collect_steps_status_signal(step.get("steps"))
)
if signal == const.STATUS_BROKEN:
return signal
return signal
def _collect_result_status_signal(data: dict) -> RunStatusSignal:
"""Собирает итоговый сигнал статуса из теста, его шагов, befores и afters."""
signal = _status_signal_from_value(data.get(const.STATUS_KEY))
signal = _merge_status_signals(
signal, _collect_steps_status_signal(data.get("steps"))
)
for section in ("befores", "afters"):
for entry in data.get(section, []):
signal = _merge_status_signals(
signal, _status_signal_from_value(entry.get(const.STATUS_KEY))
)
signal = _merge_status_signals(
signal, _collect_steps_status_signal(entry.get("steps"))
)
if signal == const.STATUS_BROKEN:
return signal
return signal
def _run_status_from_signal(signal: RunStatusSignal) -> str:
if signal == const.STATUS_BROKEN:
return const.STATUS_BROKEN
if signal == const.STATUS_FAIL:
return const.STATUS_FAIL
return const.STATUS_PASS
def allowed_file(filename: str) -> bool:
"""
По точке находим расширение файла и проверяем на соответствие списка разрешенных в ALLOWED_EXTENSIONS
"""
if _is_allure_results_archive(filename):
return True
return (
"." in filename
and filename.rsplit(".", 1)[1].lower() in const.ALLOWED_EXTENSIONS
)
def process_and_upload_file(run_name: str, file: FileStorage) -> str:
"""
Валидирует, обрабатывает и загружает файл в MinIO.
Пытается извлечь stand из environment.properties и сохранить его в БД.
"""
try:
filename = _validate_upload_file(file)
logger.info("Тип файла: %s, имя файла: %s", type(file), filename)
file_content = _read_file_content(file)
logger.info("Размер файла %s: %s байт", filename, len(file_content))
if _is_allure_results_archive(filename):
detected_stand = _extract_stand_from_archive(file_content)
_upload_file_to_minio(
run_name, const.ALLURE_RESULTS_ARCHIVE_NAME, file_content
)
else:
detected_stand = _extract_stand_value(filename, file_content)
_upload_file_to_minio(run_name, filename, file_content)
if detected_stand:
logger.info(
"Обнаружен stand='%s' в environment.properties для run=%s",
detected_stand,
run_name,
)
_persist_detected_stand(run_name, detected_stand)
return filename
except OSError:
logger.exception("Ошибка обработки файла", filename=file.filename)
raise
def parse_json_file(file: Any) -> Optional[dict]:
"""Парсит содержимое файла и возвращает json данные."""
try:
content = file.read().decode(const.ENCODING)
return json.loads(content)
except json.JSONDecodeError:
logger.exception("Ошибка при чтении файла", filename=file.filename)
return None
def format_timestamp(timestamp: int) -> str:
"""Форматирует временную метку в миллисекундах в строку по заданному формату."""
return datetime.datetime.fromtimestamp(
timestamp / const.TIMESTAMP_DIVISOR
).strftime(const.DB_DATE_FORMAT)
def check_all_tests_passed_run(
files: Sequence[FileStorage],
) -> dict[str, Optional[str]]:
"""
Проверяет, прошли ли все автотесты успешно, и возвращает статус,
а также время начала и окончания выполнения тестов.
Метод анализирует список файлов, содержащих результаты выполнения автотестов,
и определяет общий статус тестирования по приоритету:
'broken' > 'fail' > 'passed'.
Статус 'skipped' не влияет на итоговый статус тестрана.
Время начала и окончания тестов определяется либо из контейнерного файла,
если он присутствует, либо из файлов с результатами, если контейнерный файл отсутствует.
Параметры:
files (list): Список файлов, содержащих результаты выполнения автотестов.
Каждый элемент списка должен иметь атрибут 'filename', который
используется для идентификации типа файла.
Возвращает:
dict: Словарь с ключами и значениями:
- const.STATUS_KEY: Статус выполнения тестов
('passed', 'fail', 'broken' или 'skipped').
- const.START_RUN_KEY: Время начала выполнения тестов в строковом формате
(или None, если время не определено).
- const.STOP_RUN_KEY: Время окончания выполнения тестов в строковом формате
(или None, если время не определено).
Обработка:
- Инициализирует переменные для статуса тестов и времени начала/окончания.
- Перебирает файлы и анализирует их содержимое.
- Если файл является результатом выполнения тестов, проверяет статус и извлекает
временные метки.
- Если файл является контейнером, извлекает время начала и окончания.
- В случае отсутствия контейнера, использует минимальное и максимальное время
из файлов с результатами.
- Конвертирует временные метки в строковый формат.
- Возвращает словарь с итоговыми данными.
"""
status_signal: RunStatusSignal = "none"
has_non_skipped_status = False
has_skipped_like_status = False
has_payload_entries = False
has_valid_payload_data = False
logger.info("Проверка статусов автотестов внутри данного отчета")
result_start_times: list[int] = []
result_stop_times: list[int] = []
container_start_times: list[int] = []
container_stop_times: list[int] = []
container_start_ms: Optional[int] = None
container_stop_ms: Optional[int] = None
for file in files:
filename = getattr(file, "filename", "") or ""
if _is_allure_results_archive(filename):
archive_entries = _read_allure_json_from_archive(file)
if archive_entries:
has_payload_entries = True
for entry_name, data in archive_entries:
if not data:
status_signal = _merge_status_signals(status_signal, const.STATUS_FAIL)
continue
has_valid_payload_data = True
if entry_name.endswith(const.RESULT_NAMING):
# Итоговый статус прогона считаем только по статусам тест-кейсов.
file_status_signal = _status_signal_from_value(
data.get(const.STATUS_KEY)
)
status_signal = _merge_status_signals(
status_signal, file_status_signal
)
test_status = _normalize_status_value(data.get(const.STATUS_KEY))
if test_status in {const.STATUS_SKIPPED, const.STATUS_DESELECTED}:
has_skipped_like_status = True
elif test_status is not None:
has_non_skipped_status = True
start_ms = _safe_int(data.get(const.START_RUN_KEY))
stop_ms = _safe_int(data.get(const.STOP_RUN_KEY))
if start_ms is not None:
result_start_times.append(start_ms)
if stop_ms is not None:
result_stop_times.append(stop_ms)
elif entry_name.endswith(const.CONTAINER_NAMING):
container_start = _safe_int(data.get(const.START_RUN_KEY))
container_stop = _safe_int(data.get(const.STOP_RUN_KEY))
if container_start is not None:
container_start_times.append(container_start)
if container_stop is not None:
container_stop_times.append(container_stop)
elif filename.endswith(const.RESULT_NAMING):
has_payload_entries = True
data = parse_json_file(file)
if not data:
status_signal = _merge_status_signals(status_signal, const.STATUS_FAIL)
logger.warning("Файл %s не содержит валидный JSON", filename)
else:
has_valid_payload_data = True
file_status_signal = _status_signal_from_value(
data.get(const.STATUS_KEY)
)
status_signal = _merge_status_signals(status_signal, file_status_signal)
if file_status_signal != "none":
logger.info(
"В файле %s обнаружен неуспешный статус теста", filename
)
test_status = _normalize_status_value(data.get(const.STATUS_KEY))
if test_status in {const.STATUS_SKIPPED, const.STATUS_DESELECTED}:
has_skipped_like_status = True
elif test_status is not None:
has_non_skipped_status = True
start_ms = _safe_int(data.get(const.START_RUN_KEY))
stop_ms = _safe_int(data.get(const.STOP_RUN_KEY))
if start_ms is not None:
result_start_times.append(start_ms)
if stop_ms is not None:
result_stop_times.append(stop_ms)
elif filename.endswith(const.CONTAINER_NAMING):
has_payload_entries = True
data = parse_json_file(file)
if data:
has_valid_payload_data = True
container_start = _safe_int(data.get(const.START_RUN_KEY))
container_stop = _safe_int(data.get(const.STOP_RUN_KEY))
if container_start is not None:
container_start_times.append(container_start)
if container_stop is not None:
container_stop_times.append(container_stop)
if container_start_times:
container_start_ms = min(container_start_times)
if container_stop_times:
container_stop_ms = max(container_stop_times)
if container_start_ms is None and result_start_times:
container_start_ms = min(result_start_times)
if container_stop_ms is None and result_stop_times:
container_stop_ms = max(result_stop_times)
start_time_str = (
format_timestamp(container_start_ms) if container_start_ms else None
)
stop_time_str = format_timestamp(container_stop_ms) if container_stop_ms else None
if not has_payload_entries or not has_valid_payload_data:
if status_signal == "none":
status = const.STATUS_NO_DATA
elif (
status_signal == "none"
and has_skipped_like_status
and not has_non_skipped_status
):
status = const.STATUS_SKIPPED
else:
status = _run_status_from_signal(status_signal)
logger.info(
"Итоговый статус тестов: %s, start=%s, stop=%s",
status,
start_time_str,
stop_time_str,
)
return {
const.STATUS_KEY: status,
const.START_RUN_KEY: start_time_str,
const.STOP_RUN_KEY: stop_time_str,
}
def create_temporary_test_result():
"""
Создает временную запись тестового результата в базе данных, при необходимости создавая соответствующую таблицу.
Возвращает:
- Объект `TestResult`, представляющий созданную запись о тестовом результате.
Поведение:
- Проверяет наличие таблицы `TestResult`. Если она не существует, создает её.
- Создает новую запись `TestResult` с параметрами по умолчанию:
- `run_name`: имя запуска формируется на основе константы `DEFAULT_RUN_NAME` и текущего времени.
- `start_date`: `None`, так как результаты теста еще не распарсили.
- `end_date`: `None`, так как результаты теста еще не распарсили.
- `status`: устанавливается в `PENDING_STATUS`, показывая, что еще нет результата.
- `file_link`: `None` изначально, поскольку ссылка на файл результата задается позже.
- Добавляет новую запись в сеанс базы данных и сохраняет изменения.
Исключения:
- В случае ошибки базы данных, выполняет откат текущего сеанса и логирует ошибку,
затем повторно вызывает исключение `DatabaseError`.
"""
try:
# Создаем таблицу, если она еще не создана
inspector = inspect(db.engine)
if not inspector.has_table(TestResult.__tablename__):
db.create_all()
# Создаем новый тестовый результат
new_result = TestResult(
run_name=f"{const.DEFAULT_RUN_NAME}_{datetime.datetime.now()}",
start_date=None,
end_date=None,
status=const.PENDING_STATUS,
)
# Добавляем и коммитим новую запись
db.session.add(new_result)
db.session.commit()
return new_result
except DatabaseError as error_msg:
# Обработка ошибки базы данных
db.session.rollback()
logger.exception("Ошибка при создании записи в базе данных", exc_info=error_msg)
raise
def update_test_result(new_result: "TestResult", test_run_info: dict) -> None:
"""Обновляет параметры тестового запуска в БД."""
run_id = new_result.id
run_name = f"run_{run_id}_{test_run_info.get(const.START_RUN_KEY)}"
new_result.run_name = run_name
new_result.status = test_run_info.get(const.STATUS_KEY)
new_result.start_date = test_run_info.get(const.START_RUN_KEY)
new_result.end_date = test_run_info.get(const.STOP_RUN_KEY)
db.session.commit()
def check_files_size(files: list, max_size: int = None) -> bool:
"""
Проверка размера загружаемых файлов.
files - список файлов для проверки.
max_size - максимальный допустимый размер в байтах.
"""
# Получение размера по умолчанию из переменной или установки стандартного 50 МБ
if max_size is None:
max_size = const.MAX_FILE_SIZE # 50 MB по умолчанию
# Считаем общий размер файлов
total_size = 0
for file in files:
file.seek(0, 2) # Переместить курсор в конец файла
total_size += file.tell() # Получить текущую позицию (размер в байтах)
file.seek(0) # Сбросить курсор на начало
# Проверка превышения лимита
if total_size > max_size:
logger.error(
"Общий размер загружаемых файлов превышен",
total_size=total_size,
max_size=max_size,
)
abort(400, description="Общий размер загружаемых файлов превышает допустимый")
logger.info(
"Размер файлов в пределах допустимого", total_size=total_size, max_size=max_size
)
return True
def _format_datetime(value: Optional[datetime.datetime]) -> Optional[str]:
if not value:
return None
try:
# Считаем входные даты в UTC, приводим к часовому поясу Москвы
moscow_tz = ZoneInfo("Europe/Moscow")
if value.tzinfo is None:
value = value.replace(tzinfo=datetime.timezone.utc)
value = value.astimezone(moscow_tz)
except Exception:
# Если что-то пошло не так, возвращаем как есть
pass
return value.strftime(const.VIEW_DATE_FORMAT)
def _serialize_test_result(result: TestResult) -> Dict[str, Any]:
"""Приводит TestResult к словарю для фронтенда."""
return {
"id": result.id,
"run_name": result.run_name,
"start_date": _format_datetime(result.start_date),
"end_date": _format_datetime(result.end_date),
"stand": result.stand or None,
"status": result.status,
}
def extract_filter_values(param_name: str) -> List[str]:
"""
Возвращает очищенный список значений фильтра из query string.
Удаляет прочерки и пустые значения, поддерживает нотации name и name[].
"""
values = flask.request.args.getlist(param_name)
if not values:
values = flask.request.args.getlist(f"{param_name}[]")
if len(values) == 1 and "," in (values[0] or ""):
values = [chunk.strip() for chunk in values[0].split(",")]
cleaned: List[str] = []
seen = set()
for value in values:
if value is None:
continue
candidate = value.strip()
if not candidate or candidate == "-":
continue
if candidate in seen:
continue
seen.add(candidate)
cleaned.append(candidate)
return cleaned
def _normalize_filter_values(values: Optional[Sequence[str]]) -> List[str]:
"""Подготавливает список значений для безопасного использования в запросах."""
if not values:
return []
normalized: List[str] = []
seen = set()
for value in values:
if value is None:
continue
cleaned = str(value).strip()
if not cleaned or cleaned == "-":
continue
if cleaned in seen:
continue
seen.add(cleaned)
normalized.append(cleaned)
return normalized
def _collect_distinct_column_values(column) -> List[str]:
"""Возвращает уникальные значения столбца для не удалённых записей."""
rows = (
db.session.query(column)
.filter(
TestResult.is_deleted.is_(False),
column.isnot(None),
column != "",
column != "-",
)
.distinct()
.order_by(column.asc())
.all()
)
values: List[str] = []
seen = set()
for (value,) in rows:
if value is None:
continue
cleaned = value.strip() if isinstance(value, str) else value
if not cleaned or cleaned == "-":
continue
if cleaned in seen:
continue
seen.add(cleaned)
values.append(cleaned)
return values
def _get_available_report_filters() -> Dict[str, List[str]]:
"""Формирует справочник доступных фильтров для отчётов."""
return {
"statuses": _collect_distinct_column_values(TestResult.status),
"stands": _collect_distinct_column_values(TestResult.stand),
}
def _has_older_runs(oldest_id: int) -> bool:
"""Проверяет наличие более старых записей по id."""
return (
TestResult.query.filter(
TestResult.is_deleted.is_(False), TestResult.id < oldest_id
)
.order_by(TestResult.id.desc())
.limit(1)
.first()
is not None
)
def _has_newer_runs(newest_id: int) -> bool:
"""Проверяет наличие более новых записей по id."""
return (
TestResult.query.filter(
TestResult.is_deleted.is_(False), TestResult.id > newest_id
)
.order_by(TestResult.id.asc())
.limit(1)
.first()
is not None
)
def _parse_date_param(
value: Optional[str], end_of_day: bool = False
) -> Optional[datetime.datetime]:
"""
Парсит строку даты в формате YYYY-MM-DD в datetime.
Если end_of_day=True, устанавливает время 23:59:59, иначе 00:00:00.
"""
if not value:
return None
try:
date = datetime.datetime.strptime(value.strip(), "%Y-%m-%d")
if end_of_day:
date = date.replace(hour=23, minute=59, second=59)
return date
except ValueError:
logger.warning(f"Некорректный формат даты: {value}")
return None
def fetch_reports(
cursor: Optional[int],
limit: int,
direction: str = "next",
statuses: Optional[Sequence[str]] = None,
stands: Optional[Sequence[str]] = None,
start_date_from: Optional[str] = None,
start_date_to: Optional[str] = None,
) -> Dict[str, Any]:
"""
Возвращает страницу отчетов с курсорной пагинацией.
direction: 'next' (старее) или 'prev' (новее).
statuses/stands — списки значений для фильтрации (множество значений).
start_date_from/start_date_to — фильтр по дате старта (формат YYYY-MM-DD).
"""
if direction not in {"next", "prev"}:
logger.error(f"Получено некорректное направление пагинации: {direction}")
raise ValueError("Направление должно быть либо 'next' или 'prev'")
# Парсим даты
date_from = _parse_date_param(start_date_from, end_of_day=False)
date_to = _parse_date_param(start_date_to, end_of_day=True)
# Валидация: дата "с" должна быть меньше или равна дате "до"
if date_from and date_to and date_from > date_to:
raise ValueError("Дата 'с' должна быть меньше или равна дате 'до'")
available_filters = _get_available_report_filters()
normalized_statuses = _normalize_filter_values(statuses)
normalized_stands = _normalize_filter_values(stands)
base_query = TestResult.query.filter_by(is_deleted=False)
if cursor:
if direction == "next":
base_query = base_query.filter(TestResult.id < cursor)
else:
base_query = base_query.filter(TestResult.id > cursor)
if normalized_statuses:
base_query = base_query.filter(TestResult.status.in_(normalized_statuses))
if normalized_stands:
base_query = base_query.filter(TestResult.stand.in_(normalized_stands))
# Фильтр по дате старта
if date_from:
base_query = base_query.filter(TestResult.start_date >= date_from)
if date_to:
base_query = base_query.filter(TestResult.start_date <= date_to)
order_column = TestResult.id.desc()
if direction == "prev":
order_column = TestResult.id.asc()
results = base_query.order_by(order_column).limit(limit + 1).all()
has_more_in_direction = len(results) > limit
items = results[:limit]
if direction == "prev":
items = list(reversed(items))
serialized = [_serialize_test_result(item) for item in items]
if not items:
return {
"items": [],
"next_cursor": None,
"prev_cursor": None,
"has_next": False,
"has_prev": False,
"filters": available_filters,
}
newest_id = items[0].id
oldest_id = items[-1].id
has_prev = _has_newer_runs(newest_id)
has_next = _has_older_runs(oldest_id)
# Обновляем флаги, учитывая результат текущего запроса
if direction == "next" and cursor:
has_prev = True
if direction == "prev":
has_next = True
if not has_more_in_direction:
has_prev = False
return {
"items": serialized,
"next_cursor": oldest_id if has_next else None,
"prev_cursor": newest_id if has_prev else None,
"has_next": has_next,
"has_prev": has_prev,
"filters": available_filters,
}
def log_reports(results_present: bool) -> None:
"""Логирует состояние страницы отчётов."""
if results_present:
logger.info("Обработан запрос на страницу списка отчетов", status_code=200)
else:
logger.info(
"Обработан запрос на страницу списка отчетов, список отчетов пуст",
status_code=200,
)
def generate_and_upload_report(run_name: str) -> None:
"""
Генерирует и загружает allure-report в MinIO.
Аргумент run_name - название тест-рана, используемое для директории allure-result и allure-report.
"""
temp_dir = tempfile.mkdtemp() # Создаем временную директорию для результатов
report_dir = tempfile.mkdtemp() # Создаем временную директорию для отчёта
try:
logger.info("Начало скачивания файлов из MinIO")
download_allure_results(run_name, temp_dir)
results_dir_for_generation = _resolve_allure_results_dir(temp_dir)
logger.info("Начало генерации allure-report")
generate_allure_report(results_dir_for_generation, report_dir)
logger.info("Загрузка allure-report в MinIO")
upload_report_to_minio(run_name, report_dir)
finally:
logger.info("Очистка временных директорий")
cleanup_temporary_directories([temp_dir, report_dir])
def _resolve_allure_results_dir(base_dir: str) -> str:
"""
Находит директорию с allure results.
"""
def _is_allure_payload_name(name: str) -> bool:
return name.endswith(const.RESULT_NAMING) or name.endswith(
const.CONTAINER_NAMING
)
base_result_files = [
name
for name in os.listdir(base_dir)
if os.path.isfile(os.path.join(base_dir, name))
and _is_allure_payload_name(name)
]
if base_result_files:
return base_dir
directories_with_results: set[str] = set()
for root, _, files in os.walk(base_dir):
for name in files:
if _is_allure_payload_name(name):
directories_with_results.add(root)
break
if len(directories_with_results) == 1:
resolved_dir = next(iter(directories_with_results))
logger.info(
"Определена вложенная директория allure-results для генерации",
base_dir=base_dir,
resolved_dir=resolved_dir,
)
return resolved_dir
if not directories_with_results:
raise RuntimeError(
"После распаковки архива не найдены файлы результатов Allure "
"(*-result.json или *-container.json)."
)
raise RuntimeError(
"После распаковки обнаружено несколько директорий с файлами Allure "
"(*-result.json/*-container.json); "
"невозможно однозначно выбрать источник для генерации отчета."
)
def download_allure_results(
allure_results_directory: str, destination_dir: str
) -> None:
"""
Загружает результаты Allure из MinIO в указанную директорию.
Аргументы:
allure_results_directory - директория в MinIO с allure-results.
destination_dir - путь к локальной директории для сохранения allure-results.
"""
archive_object_name = (
f"{allure_results_directory}/{const.ALLURE_RESULTS_ARCHIVE_NAME}"
)
# Формат хранения: один tar.gz на прогон.
try:
minio_client.stat_object(const.ALLURE_RESULTS_BUCKET_NAME, archive_object_name)
archive_local_path = os.path.join(
destination_dir, const.ALLURE_RESULTS_ARCHIVE_NAME
)
minio_client.download_file(
const.ALLURE_RESULTS_BUCKET_NAME, archive_object_name, archive_local_path
)
with open(archive_local_path, "rb") as archive_file:
safe_extract_tar_gz_bytes(archive_file.read(), destination_dir)
try:
os.remove(archive_local_path)
except OSError:
logger.warning("Не удалось удалить временный архив %s", archive_local_path)
logger.info("Архив allure-results распакован", object=archive_object_name)
return
except S3Error:
logger.info(
"Архив allure-results.tar.gz не найден, fallback на legacy-структуру",
run_name=allure_results_directory,
)
# Legacy fallback: старые прогоны с набором отдельных файлов.
for obj in minio_client.list_objects(
const.ALLURE_RESULTS_BUCKET_NAME, prefix=f"{allure_results_directory}/"
):
file_path = os.path.join(destination_dir, obj.object_name.split("/")[-1])
minio_client.download_file(
const.ALLURE_RESULTS_BUCKET_NAME, obj.object_name, file_path
)
if os.path.exists(file_path):
logger.info(
f"Файл {file_path} загружен, размер: {os.path.getsize(file_path)} байт"
)
else:
logger.error(f"Ошибка: Файл {file_path} не загружен")
def generate_allure_report(result_dir_path: str, report_dir_path: str) -> None:
"""
Генерирует Allure-отчёт на основе результатов тестов.
На основе предоставленной директории с результатами тестов, функция выполняет
системную команду для генерации HTML-отчёта в указанной директории. В случае
ошибки генерации, логируется сообщение об ошибке и выбрасывается исключение.
result_dir_path - путь к директории, содержащей результаты тестов allure-results.
report_dir_path - путь к директории, куда будет сохранен HTML-отчёт Allure
"""
command = (
f"allure generate {result_dir_path} "
f"-o {report_dir_path} --clean --single-file"
)
try:
subprocess.run(command, shell=True, text=True, check=True)
except subprocess.CalledProcessError as error:
error_msg = (
"Нет вывода ошибки выполнения команды"
if not error.stderr
else error.stderr.strip()
)
logger.exception(
"Ошибка при генерации Allure-отчета",
description=error_msg,
error_code=error.returncode,
)
raise RuntimeError("Не удалось сгенерировать Allure-отчёт") from error
def upload_report_to_minio(run_name: str, report_dir: str) -> None:
"""
Загружает HTML-отчёт Allure в MinIO.
run_name - ммя тест-запуска для файла allure-report.
report_dir - путь к временной директории, где находится HTML-отчёт.
"""
final_report_file = os.path.join(report_dir, "index.html")
with open(final_report_file, "rb") as file:
minio_client.ensure_bucket_exists(const.ALLURE_REPORTS_BUCKET_NAME)
minio_client.put_object(
const.ALLURE_REPORTS_BUCKET_NAME,
f"{run_name}.html",
file,
os.path.getsize(final_report_file),
)
def cleanup_temporary_directories(directories: list) -> None:
"""
Удаляет указанные временные директории.
"""
for directory in directories:
if os.path.exists(directory):
shutil.rmtree(directory)
logger.info(f"Временная директория {directory} удалена")
def report_exists(run_name: str) -> bool:
"""
Проверяет наличие отчёта Allure в хранилище MinIO.
Данный метод используется для проверки существования HTML-отчета
о тестировании с определённым именем в указанном бакете MinIO.
Если отчёт существует, метод возвращает `True`, в противном
случае – `False`.
"""
try:
minio_client.minio_client.stat_object(
const.ALLURE_REPORTS_BUCKET_NAME, f"{run_name}.html"
)
return True
except S3Error:
return False
def log_and_abort(result_id: int, testrun) -> None:
"""
Логирование и окончание запроса, если тестран не существует или помечен как удаленный.
"""
if not testrun:
logger.error(f"Test run с ID {result_id} не найден.")
abort(404, description=f"Test run с ID {result_id} не найден.")
if testrun.is_deleted:
logger.warning(f"Test run {result_id} помечен как удаленный.")
abort(404, description=f"Test run {testrun.run_name} помечен как удаленный.")
def get_or_generate_report(run_name: str):
"""
Получение или генерация allure-report
Загрузка allure-report в MinIO
"""
if report_exists(run_name):
return minio_client.minio_client.get_object(
const.ALLURE_REPORTS_BUCKET_NAME, f"{run_name}.html"
)
generate_and_upload_report(run_name)
return minio_client.minio_client.get_object(
const.ALLURE_REPORTS_BUCKET_NAME, f"{run_name}.html"
)
руты
@bp.route("/upload", methods=["POST"])
def upload_results():
"""
API-метод для загрузки файлов и создания тестового запуска
"""
files = testrun_helpers.get_request_files()
testrun_helpers.check_files_size(files)
try:
testrun_helpers.validate_upload_files(files)
except testrun_helpers.UploadValidationError as exc:
logger.warning(
"Валидация загружаемых файлов не пройдена",
error_code=exc.code,
message=exc.message,
)
return flask.jsonify(
{
"success": False,
"error": exc.code,
"message": exc.message,
}
), 400
new_result = testrun_helpers.create_temp_test_result()
test_run_info = testrun_helpers.extract_test_run_info(files)
try:
testrun_helpers.update_test_result(new_result, test_run_info)
logger.info(f"Обновлены данные тестрана с ID: {new_result.id}")
except DatabaseError as error_msg:
db.session.rollback()
logger.exception("Ошибка при сохранении статуса тестрана в базу данных")
flask.abort(500, description=str(error_msg))
success_files, error_files = testrun_helpers.upload_all_files(
new_result.run_name, files
)
if success_files:
logger.info(f"Успешно загруженные файлы в MinIO: {', '.join(success_files)}")
if error_files:
logger.warning(f"Ошибка обработки следующих файлов: {', '.join(error_files)}")
flask.abort(500, description="Некоторые файлы не были успешно обработаны")
testrun = testrun_helpers.get_existing_run_or_abort(new_result.id)
testrun_helpers.get_or_generate_report(testrun.run_name)
response = flask.jsonify(
{
"success": True,
"run_id": new_result.id,
"message": "Файлы успешно загружены",
}
)
response_code = 201
logger.info("Файлы успешно загружены", status_code=response_code)
return response, response_code
static\js\reports.js
} else if (statusValue === "skipped") {
statusClass = "text-gray-500";
} else if (statusValue === "no_data") {
statusClass = "text-gray-400";
}
