test(#42): тест идентичности копии контракта по MAJOR, ARM_KA_DOCS (S-53, шаг 0 docs#116)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
Andrey Epifancev 2026-10-18 13:19:37 +04:00
parent f095535250
commit 0005c1d234
7 changed files with 371 additions and 16 deletions

View File

@ -84,8 +84,8 @@ SMB-сессия, S3, NATS, воркер; `GET /healthcheck`; gzip/proxy middlew
- `api/src/normalizer/rules.py` — `load_rules(SCAN_RULES_DIR)`: pydantic-модели справочников, `OcrRules` (выбор редакции
по дате / по кодам); `api/rules/` — `statement_lines.yaml` (редакции `66n`, `fsbu4`), `ocr_lookalikes.yaml`,
`statement_labels.yaml`.
- `api/src/contract/` + `api/contracts/` — pydantic-модели и байт-в-байт копии схем `../docs/contracts`; идентичность
проверяет `api/tests/unit/test_contract.py` (входит в `make check`).
- `api/src/contract/` + `api/contracts/` — pydantic-модели и байт-в-байт копии схем `../docs/contracts`; идентичность по правилу MAJOR (S-53,
`tests/contract_identity.py`; docs — `../docs` или `ARM_KA_DOCS`) проверяет `api/tests/unit/test_contract.py` (входит в `make check`).
- `api/src/scan_emulator/` — эмулятор ИС Сканирование (`SCAN_FIXTURES_DIR`, `SCAN_EMULATOR_DELAY`,
`SCAN_EMULATOR_POLL_SECONDS`, `SCAN_EMULATOR_DEFAULT`; пара по sha256 содержимого); в compose контейнера пока нет —
`make emulator`.

View File

@ -0,0 +1,109 @@
"""Правило сверки копии контракта `api/contracts/` с `docs/contracts/` на окне перехода v1 → v2 (S-53).
Правило — из спеки docs `recognition-contract`, требование «Копия контракта в сервисе идентична источнику»:
схема копии `<имя>.schema.json` с `$id` `…/v<M>` сверяется с `<имя>.v<M>.schema.json` docs, если такой файл есть,
иначе с основным `<имя>.schema.json`; пример `examples/<имя>.json` — с `examples/v<M>/<имя>.json`, если он есть
(`M` — MAJOR схемы копии вида примера), иначе с `examples/<имя>.json`; прочие файлы копии (`.v<N>`-схемы,
`examples/v<N>/`) — с одноимённым файлом docs. Файл побайтно одинаков в arm-core, tabular-parser и scan-gateway
(`api/tests/contract_identity.py`); правка — во всех трёх, сверка — `shasum -a 256`.
"""
import json
import re
from pathlib import Path
# Порядок значим: «result-message-» начинается с «result-», берётся первый подходящий префикс.
KIND_BY_PREFIX: tuple[tuple[str, str], ...] = (
("request-", "recognition-request"),
("result-message-", "recognition-result-message"),
("result-", "recognition-result"),
)
_EXAMPLE_DEPTH = 2 # examples/<файл>.json
_ID_MAJOR = re.compile(r"/v(\d+)$")
_WINDOW_SCHEMA = re.compile(r"\.v(\d+)\.schema\.json$")
_WINDOW_EXAMPLE = re.compile(r"^examples/v(\d+)/[^/]+\.json$")
def schema_major(schema_path: Path) -> int:
"""MAJOR схемы из её `$id` (`…/v<M>`)."""
schema = json.loads(schema_path.read_text(encoding="utf-8"))
schema_id = schema.get("$id") if isinstance(schema, dict) else None
match = _ID_MAJOR.search(schema_id) if isinstance(schema_id, str) else None
if match is None:
raise ValueError(f"{schema_path.name}: в $id нет MAJOR (…/v<M>)")
return int(match.group(1))
def example_kind(file_name: str) -> str:
"""Имя схемы вида примера по префиксу имени файла."""
for prefix, kind in KIND_BY_PREFIX:
if file_name.startswith(prefix):
return kind
raise ValueError(f"{file_name}: вид примера не определён по префиксу (request-, result-message-, result-)")
def docs_pair(copy_rel: str, copy_dir: Path, docs_dir: Path) -> str:
"""Относительный POSIX-путь файла docs, с которым сверяется файл копии `copy_rel`."""
parts = copy_rel.split("/")
if len(parts) == 1 and copy_rel.endswith(".schema.json") and not _WINDOW_SCHEMA.search(copy_rel):
major = schema_major(copy_dir / copy_rel)
candidate = copy_rel.removesuffix(".schema.json") + f".v{major}.schema.json"
return candidate if (docs_dir / candidate).is_file() else copy_rel
if len(parts) == _EXAMPLE_DEPTH and parts[0] == "examples" and copy_rel.endswith(".json"):
kind = example_kind(parts[1])
kind_schema = copy_dir / f"{kind}.schema.json"
if not kind_schema.is_file():
raise ValueError(f"{copy_rel}: нет схемы {kind}.schema.json в копии")
candidate = f"examples/v{schema_major(kind_schema)}/{parts[1]}"
return candidate if (docs_dir / candidate).is_file() else copy_rel
return copy_rel
def held_window_majors(copy_rels: set[str]) -> set[int]:
"""MAJOR, которые копия держит файлами `.v<N>` (хотя бы один файл — держит все)."""
majors: set[int] = set()
for rel in copy_rels:
match = _WINDOW_SCHEMA.search(rel) if "/" not in rel else _WINDOW_EXAMPLE.match(rel)
if match:
majors.add(int(match.group(1)))
return majors
def _json_files(root: Path) -> set[str]:
return {p.relative_to(root).as_posix() for p in root.rglob("*.json")}
def _is_required_in_copy(docs_rel: str, held: set[int]) -> bool:
parts = docs_rel.split("/")
if len(parts) == 1 and docs_rel.endswith(".schema.json"):
match = _WINDOW_SCHEMA.search(docs_rel)
return match is None or int(match.group(1)) in held
if len(parts) == _EXAMPLE_DEPTH and parts[0] == "examples":
return True
match = _WINDOW_EXAMPLE.match(docs_rel)
return match is not None and int(match.group(1)) in held
def identity_problems(copy_dir: Path, docs_dir: Path) -> list[str]:
"""Все расхождения копии с docs; каждое начинается с относительного пути файла. Пустой список — идентична."""
copy_files = _json_files(copy_dir)
docs_files = _json_files(docs_dir)
problems: list[str] = []
for copy_rel in sorted(copy_files):
try:
pair = docs_pair(copy_rel, copy_dir, docs_dir)
except ValueError as exc:
problems.append(f"{copy_rel}: {exc}")
continue
if pair not in docs_files:
problems.append(f"{copy_rel}: нет пары в docs/contracts ({pair})")
elif (copy_dir / copy_rel).read_bytes() != (docs_dir / pair).read_bytes():
problems.append(
f"{copy_rel}: отличается от docs/contracts/{pair} — скопируйте заново (cp), не правьте руками"
)
held = held_window_majors(copy_files)
for docs_rel in sorted(docs_files - copy_files):
if _is_required_in_copy(docs_rel, held):
problems.append(f"{docs_rel}: есть в docs/contracts, нет в api/contracts")
return sorted(problems, key=lambda p: p.split(":", 1)[0])

View File

@ -1,11 +1,13 @@
"""Контракт обмена с сервисами распознавания v1.0 (spec recognition-contract, slice-0-skeleton).
Быстрые тесты без БД: вендоренные схемы `api/contracts/` идентичны `../docs/contracts/`; каждый пример проходит и
JSON Schema, и pydantic-модель (с round-trip); модели и схемы выносят один и тот же вердикт на негативных кейсах.
Быстрые тесты без БД: вендоренные схемы `api/contracts/` идентичны `../docs/contracts/` по правилу MAJOR (S-53);
каждый пример проходит и JSON Schema, и pydantic-модель (с round-trip); модели и схемы выносят один и тот же
вердикт на негативных кейсах.
"""
import copy
import json
import os
import uuid
from collections.abc import Callable
from datetime import UTC, datetime
@ -43,6 +45,7 @@ from src.contract import (
request_subject,
result_msg_id,
)
from tests.contract_identity import identity_problems
API_DIR = Path(__file__).resolve().parents[2]
CONTRACTS_DIR = API_DIR / "contracts"
@ -113,24 +116,30 @@ def _model_ok(kind: str, payload: dict[str, Any]) -> bool:
def _docs_contracts_dir() -> Path:
"""`docs/contracts` соседнего репозитория: `<корень scan-gateway>/../docs` (workspace и worktree `.wt/<срез>/`)."""
return API_DIR.parent.parent / "docs" / "contracts"
"""`docs/contracts`: каталог docs из `ARM_KA_DOCS`, иначе соседний `docs` рядом с репозиторием `scan-gateway`."""
docs = os.environ.get("ARM_KA_DOCS")
docs_dir = Path(docs) if docs else API_DIR.parent.parent / "docs"
return docs_dir / "contracts"
def test_docs_contracts_dir_uses_arm_ka_docs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv("ARM_KA_DOCS", str(tmp_path))
assert _docs_contracts_dir() == tmp_path / "contracts"
def test_docs_contracts_dir_defaults_to_sibling_docs(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("ARM_KA_DOCS", raising=False)
assert _docs_contracts_dir() == API_DIR.parent.parent / "docs" / "contracts"
def test_vendored_contracts_are_identical_to_docs() -> None:
docs_dir = _docs_contracts_dir()
if not docs_dir.is_dir():
pytest.skip("docs недоступен: тест идентичности контракта пропущен (положите репозиторий docs рядом)")
docs_files = {p.relative_to(docs_dir).as_posix() for p in docs_dir.rglob("*.json")}
own_files = {p.relative_to(CONTRACTS_DIR).as_posix() for p in CONTRACTS_DIR.rglob("*.json")}
assert own_files == docs_files, (
f"набор файлов контракта разошёлся: только в api/contracts: {sorted(own_files - docs_files)}, "
f"только в docs/contracts: {sorted(docs_files - own_files)}"
)
for name in sorted(docs_files):
assert (CONTRACTS_DIR / name).read_bytes() == (docs_dir / name).read_bytes(), (
f"файл {name} в api/contracts отличается от docs/contracts: скопируйте заново (cp), не правьте руками"
pytest.skip(
"docs недоступен: тест идентичности контракта пропущен (задайте ARM_KA_DOCS или положите docs рядом)"
)
problems = identity_problems(CONTRACTS_DIR, docs_dir)
assert problems == [], "\n".join(problems)
def test_example_table_covers_all_example_files() -> None:

View File

@ -0,0 +1,190 @@
"""Правило теста идентичности копии контракта (S-53, спека recognition-contract). Файл одинаков в arm-core,
tabular-parser и scan-gateway; правка — сразу во всех трёх."""
import json
from pathlib import Path
import pytest
from tests.contract_identity import docs_pair, example_kind, identity_problems, schema_major
REQ, MSG, RES = "recognition-request", "recognition-result-message", "recognition-result"
def _schema(name: str, major: int) -> str:
return json.dumps({"$id": f"arm-ka/{name}/v{major}"})
def _example(major: int, mark: str) -> str:
return json.dumps({"schema_version": f"{major}.0", "mark": mark})
def _v1_copy() -> dict[str, str]: # копия сервиса сегодня = docs main до #117
return {
f"{REQ}.schema.json": _schema(REQ, 1),
f"{MSG}.schema.json": _schema(MSG, 1),
f"{RES}.schema.json": _schema(RES, 1),
"examples/request-image.json": _example(1, "request"),
"examples/result-message-done.json": _example(1, "message"),
"examples/result-statement-excel.json": _example(1, "result"),
}
def _v2_main() -> dict[str, str]: # копия сервиса после шага 5 = docs после #117 без файлов окна; запрос остаётся v1
return {
f"{REQ}.schema.json": _schema(REQ, 1),
f"{MSG}.schema.json": _schema(MSG, 2),
f"{RES}.schema.json": _schema(RES, 2),
"examples/request-image.json": _example(1, "request"),
"examples/result-message-done.json": _example(2, "message"),
"examples/result-statement-excel.json": _example(2, "result"),
}
def _docs_window() -> dict[str, str]: # docs после #117 = копия ядра после шага 3
v1 = _v1_copy()
return {
**_v2_main(),
f"{MSG}.v1.schema.json": v1[f"{MSG}.schema.json"],
f"{RES}.v1.schema.json": v1[f"{RES}.schema.json"],
"examples/v1/result-message-done.json": v1["examples/result-message-done.json"],
"examples/v1/result-statement-excel.json": v1["examples/result-statement-excel.json"],
}
def _dirs(tmp_path: Path, copy: dict[str, str], docs: dict[str, str]) -> tuple[Path, Path]:
"""Пишет файлы в tmp_path/copy и tmp_path/docs (подкаталоги создаёт), возвращает (copy_dir, docs_dir)."""
result = []
for name, files in (("copy", copy), ("docs", docs)):
root = tmp_path / name
root.mkdir()
for rel, text in files.items():
target = root / rel
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(text, encoding="utf-8")
result.append(root)
return result[0], result[1]
def _without(files: dict[str, str], rel: str) -> dict[str, str]:
return {k: v for k, v in files.items() if k != rel}
def test_step0_before_docs_merge_passes(tmp_path: Path) -> None:
c, d = _dirs(tmp_path, _v1_copy(), _v1_copy())
assert identity_problems(c, d) == []
assert docs_pair(f"{RES}.schema.json", c, d) == f"{RES}.schema.json"
def test_stale_copy_fails_naming_file(tmp_path: Path) -> None:
docs = _v1_copy()
docs[f"{RES}.schema.json"] = json.dumps({"$id": "arm-ka/recognition-result/v1", "new": 1})
c, d = _dirs(tmp_path, _v1_copy(), docs)
problems = identity_problems(c, d)
assert len(problems) == 1
assert problems[0].startswith(f"{RES}.schema.json")
assert "отличается" in problems[0]
def test_service_on_v1_after_v2_merge_passes(tmp_path: Path) -> None:
c, d = _dirs(tmp_path, _v1_copy(), _docs_window())
assert identity_problems(c, d) == []
assert docs_pair(f"{RES}.schema.json", c, d) == f"{RES}.v1.schema.json"
assert docs_pair(f"{MSG}.schema.json", c, d) == f"{MSG}.v1.schema.json"
def test_request_v1_pairs_with_main_after_v2_merge(tmp_path: Path) -> None:
c, d = _dirs(tmp_path, _v1_copy(), _docs_window())
assert docs_pair(f"{REQ}.schema.json", c, d) == f"{REQ}.schema.json"
assert docs_pair("examples/request-image.json", c, d) == "examples/request-image.json"
def test_previous_major_examples_pair_with_v1_dir(tmp_path: Path) -> None:
c, d = _dirs(tmp_path, _v1_copy(), _docs_window())
for name in ("result-statement-excel.json", "result-message-done.json"):
assert docs_pair(f"examples/{name}", c, d) == f"examples/v1/{name}"
def test_service_on_v2_passes_and_ignores_v1_files(tmp_path: Path) -> None:
c, d = _dirs(tmp_path, _v2_main(), _docs_window())
assert identity_problems(c, d) == []
assert docs_pair(f"{RES}.schema.json", c, d) == f"{RES}.schema.json"
assert docs_pair("examples/result-statement-excel.json", c, d) == "examples/result-statement-excel.json"
def test_core_window_copy_passes(tmp_path: Path) -> None:
c, d = _dirs(tmp_path, _docs_window(), _docs_window())
assert identity_problems(c, d) == []
assert docs_pair(f"{RES}.v1.schema.json", c, d) == f"{RES}.v1.schema.json"
assert docs_pair("examples/v1/result-statement-excel.json", c, d) == "examples/v1/result-statement-excel.json"
def test_core_window_copy_requires_every_v1_docs_file(tmp_path: Path) -> None:
copy = _without(_docs_window(), "examples/v1/result-message-done.json")
c, d = _dirs(tmp_path, copy, _docs_window())
assert "examples/v1/result-message-done.json: есть в docs/contracts, нет в api/contracts" in identity_problems(c, d)
def test_window_copy_without_docs_pair_fails(tmp_path: Path) -> None:
c, d = _dirs(tmp_path, _docs_window(), _v2_main())
problems = identity_problems(c, d)
for rel in (f"{RES}.v1.schema.json", "examples/v1/result-statement-excel.json"):
found = [p for p in problems if p.startswith(rel)]
assert found, rel
assert "нет пары" in found[0]
def test_top_level_docs_file_missing_in_copy_fails(tmp_path: Path) -> None:
copy = _without(_v1_copy(), "examples/request-image.json")
c, d = _dirs(tmp_path, copy, _v1_copy())
assert "examples/request-image.json: есть в docs/contracts, нет в api/contracts" in identity_problems(c, d)
def test_extra_copy_file_fails(tmp_path: Path) -> None:
copy = {**_v1_copy(), "examples/result-new.json": _example(1, "result")}
c, d = _dirs(tmp_path, copy, _v1_copy())
problems = identity_problems(c, d)
assert len(problems) == 1
assert problems[0].startswith("examples/result-new.json")
assert "нет пары" in problems[0]
def test_new_top_level_docs_example_is_reported(tmp_path: Path) -> None:
docs = {**_docs_window(), "examples/result-statement-ocr.json": _example(2, "result")}
c, d = _dirs(tmp_path, _v1_copy(), docs)
problems = identity_problems(c, d)
assert len(problems) == 1
assert problems[0].startswith("examples/result-statement-ocr.json")
def test_unknown_example_prefix_fails(tmp_path: Path) -> None:
files = {**_v1_copy(), "examples/foo.json": _example(1, "x")}
c, d = _dirs(tmp_path, files, files)
problems = identity_problems(c, d)
assert len(problems) == 1
assert problems[0].startswith("examples/foo.json")
assert "вид примера" in problems[0]
with pytest.raises(ValueError, match="вид примера"):
example_kind("foo.json")
def test_schema_without_major_fails(tmp_path: Path) -> None:
copy = {**_v1_copy(), f"{RES}.schema.json": json.dumps({"$id": "arm-ka/recognition-result"})}
c, d = _dirs(tmp_path, copy, _v1_copy())
problems = identity_problems(c, d)
assert any(p.startswith(f"{RES}.schema.json") and "$id" in p for p in problems)
with pytest.raises(ValueError):
schema_major(c / f"{RES}.schema.json")
def test_example_kind_by_longest_prefix(tmp_path: Path) -> None:
copy = {
**_v1_copy(),
f"{RES}.schema.json": _schema(RES, 2),
"examples/result-statement-excel.json": _example(2, "result"),
}
docs = {**copy, "examples/v1/result-message-done.json": _example(1, "message")}
c, d = _dirs(tmp_path, copy, docs)
assert example_kind("result-message-done.json") == MSG
assert docs_pair("examples/result-message-done.json", c, d) == "examples/v1/result-message-done.json"
assert docs_pair("examples/result-statement-excel.json", c, d) == "examples/result-statement-excel.json"

View File

@ -0,0 +1,22 @@
## Why
Окно перехода контракта v1 → v2 (docs change `unified-statement-normalization`, шаг 0 выкладки): после слияния docs PR #117
основные схемы в `docs/contracts/` станут v2.0, а прежние v1 лягут рядом файлами `*.v1.schema.json` и `examples/v1/`.
Текущий тест идентичности копии сверяет наборы файлов целиком и покраснеет сразу после слияния. Правило S-53 спеки docs
`recognition-contract` сверяет копию с файлом docs своего MAJOR.
## What Changes
- Тест идентичности копии `api/contracts/` использует правило MAJOR (`api/tests/contract_identity.py`, побайтно одинаковый
файл в arm-core, tabular-parser и scan-gateway) вместо сравнения наборов файлов целиком.
- Требование «Копия контракта идентична источнику» уточнено под окно перехода.
## Non-goals
- Копии `api/contracts/` и pydantic-модели не меняются; переход сервиса на контракт 2.0 — шаг 5 выкладки, отдельный change.
## Impact
- `api/tests/` (тест идентичности, правило, сценарии), `CLAUDE.md`.
- Issue: RSHB/arm-scan-gateway#42; план —
`docs/openspec/changes/unified-statement-normalization/plans/2026-10-18-plan-0-contract-identity.md`.

View File

@ -0,0 +1,22 @@
## MODIFIED Requirements
### Requirement: Копия контракта идентична источнику
Каждый файл `api/contracts/**/*.json` SHALL быть байт в байт равен своей паре в `../docs/contracts/` (или в `contracts`
каталога из `ARM_KA_DOCS`) по правилу спеки docs `recognition-contract`, требование «Копия контракта в сервисе
идентична источнику»: схема с `$id` `…/v<M>` — с `<имя>.v<M>.schema.json`, если такой файл есть в docs, иначе с
основным; пример — с `examples/v<M>/<имя>.json`, если он есть, иначе с `examples/<имя>.json`; файлы docs другого
`MAJOR` не сверяются. pydantic-модели SHALL валидировать все примеры копии `api/contracts/examples/*.json` и порождать
JSON, проходящий схему копии. Проверка входит в `make check`.
#### Scenario: `make check`
- **WHEN** выполняются тесты контракта при доступном `../docs` (или `ARM_KA_DOCS`)
- **THEN** идентичность подтверждена, примеры валидны; при недоступном docs тест идентичности пропущен с предупреждением
#### Scenario: Окно перехода на контракт 2.0
- **WHEN** в docs основной `recognition-result.schema.json` уже v2.0, рядом `recognition-result.v1.schema.json` и
`examples/v1/`, а копия сервиса — v1.2
- **THEN** копия сверяется с файлами `.v1` и `examples/v1/` (запрос — с основным файлом), тест проходит
#### Scenario: Копия устарела
- **WHEN** в docs изменён файл, с которым сверяется копия, а копия — нет
- **THEN** тест идентичности падает, сообщение называет файл

View File

@ -0,0 +1,3 @@
## 1. Тест идентичности по правилу MAJOR
- [x] 1.1 Выполнить задачи плана 0 (`docs`, `openspec/changes/unified-statement-normalization/plans/2026-10-18-plan-0-contract-identity.md`) для этого репозитория: правило, сценарии, реальный тест, проверка `make check` на docs main и на ветке `feat/116-unified-normalization`.