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"
SYNTHETIC_DIR = "synthetic"
REAL_PART = "real"
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 golden_path(fixtures_root: Path, file: Path) -> Path:
    """`golden/expected_osv_parse/<путь файла относительно osv/ без расширения>.json` (зеркало дерева `osv/`)."""
    relative = file.relative_to(fixtures_root / OSV_DIR)
    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/`). Если в `files` нет ни одного файла части (каталог отсутствует или пуст — например, fixtures без
    синтетики), её снимки не лишние: они попадают в `report.protected` и не удаляются (T12.3).
    Для изменившегося снимка в `report.diffs` — unified diff; для нового — весь текст как добавленный.
    """
    report = SyncReport()
    expected: set[Path] = set()
    files = list(files)
    present_parts = {_source_part(file.relative_to(fixtures_root / OSV_DIR)) 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` или `real`."""
    return SYNTHETIC_DIR if relative.parts and relative.parts[0] == SYNTHETIC_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,
        )