Aislo/resources/tester/test_322_0_golden.py

"""PLAN 322-0 전수 표 시험 — 구조물 42종 × 견본마다 일곱 갈래 결과를 기록(`golden_322_structures.json`)과 맞춤.

갈래 = 심는 측점(TS `structureStationMarks` · `pipeStationMarks` · 파이썬 `resolve_extra_stations`) · 카드 이름
(파이썬 측점 이름 · TS 미리 뜬 횡단 이름 · 표시 이름) · 형상 얹기 열쇠(파이썬 attach 차례 · TS `Section_Store` 초안 얹기)
· 3D 몸체 · 알약 띠 · 고르기 목록(B05 · B06 `pickGroups`) · 옹벽 물빼기.
견본 = 종류마다 기본값 한 벌(main) · 빈칸(blank) · 배관은 흄관(hume) 하나 더 — 점형 · 부지형 = 101 m · 구간형 = 93 ~ 113 m
(기준 101) · 관 지점 = 101 m(기슭막이 96 ~ 106 m) · 격자 = 0 ~ 200 m 20 m 마다.

기록 다시 쓰기 = `GOLDEN_322_RECORD=1` · ★ = 계획(PLAN_구조물표 4장)대로 바뀔 칸 — 기록에 「지금 → 다음」을 적고
`STAGE` 가 그 단계를 넘으면 다음 값과 맞춤. TS 쪽 = `helper_322_0_golden.cjs`(Node).
"""

from __future__ import annotations

import copy
import json
import os
import subprocess
import tempfile
from pathlib import Path

import pytest

from B05_Profile.B05_Profile_Engine_Sections import relabel_pipe_stations, resolve_extra_stations
from B05_Profile.B05_Profile_Structures_Repository import structures_file_path
from B05_Profile.B05_Profile_Structures_Schema import load_definition
from B06_Section.B06_Section_Engine_Culvert import attach_culvert_sets, pipe_points_file
from B06_Section.B06_Section_Engine_Structures_Wall import attach_wall_structures
from B06_Section.B06_Section_Engine_Tilt import attach_underdrains
from common_util.common_util_drainage_pipes import parse_pipe_points
from common_util.common_util_structure_pavement import attach_pavement_structures
from M02_MasterTemplete.StructOptions.M02_SO_Options import with_visible

HERE = Path(__file__).parent
ROOT = HERE.parent.parent
GOLDEN = HERE / "golden_322_structures.json"
FIXTURE = HERE / "fixtures" / "b06_73_4_body_guard_sections.json"
#: 지금 단계 — ★ 칸은 `단계` 가 이 값 이하면 「다음」 값과 맞춤
STAGE = 1
GRID = [float(at) for at in range(0, 201, 20)]
SHAPE_KEYS = ("culvert", "ford", "box", "ford_pavement", "revetment", "pavement", "underdrain")

pytestmark = pytest.mark.skipif(
    not (ROOT / "config" / "node_modules").is_dir(), reason="프론트엔드 의존성 미설치"
)


def _plain_section() -> dict:
    sections = json.loads(FIXTURE.read_text(encoding="utf-8"))["cross_sections"]
    plain = copy.deepcopy(next(s for s in sections if s["chainage_m"] == 180))
    for key in (*SHAPE_KEYS, "structure"):
        plain.pop(key, None)
    return plain


def _samples(definition: dict) -> list[dict]:
    out: list[dict] = []
    for item in definition["types"]:
        defaults = {
            option["key"]: option["default"]
            for option in item.get("options", [])
            if option.get("default") is not None
        }
        cases = {"main": defaults, "blank": {}}
        if item["type_id"] == "pipe":
            cases["hume"] = {**defaults, "pipe_kind": "흄관", "pipe_diameter_mm": "800"}
        for case, options in cases.items():
            pipe = structure = None
            if item.get("managed_by"):
                pipe = {"chainage_m": 101.0, "source": "user", "facility": item["type_id"]}
                if item["type_id"] == "revetment":
                    pipe.update(start_m=96.0, end_m=106.0)
                if options:
                    pipe["options"] = options
            else:
                structure = {
                    "structure_id": "s1",
                    "type_id": item["type_id"],
                    "placement": item["placement"],
                    "chainage_m": 101.0,
                    "options": options,
                    "memo": "",
                    "placement_source": "manual",
                    "status": "draft",
                    "revision": 0,
                    "geometry": None,
                }
                if item["placement"] == "interval":
                    structure.update(start_m=93.0, end_m=113.0)
            out.append(
                {"type_id": item["type_id"], "case": case, "pipe": pipe, "structure": structure}
            )
    return out


def _shape_of(section: dict) -> dict:
    """TS 짝 `shapeOf` 와 같은 꼴."""
    out: dict = {}
    for key in SHAPE_KEYS:
        spec = section.get(key)
        if not spec:
            continue
        if key == "culvert":
            out[key] = {
                "hidden_pipe": bool(spec.get("hidden_pipe")),
                "body": bool(spec.get("body")),
            }
        elif key == "revetment":
            out[key] = {"type_id": spec.get("type_id"), "form": spec.get("form")}
        elif key == "pavement":
            out[key] = sorted(p["type_id"] for p in spec)
        else:
            out[key] = True
    return out


