arm-core/api/tests/test_contract_identity.py

"""Правило теста идентичности копии контракта (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"