Aislo/resources/tester/test_322_3_plan_mirror.py

"""PLAN 322-3 측점 계획 한 벌 — 파이썬 `plan_stations` = TS `planStations`(거울) · 서버 공급자가 같은 목록.

- 측점 = 표 횡단도(anchor 기준 · span3 시작 · 기준 · 끝 · none 없음) · 카드 이름 = 표 카드 이름.
- 같은 m(0.005) 는 하나 — 이름 = 관이 먼저 · 아니면 앞선 것 · 덮는 구조물 id 모두.
- 빈칸 경고 = 형상 부품 필수 칸 중 값이 빈 것(관경 · 벽 높이 · 맹암거 높이 · 물넘이 높이).
- 서버 `resolve_extra_stations` = 계획의 (측점, 이름) — 겹친 자리는 한 번(옛 「마지막 이름」 규칙 걷음).
TS 쪽 = `helper_322_3_plan.cjs`(Node).
"""

from __future__ import annotations

import json
import subprocess
import tempfile
from pathlib import Path

import pytest

from B05_Profile.B05_Profile_Engine_Sections import resolve_extra_stations
from B05_Profile.B05_Profile_Structures_Schema import StructureInstance, load_definition
from common_util.common_util_drainage_pipes import parse_pipe_points
from common_util.common_util_structure_station import plan_stations
from M02_MasterTemplete.StructOptions.M02_SO_Options import with_visible

HERE = Path(__file__).parent
ROOT = HERE.parent.parent

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


def _s(type_id: str, sid: str, **at) -> dict:
    placement = "interval" if "start_m" in at else "point"
    return {
        "structure_id": sid,
        "type_id": type_id,
        "placement": placement,
        "options": at.pop("options", {}),
        **at,
    }


CASES = {
    "mixed": {
        "pipes": [
            {"chainage_m": 95.0, "source": "user"},
            {"chainage_m": 160.0, "source": "user", "facility": "box_culvert"},
            {"chainage_m": 230.0, "source": "user", "facility": "ford_pavement"},
            {
                "chainage_m": 400.0,
                "source": "user",
                "facility": "revetment",
                "start_m": 395.0,
                "end_m": 410.0,
                "options": {"height_m": 2},
            },
        ],
        "structures": [
            _s("cross_drain_exposed", "s1", chainage_m=120.0),
            _s(
                "retaining_wall",
                "s2",
                start_m=500.0,
                chainage_m=510.0,
                end_m=530.0,
                options={"height_m": 3},
            ),
            _s("berm", "s3", start_m=530.0, chainage_m=545.0, end_m=560.0),
            _s("erosion_check", "s4", chainage_m=600.0),
            _s("pavement_concrete", "s5", start_m=650.0, chainage_m=650.0, end_m=700.0),
            _s("cross_underdrain", "s6", chainage_m=95.003),
        ],
    },
    "pipe_kinds": {
        "pipes": [
            {"chainage_m": 10.0, "options": {"pipe_kind": "흄관", "pipe_diameter_mm": "800"}},
            {"chainage_m": 20.0, "options": {"pipe_diameter_mm": ""}},
            {"chainage_m": 30.0, "facility": "ford_bridge"},
        ],
        "structures": [],
    },
    "empty": {"pipes": [], "structures": []},
    # 368-2 추가 측점 — 이름 "" · 주인 plus-{m} · 같은 m 에 관 · 구조물이 있으면 그쪽이 이김
    "plus": {
        "pipes": [{"chainage_m": 95.0, "source": "user"}],
        "structures": [
            _s(
                "retaining_wall",
                "s2",
                start_m=500.0,
                chainage_m=510.0,
                end_m=530.0,
                options={"height_m": 3},
            )
        ],
        "plus": [47.0, 12.0, 95.0, 510.0, 47.0],
    },
}


@pytest.fixture(scope="module")
def ts() -> dict:
    definition = load_definition()
    # 관 id = 파이썬이 읽을 때 매기는 고정 id(347) — TS 에도 같은 id 로 넘김
    cases = {
        name: {
            **case,
            "pipes": [
                {**pipe, "id": parsed.id}
                for pipe, parsed in zip(case["pipes"], parse_pipe_points(case["pipes"]))
            ],
        }
        for name, case in CASES.items()
    }
    payload = {
        "definition": definition,
        "types": with_visible().model_dump()["types"],
        "cases": cases,
    }
    with tempfile.TemporaryDirectory() as folder:
        path = Path(folder) / "input.json"
        path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
        out = subprocess.run(
            ["node", str(HERE / "helper_322_3_plan.cjs"), str(path)],
            capture_output=True,
            text=True,
            encoding="utf-8",
            timeout=120,
            check=True,
        )
    return json.loads(out.stdout)


