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

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__ = [
    "SNAPSHOT_DIR",
    "SyncReport",
    "VOLATILE",
    "golden_path",
    "normalize_result",
    "render",
    "snapshot_text",
    "sync_snapshots",
    "unified_diff",
]

SNAPSHOT_DIR = "expected_osv_parse"
GOLDEN_DIR = "golden"
OSV_DIR = "osv"
JSON_INDENT = 2
# Поля, которые меняются от запуска к запуску: идентификаторы заглушек, версия сервиса, фактическая длительность
VOLATILE: tuple[str | tuple[str, str], ...] = ("job_id", "file_id", ("service", "version"), ("metrics", "duration_ms"))


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)
    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`.
    Для изменившегося снимка в `report.diffs` — unified diff; для нового — весь текст как добавленный.
    """
    report = SyncReport()
    expected: set[Path] = set()
    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 not in expected:
            report.orphans.append(stale.relative_to(fixtures_root))
            if write:
                stale.unlink()
    return report