arm-tabular-parser/api/tests/golden_helpers.py

"""Golden-снимки разбора ОСВ (design D-10): результат парсера в `fixtures/golden/expected_osv_parse/`.

Снимок — регрессия, а не оракул: он фиксирует то, что парсер выдаёт сейчас. Корректность проверяют ручная сверка
«Итого» трёх файлов (отчёт задачи) и property-тест `tests/fixtures/test_osv_property.py`.

Результат — тот же, что печатает CLI `src.parser.cli` (`core.worker.build_result` → `to_wire`), без полей, меняющихся
от запуска к запуску (`VOLATILE`). Порядок ключей и `warnings` не трогается: он значим для diff.
"""

import copy
import difflib
import json
from collections.abc import Iterable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import pytest

from src.contract import ResultStatus, ServiceInfo
from src.core.config import Settings
from src.core.worker import build_result
from src.parser import Parser
from src.parser.cli import STUB_ID, _error_payload, _file_type

__all__ = [
    "MISSING",
    "SNAPSHOT_DIR",
    "SyncReport",
    "VOLATILE",
    "check_snapshot",
    "golden_path",
    "normalize_result",
    "render",
    "snapshot_text",
    "sync_snapshots",
    "unified_diff",
]

SNAPSHOT_DIR = "expected_osv_parse"
GOLDEN_DIR = "golden"
OSV_DIR = "osv"
OSV_DKB_DIR = "osv-dkb"
SYNTHETIC_DIR = "synthetic"
REAL_PART = "real"
DKB_PART = "dkb"
JSON_INDENT = 2
# Поля, которые меняются от запуска к запуску: идентификаторы заглушек, версия сервиса, фактическая длительность
VOLATILE: tuple[str | tuple[str, str], ...] = ("job_id", "file_id", ("service", "version"), ("metrics", "duration_ms"))
MISSING = "нет снимка: выполните make golden-update ({path})"
MAX_DIFF_CHARS = 6000


def normalize_result(result: dict[str, Any]) -> dict[str, Any]:
    """Копия результата без `VOLATILE`-полей; порядок остальных ключей и элементов списков сохраняется."""
    normalized = copy.deepcopy(result)
    for volatile in VOLATILE:
        if isinstance(volatile, str):
            normalized.pop(volatile, None)
        else:
            parent, key = volatile
            inner = normalized.get(parent)
            if isinstance(inner, dict):
                inner.pop(key, None)
    return normalized


def _mirrored(fixtures_root: Path, file: Path) -> Path:
    """Путь файла в зеркале снимков: относительно `osv/`, а для пакета ДКБ — `osv-dkb/<путь внутри osv-dkb>`."""
    dkb = fixtures_root / OSV_DKB_DIR
    if dkb in file.parents:
        return Path(OSV_DKB_DIR) / file.relative_to(dkb)
    return file.relative_to(fixtures_root / OSV_DIR)


def golden_path(fixtures_root: Path, file: Path) -> Path:
    """`golden/expected_osv_parse/<путь файла без расширения>.json`: зеркало дерева `osv/`, а файлы пакета ДКБ —
    в подкаталоге `osv-dkb/`."""
    relative = _mirrored(fixtures_root, file)
    return fixtures_root / GOLDEN_DIR / SNAPSHOT_DIR / relative.parent / f"{relative.stem}.json"


def render(parser: Parser, file: Path) -> dict[str, Any]:
    """Результат разбора файла, как его выдаёт CLI: `to_wire` с фиксированными `job_id`/`file_id`, затем нормализация.

    Ошибка разбора выглядит как в CLI — `{"status": "error", "error": {...}}`; файл с неизвестным расширением — ошибка
    использования (`ValueError`).
    """
    file_type = _file_type(file, None)
    if file_type is None:
        raise ValueError(f"неизвестное расширение файла {file.suffix!r}: {file}")
    outcome = parser.parse(file.read_bytes(), file_type)
    if outcome.status is ResultStatus.ERROR:
        return normalize_result(_error_payload(outcome))
    service = ServiceInfo.model_validate(
        {"name": Settings.model_fields["SERVICE_NAME"].default, "version": Settings.model_fields["VERSION"].default}
    )
    result = build_result(outcome, job_id=STUB_ID, file_id=STUB_ID, service=service, metrics=outcome.metrics)
    return normalize_result(result.to_wire())


def snapshot_text(normalized: dict[str, Any]) -> str:
    """Текст снимка: как вывод CLI — без экранирования кириллицы, отступ 2, порядок ключей модели, «\\n» в конце."""
    return json.dumps(normalized, ensure_ascii=False, indent=JSON_INDENT) + "\n"