def _python_row(sample: dict, root: Path, plain: dict) -> dict:
    """서버 길 — 정본 파일을 쓰고 측점 만들기 · 횡단 얹기 차례(`_read_section_detail`)를 그대로."""
    if sample["structure"]:
        path = Path(structures_file_path(str(root)))
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(
            json.dumps({"revision": 1, "structures": [sample["structure"]]}), encoding="utf-8"
        )
    pipes = parse_pipe_points([sample["pipe"]]) if sample["pipe"] else []
    if sample["pipe"]:
        path = pipe_points_file(root)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps({"points": [sample["pipe"]]}), encoding="utf-8")
    extras = resolve_extra_stations(root, pipes)
    sections = [{**copy.deepcopy(plain), "chainage_m": at, "kind": "regular"} for at in GRID]
    for at, label in extras:
        hit = next((s for s in sections if abs(s["chainage_m"] - at) < 0.005), None)
        if hit is None:
            hit = {**copy.deepcopy(plain), "chainage_m": at, "kind": "irregular"}
            sections.append(hit)
        hit["structure"] = label
    sections.sort(key=lambda s: s["chainage_m"])
    attach_culvert_sets(root, sections)
    relabel_pipe_stations(root, sections)
    attach_wall_structures(root, sections)
    attach_pavement_structures(root, sections)
    attach_underdrains(root, sections)
    return {
        "station_py": [f"{label}@{at:.2f}" for at, label in extras],
        "cards_py": {
            f"{s['chainage_m']:.2f}": {"title": s.get("structure"), "shape": _shape_of(s)}
            for s in sections
            if s.get("structure") or _shape_of(s)
        },
    }


@pytest.fixture(scope="module")
def result() -> dict:
    with tempfile.TemporaryDirectory() as folder:
        return _run(Path(folder))


def _run(work: Path) -> dict:
    definition = load_definition()
    samples = _samples(definition)
    plain = _plain_section()
    payload = {
        "types": with_visible().model_dump()["types"],
        "definition": definition,
        "samples": samples,
        "grid": GRID,
        "plain_section": plain,
    }
    (work / "input.json").write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
    out = subprocess.run(
        ["node", str(HERE / "helper_322_0_golden.cjs"), str(work / "input.json")],
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=300,
        check=True,
    )
    ts = json.loads(out.stdout)
    rows: dict = {}
    for index, sample in enumerate(samples):
        key = f"{sample['type_id']}/{sample['case']}"
        rows[key] = {**ts["samples"][key], **_python_row(sample, work / f"p{index}", plain)}
    return {"samples": rows, "pick": ts["pick"], "version": ts["version"], "notice": ts["notice"]}


def test_type_cache_follows_definition_version(result: dict) -> None:
    """322-0 ③ — 화면 종류 캐시: 같은 판이면 그대로 · M02 저장으로 판이 바뀌면 탭이 다시 보일 때 버리고 다시 받음."""
    assert result["version"] == {
        "same": False,
        "sameCalls": 0,
        "reloads": 1,
        "staleCalls": 1,
        "removed": True,
    }


def test_stale_definition_notice_once(result: dict) -> None:
    """322-2 판 알림 — structures.json 판 ≠ 지금 판이면 한 줄 · 탭 수명 동안 한 번 · 같은 판 · 판 없음은 없음."""
    now = load_definition()["판"] + 1  # 시험 서버가 판을 하나 올린 뒤
    assert result["notice"] == [
        f"구조물 정의가 판 {now} 으로 바뀜 — 저장하면 반영",
        None,
        None,
        None,
    ]


def _cells(table: dict) -> dict[tuple[str, str, str], object]:
    return {
        (part, key, column): value
        for part in ("samples", "pick")
        for key, row in table[part].items()
        for column, value in row.items()
    }


def test_golden_matches(result: dict) -> None:
    if os.environ.get("GOLDEN_322_RECORD") == "1":
        old = json.loads(GOLDEN.read_text(encoding="utf-8")) if GOLDEN.exists() else {}
        GOLDEN.write_text(
            json.dumps(
                {
                    "★ 메모": old.get("★ 메모", []),
                    "★": old.get("★", []),
                    "samples": result["samples"],
                    "pick": result["pick"],
                },
                ensure_ascii=False,
                indent=1,
            )
            + "\n",
            encoding="utf-8",
        )
    golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
    star = {(s["part"], s["key"], s["column"]): s for s in golden["★"]}
    want = _cells(golden)
    got = _cells(result)
    assert set(got) == set(want), "견본 · 칸 목록이 기록과 다름"
    wrong = []
    for cell, value in want.items():
        mark = star.get(cell)
        expect = mark["다음"] if mark and STAGE >= mark["단계"] else value
        if got[cell] != expect:
            wrong.append(f"{cell}: 기록 {expect!r} · 지금 {got[cell]!r}")
    assert not wrong, "\n".join(wrong[:40])


def test_golden_covers_every_type(result: dict) -> None:
    types = {item["type_id"] for item in load_definition()["types"]}
    assert len(types) == 42
    assert {key.split("/")[0] for key in result["samples"]} == types
    assert set(result["pick"]) == types


def test_star_cells_exist() -> None:
    golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
    cells = _cells(golden)
    for mark in golden["★"]:
        assert (mark["part"], mark["key"], mark["column"]) in cells, mark