"""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, require_statements_version from src.parser.cli import STUB_ID, _error_payload, _file_type __all__ = [ "MISSING", "SNAPSHOT_DIR", "STATEMENT_SNAPSHOT_DIR", "SyncReport", "VOLATILE", "check_snapshot", "golden_path", "normalize_result", "render", "snapshot_text", "sync_snapshots", "unified_diff", ] SNAPSHOT_DIR = "expected_osv_parse" STATEMENT_SNAPSHOT_DIR = "expected_statement_parse" STATEMENTS_DIR = "statements-xlsx" GOLDEN_DIR = "golden" OSV_DIR = "osv" OSV_DKB_DIR = "osv-dkb" SYNTHETIC_DIR = "synthetic" REAL_PART = "real" DKB_PART = "dkb" STATEMENTS_PART = "statements" STATEMENTS_SYNTHETIC_PART = "statements-synthetic" JSON_INDENT = 2 # Поля, которые меняются от запуска к запуску: идентификаторы заглушек, версия сервиса, фактическая длительность VOLATILE: tuple[str | tuple[str, str], ...] = ( "job_id", "file_id", ("service", "version"), ("service", "statements_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 _is_statement(fixtures_root: Path, file: Path) -> bool: return fixtures_root / STATEMENTS_DIR in file.parents def _mirrored(fixtures_root: Path, file: Path) -> Path: """Путь файла в зеркале снимков: для форм — относительно `statements-xlsx/`, для пакета ДКБ — `osv-dkb/<путь внутри osv-dkb>`, иначе относительно `osv/`.""" if _is_statement(fixtures_root, file): return file.relative_to(fixtures_root / STATEMENTS_DIR) 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/`; для форм `statements-xlsx/**` — `golden/expected_statement_parse/<путь>.json`.""" relative = _mirrored(fixtures_root, file) snapshot_dir = STATEMENT_SNAPSHOT_DIR if _is_statement(fixtures_root, file) else SNAPSHOT_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, "statements_version": require_statements_version(), } ) 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 = {_file_part(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") for snapshot_dir in (SNAPSHOT_DIR, STATEMENT_SNAPSHOT_DIR): 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 _snapshot_part(snapshot_dir, 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 _file_part(fixtures_root: Path, file: Path) -> str: """Часть дерева файла: `synthetic`, `dkb`, `real` (ОСВ), `statements` или `statements-synthetic` (формы).""" mirrored = _mirrored(fixtures_root, file) if _is_statement(fixtures_root, file): return _statement_part(mirrored) return _osv_part(mirrored) def _snapshot_part(snapshot_dir: str, relative: Path) -> str: """Часть дерева по пути снимка относительно его каталога (`expected_osv_parse` или `expected_statement_parse`).""" return _statement_part(relative) if snapshot_dir == STATEMENT_SNAPSHOT_DIR else _osv_part(relative) def _osv_part(relative: Path) -> str: 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 _statement_part(relative: Path) -> str: first = relative.parts[0] if relative.parts else "" return STATEMENTS_SYNTHETIC_PART if first == SYNTHETIC_DIR else STATEMENTS_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, )