Загрузка данных
return stand
def _extract_description(files: Sequence[FileStorage]) -> Optional[str]:
"""Достаёт description из environment.properties"""
description: Optional[str] = None
for file_storage in files:
filename = file_storage.filename or ""
_rewind(file_storage)
content = file_storage.stream.read()
_rewind(file_storage)
if testrun_helpers._is_allure_results_archive(filename):
description = (
testrun_helpers._extract_description_from_archive(content)
or description
)
elif os.path.basename(filename) == "environment.properties":
description = (
testrun_helpers._extract_description_value(
"environment.properties", content
)
or description
)
return description
def _build_file_storages(
minio_client: MinioClient, objects: Iterable[MinioObjectLike]
) -> list[FileStorage]:
"""Скачивает объекты одного прогона и готовит их к check_all_tests_passed_run."""
files: list[FileStorage] = []
for obj in objects:
if not obj.object_name:
continue
payload = _read_object_bytes(minio_client, obj.object_name)
if not payload:
continue
files.append(_as_file_storage(obj.object_name, payload))
return files
def _created_at_from_objects(objects: Sequence[MinioObjectLike]) -> datetime.datetime:
"""created_at прогона: самый поздний LastModified объектов в префиксе."""
timestamps = [_to_naive_utc(obj.last_modified) for obj in objects]
known = [item for item in timestamps if item is not None]
if not known:
return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
return max(known)
def _print_plan(
run_name: str,
info: Mapping[str, Any],
stand: Optional[str],
description: Optional[str],
created_at: datetime.datetime,
action: str,
) -> None:
"""Печатает одну строку плана вставки / результата."""
stats = info.get(const.STATUS_STATS_KEY) or {}
print(
f"{action} run_name={run_name} status={info.get(const.STATUS_KEY)} "
f"stand={stand!r} description={description!r} "
f"start={info.get(const.START_RUN_KEY)} "
f"stop={info.get(const.STOP_RUN_KEY)} "
f"passed={stats.get(const.STATUS_PASSED, 0)} "
f"failed={stats.get(const.STATUS_FAILED, 0)} "
f"broken={stats.get(const.STATUS_BROKEN, 0)} "
f"skipped={stats.get(const.STATUS_SKIPPED, 0)} "
f"created_at={created_at}"
)
def rebuild(dry_run: bool) -> None:
"""Сканирует MinIO и вставляет отсутствующие прогоны в testrun_results."""
app = create_app()
minio_client = MinioClient()
with app.app_context():
existing = {
name for (name,) in db.session.query(TestResult.run_name).all()
}
groups = _group_objects_by_run(minio_client)
print(f"Найдено префиксов в MinIO: {len(groups)}")
print(f"Уже есть в testrun_results: {len(existing)}")
skipped = 0
inserted = 0
empty = 0
for run_name in sorted(groups):
objects = groups[run_name]
if run_name in existing:
print(f"SKIP already exists run_name={run_name}")
skipped += 1
continue
files = _build_file_storages(minio_client, objects)
if not files:
print(f"SKIP empty prefix run_name={run_name}")
empty += 1
continue
info = testrun_helpers.check_all_tests_passed_run(files)
stand = _extract_stand(files)
description = _extract_description(files)
created_at = _created_at_from_objects(objects)
stats = info.get(const.STATUS_STATS_KEY) or {}
if dry_run:
_print_plan(run_name, info, stand, description, created_at, "DRY-RUN insert")
inserted += 1
continue
result = TestResult(
run_name=run_name,
start_date=_parse_run_datetime(info.get(const.START_RUN_KEY)),
end_date=_parse_run_datetime(info.get(const.STOP_RUN_KEY)),
stand=stand,
description=description,
status=info.get(const.STATUS_KEY),
passed_count=stats.get(const.STATUS_PASSED, 0),
failed_count=stats.get(const.STATUS_FAILED, 0),
broken_count=stats.get(const.STATUS_BROKEN, 0),
skipped_count=stats.get(const.STATUS_SKIPPED, 0),
created_at=created_at,
is_deleted=False,
)
db.session.add(result)
db.session.commit()
_print_plan(run_name, info, stand, description, created_at, "INSERTED")
inserted += 1
print(
f"Готово. skipped={skipped} empty={empty} "
f"{'would_insert' if dry_run else 'inserted'}={inserted}"
)