def _py(case: dict) -> list[dict]:
    pipes = parse_pipe_points(case["pipes"])
    structures = [StructureInstance.model_validate(s) for s in case["structures"]]
    return plan_stations(pipes, structures, case.get("plus", ()))


def test_mirror_python_equals_ts(ts: dict) -> None:
    for name, case in CASES.items():
        assert _py(case) == ts[name], name


def test_plan_rules(ts: dict) -> None:
    mixed = ts["mixed"]
    assert [round(s["chainage_m"], 3) for s in mixed] == [
        95.0,
        160.0,
        230.0,
        395.0,
        400.0,
        410.0,
        500.0,
        510.0,
        530.0,
        545.0,
        560.0,
    ]
    first = mixed[0]  # 관 95 + 맹암거 95.003 = 한 측점 · 이름 = 관 · 빈칸 = 관경 + 맹암거 높이
    assert first["owners"] == ["p9500", "s6"]
    assert first["title"] == "파형강관 D1000"
    assert first["blank"] == ["pipe_diameter_mm", "culvert_height_m"]
    joint = next(s for s in mixed if s["chainage_m"] == 530.0)  # 옹벽 끝 = 소단 시작
    assert joint["owners"] == ["s2", "s3"] and joint["title"] == "옹벽" and joint["role"] == "end"
    assert all(s["title"] != "노출형 횡단수로" for s in mixed)  # ★1 횡단도 none
    kinds = ts["pipe_kinds"]
    assert [s["title"] for s in kinds] == ["흄관 D800", "파형강관 D1000", "세월교"]
    assert kinds[1]["blank"] == ["pipe_diameter_mm"] and kinds[0]["blank"] == []
    assert ts["empty"] == []


def test_plus_stations_rules(ts: dict) -> None:
    plus = ts["plus"]
    assert [(s["chainage_m"], s["title"], s["owners"]) for s in plus][:3] == [
        (12.0, "", ["plus-12"]),
        (47.0, "", ["plus-47"]),
        (95.0, "파형강관 D1000", ["p9500"]),
    ]
    assert next(s for s in plus if s["chainage_m"] == 510.0)["owners"] == ["s2"]


def test_server_supply_reads_plus_file(tmp_path: Path) -> None:
    """서버 공급자 = 추가 측점 정본(plus_stations.json)까지 — 이름 ""."""
    route = tmp_path / "B05_Profile" / "route"
    route.mkdir(parents=True)
    rows = [{"chainage_m": 47.0, "x": 1, "y": 2}, {"chainage_m": 12.0, "x": 0, "y": 0}]
    (route / "plus_stations.json").write_text(json.dumps({"stations": rows}), encoding="utf-8")
    extras = resolve_extra_stations(tmp_path, parse_pipe_points([{"chainage_m": 95.0}]))
    assert list(extras) == [(12.0, ""), (47.0, ""), (95.0, "파형강관 D1000")]


def test_server_supply_is_plan(tmp_path: Path) -> None:
    case = CASES["mixed"]
    route = tmp_path / "B05_Profile" / "route"
    route.mkdir(parents=True)
    (route / "structures.json").write_text(
        json.dumps({"revision": 1, "structures": case["structures"]}, ensure_ascii=False),
        encoding="utf-8",
    )
    extras = resolve_extra_stations(tmp_path, parse_pipe_points(case["pipes"]))
    plan = _py(case)
    assert list(extras) == [(s["chainage_m"], s["title"]) for s in plan]
    assert len({round(c, 3) for c, _ in extras}) == len(extras)  # 같은 자리 한 번


def test_stale_marks_compare_position_and_name(tmp_path: Path, monkeypatch) -> None:
    """서버 저장 낡음 = 자리 + 이름(종류 · 관 공식 · 정의 판 이름) — 같은 자리라도 이름이 다르면 맞춤."""
    from B06_Section import B06_Section_Router_Stations as stations

    folder = tmp_path / "B06_Section" / "longitudinal"
    folder.mkdir(parents=True)
    rows = [
        {"chainage_m": 10.0, "kind": "irregular", "structure": "파형강관 D800"},
        {"chainage_m": 20.0, "kind": "regular"},
        {"chainage_m": 30.0, "kind": "irregular", "structure": "옹벽"},
    ]
    (folder / "l.json").write_text(json.dumps({"stations": rows}), encoding="utf-8")
    planted = [(10.0, "파형강관 D1000"), (30.0, "옹벽")]
    monkeypatch.setattr(stations, "_planted", lambda *_: planted)
    assert stations._stale_marks(tmp_path, "") == 2  # 10 m 옛 이름 · 새 이름
    planted[0] = (10.0, "파형강관 D800")
    assert stations._stale_marks(tmp_path, "") == 0