def unified_diff(stored: str, current: str, *, name: str) -> str:
    """Unified diff «снимок в fixtures → разбор сейчас»; пустая строка, если тексты равны."""
    return "".join(
        difflib.unified_diff(
            stored.splitlines(keepends=True),
            current.splitlines(keepends=True),
            fromfile=f"{name} (в fixtures)",
            tofile=f"{name} (разбор сейчас)",
        )
    )


@dataclass
class SyncReport:
    """Итог сверки/обновления снимков: пути относительно корня fixtures."""

    created: list[Path] = field(default_factory=list)
    changed: list[Path] = field(default_factory=list)
    orphans: list[Path] = field(default_factory=list)
    protected: list[Path] = field(default_factory=list)  # снимки без исходного каталога: не удаляются (T12.3)
    unchanged: int = 0
    diffs: list[str] = field(default_factory=list)

    @property
    def differs(self) -> bool:
        return bool(self.created or self.changed or self.orphans)


def sync_snapshots(fixtures_root: Path, files: Iterable[Path], parser: Parser, *, write: bool) -> SyncReport:
    """Сверяет снимки `files` с текущим разбором; при `write` — перезаписывает изменившиеся и удаляет лишние.

    Лишний снимок — файл в `golden/expected_osv_parse/`, которому не соответствует ни один файл из `files`.
    Снимки делятся на части по исходному каталогу: `synthetic/` (синтетика `osv/synthetic`), `osv-dkb/` (пакет ДКБ,
    только файлы до `GOLDEN_MAX_BYTES`) и остальные (реальные файлы `osv/`). Если в `files` нет ни одного файла части
    (каталог отсутствует или пуст — например, fixtures без синтетики), её снимки не лишние: они попадают в
    `report.protected` и не удаляются (T12.3).
    Для изменившегося снимка в `report.diffs` — unified diff; для нового — весь текст как добавленный.
    """
    report = SyncReport()
    expected: set[Path] = set()
    files = list(files)
    present_parts = {_source_part(_mirrored(fixtures_root, file)) for file in files}
    for file in files:
        target = golden_path(fixtures_root, file)
        expected.add(target)
        current = snapshot_text(render(parser, file))
        name = target.relative_to(fixtures_root).as_posix()
        if not target.is_file():
            report.created.append(target.relative_to(fixtures_root))
            report.diffs.append(unified_diff("", current, name=name))
        else:
            stored = target.read_text(encoding="utf-8")
            if stored == current:
                report.unchanged += 1
                continue
            report.changed.append(target.relative_to(fixtures_root))
            report.diffs.append(unified_diff(stored, current, name=name))
        if write:
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(current, encoding="utf-8")
    snapshots = fixtures_root / GOLDEN_DIR / SNAPSHOT_DIR
    for stale in sorted(snapshots.rglob("*.json")) if snapshots.is_dir() else []:
        if stale in expected:
            continue
        if _source_part(stale.relative_to(snapshots)) not in present_parts:
            report.protected.append(stale.relative_to(fixtures_root))
            continue
        report.orphans.append(stale.relative_to(fixtures_root))
        if write:
            stale.unlink()
    return report


def _source_part(relative: Path) -> str:
    """Часть дерева по пути в зеркале (относительно `osv/` или каталога снимков): `synthetic`, `dkb` или `real`."""
    first = relative.parts[0] if relative.parts else ""
    if first == SYNTHETIC_DIR:
        return SYNTHETIC_DIR
    return DKB_PART if first == OSV_DKB_DIR else REAL_PART


def check_snapshot(fixtures_root: Path, file: Path, parser: Parser) -> None:
    """Сверка файла со снимком для golden-теста: нет снимка или расхождение — падение теста с понятным текстом."""
    snapshot = golden_path(fixtures_root, file)
    if not snapshot.is_file():
        pytest.fail(MISSING.format(path=snapshot.relative_to(fixtures_root).as_posix()), pytrace=False)
    stored = snapshot.read_text(encoding="utf-8")
    current = snapshot_text(render(parser, file))
    if stored != current:
        diff = unified_diff(stored, current, name=snapshot.relative_to(fixtures_root).as_posix())
        if len(diff) > MAX_DIFF_CHARS:
            diff = (
                diff[:MAX_DIFF_CHARS] + f"\n… (diff обрезан, всего {len(diff)} символов; полный — make golden-update)"
            )
        pytest.fail(
            "результат разбора отличается от снимка (намеренная правка парсера — make golden-update и коммит "
            f"в fixtures с просмотром diff):\n{diff}",
            pytrace=False,
        )