Aislo/resources/tester/test_511_b2_stations_node.py

"""PLAN 511-B B-2 (요구 320-4) — 서버 측점 생성 = 화면과 같은 TS(Node `stations.sections`).

못박는 것
  - 고정 자료(`fixtures/golden_472`) 노선 · 관 · 구조물 · 추가 측점 — `build_sections`
    Node 길 = 파이썬 `generate_sections` 길 0 차(칸 · 차례 · 값 · 측점 열쇠 `key` 만 더함) —
    곡선표 있음 · 없음 · 스냅 10 m · 측점 간격 10 m · 끝 측점 뺌.
  - DTM 아닌 지표면 = 파이썬 길(Node 안 부름 · 열쇠 없음).
  - 열쇠 = Node `stations.list`(화면 묶음 stations 단계 `buildStations`) 열쇠.
  - 창구(`/calc`)는 `stations.sections`(서버 파일 경로 입력)를 안 받음.
  - 등록부 R5 · R6 · R7 = node:stations.list · G4 = node:stations.sections(472-4 until 없음).
  - 공유 DB · 실제 storage 안 씀(고정 자료 사본).

    ./venv/Scripts/python.exe -m pytest resources/tester/test_511_b2_stations_node.py -q
"""

from __future__ import annotations

import sys
from dataclasses import replace
from pathlib import Path

import numpy as np
import pytest

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
for path in (ROOT, HERE):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

import helper_472_1_paths as paths  # noqa: E402

from B05_Profile import B05_Profile_Engine_Sections as sections  # noqa: E402
from B05_Profile import B05_Profile_Engine_Sections_Node as node  # noqa: E402
from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions  # noqa: E402
from common_util import common_util_calc_runner as runner  # noqa: E402
from common_util.common_util_surface_sampler import (  # noqa: E402
    CallableSurfaceSampler,
    build_surface_sampler,
)

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


@pytest.fixture(scope="module")
def project(tmp_path_factory):
    root = paths.stage_project(tmp_path_factory.mktemp("b2") / "project")
    summary = paths.read("summary.json")["longitudinal"]["data"]
    options = SectionGenerationOptions(
        **{k: v for k, v in summary["options"].items() if k != "extra_stations"}
    )
    polyline = sections._load_route_polyline(root, paths.ROUTE)
    pipes = sections._load_pipe_points(root, polyline)
    facilities = sections.station_facilities(root, pipes, polyline)
    extras = sections.resolve_extra_stations(
        root, facilities[0], structures=facilities[1], plus=facilities[2]
    )
    return {
        "root": root,
        "options": options,
        "polyline": polyline,
        "pipes": pipes,
        "facilities": facilities,
        "extras": extras,
        "sampler": build_surface_sampler(root / "B04_PreProcess/models", *paths.SURFACE),
        "curves": sections._load_design_curves(root),
    }


def _both(project, sampler=None, **change):
    kw = {
        "options": project["options"],
        "grade_options": None,
        "source_snapshot": {"filter": "classification"},
        "crs": "EPSG:5186",
        "design_curves": project["curves"],
        "station_snap": sections.load_station_snap(project["root"]),
        "station_snap_step": sections.load_station_snap_step(project["root"]),
        **change,
    }
    args = (project["polyline"], sampler or project["sampler"], project["pipes"], project["extras"])
    python, _ = sections.build_sections(*args, **kw)
    noded, _ = sections.build_sections(*args, facilities=project["facilities"], **kw)
    return python, noded


def _without_key(value):
    if isinstance(value, dict):
        return {k: _without_key(v) for k, v in value.items() if k != "key"}
    if isinstance(value, list):
        return [_without_key(v) for v in value]
    return value


def _first_diff(a, b, path=""):
    """첫 다른 자리(값 · 칸 차례) — 큰 글 비교를 pytest 에 안 넘김. 수 = 같은 값(0 ↔ 0.0)."""
    if isinstance(a, dict) and isinstance(b, dict):
        if list(a) != list(b):
            return f"{path}: 칸 {list(a)} != {list(b)}"
        return next((d for k in a if (d := _first_diff(a[k], b[k], f"{path}.{k}"))), None)
    if isinstance(a, list) and isinstance(b, list):
        if len(a) != len(b):
            return f"{path}: 길이 {len(a)} != {len(b)}"
        return next(
            (d for i, (x, y) in enumerate(zip(a, b)) if (d := _first_diff(x, y, f"{path}[{i}]"))),
            None,
        )
    return None if a == b else f"{path}: {a!r} != {b!r}"


def _same(python, noded):
    hit = _first_diff(python, _without_key(noded))  # 칸 차례 · 자릿수까지
    assert hit is None, hit
    keys = [s.get("key") for s in noded["longitudinal"]["stations"]]
    assert all(keys) and keys == [s["key"] for s in noded["cross_sections"]]
    return keys


@pytest.mark.parametrize(
    "case",
    ["as_saved", "no_curves", "snap10", "interval10", "no_endpoint"],
)
def test_Node_길_파이썬_길_0차(project, case):
    change = {
        "as_saved": {},
        "no_curves": {"design_curves": []},
        "snap10": {"station_snap": True, "station_snap_step": 10.0},
        "interval10": {"options": replace(project["options"], station_interval_m=10.0)},
        "no_endpoint": {"options": replace(project["options"], include_endpoint=False)},
    }[case]
    python, noded = _both(project, **change)
    keys = _same(python, noded)
    assert len(keys) == len(set(keys)) > 10
    if case == "as_saved":
        assert len(keys) == 57 and keys[0] == "end:BP" and keys[-1] == "end:EP"
        kinds = {s["kind"] for s in python["longitudinal"]["stations"]}
        assert {"bp", "ep", "regular", "irregular"} <= kinds
        assert any(k.startswith("pipe:") for k in keys)


def test_열쇠_Node_측점_목록과_같음(project):
    _, noded = _both(project)
    opts = project["options"]
    stations = node.stations_input(
        [[p[0], p[1]] for p in project["polyline"]],
        *project["facilities"],
        opts,
        project["curves"],
        sections.load_station_snap(project["root"]),
        sections.load_station_snap_step(project["root"]),
    )
    listed = node.node_stations(stations)
    assert [s["key"] for s in listed] == [s["key"] for s in noded["longitudinal"]["stations"]]
    assert [s["chainageM"] for s in listed] == [
        s["chainage_m"] for s in noded["longitudinal"]["stations"]
    ]


def test_DTM_아닌_지표면_파이썬_길(project, monkeypatch):
    called = []
    monkeypatch.setattr(runner, "run_calc_served", lambda *a: called.append(a))
    flat = CallableSurfaceSampler(lambda xy: np.full(len(xy), 100.0))
    python, noded = _both(project, sampler=flat)
    assert called == [] and python == noded
    assert "key" not in noded["longitudinal"]["stations"][0]


def test_Node_못_돌림_파이썬_길(project, monkeypatch):
    monkeypatch.setattr(runner, "run_calc_served", lambda *a: None)
    python, noded = _both(project)
    assert python == noded


def test_창구는_서버_측점_생성_안_받음():
    from fastapi import FastAPI
    from fastapi.testclient import TestClient

    from B06_Section import B06_Section_Router_Calc as calc_router

    app = FastAPI()
    app.include_router(calc_router.router)
    http = TestClient(app)
    url = "/api/projects/11111111-2222-3333-4444-555555555555/calc"
    body = {"run": ["stations.sections"], "inputs": {"stations": {"grid": "C:/x"}}}
    bad = http.post(url, json=body)
    assert bad.status_code == 400 and "창구 밖" in bad.json()["message"]
    assert runner.SERVER_ONLY_IDS <= runner.NODE_IDS


def test_등록부_측점_줄_Node():
    src = (ROOT / "A00_Common/design/logic/design_calc_registry.ts").read_text(encoding="utf-8")
    for no, run in (
        ("R5", "stations.list"),
        ("R6", "stations.list"),
        ("R7", "stations.list"),
        ("G4", "stations.sections"),
    ):
        row = src.split(f'no: "{no}"')[1].split("\n  },")[0]
        assert f'runner: "node:{run}"' in row and "until" not in row, no


def test_이월_열쇠_Node_출력_없으면_파이썬(project):
    from B05_Profile.B05_Profile_Route_Save_Parts import keyed_stations, station_keys

    _, noded = _both(project)
    stations = noded["longitudinal"]["stations"]
    keyed = keyed_stations(stations)
    pipes, structures, _ = project["facilities"]
    assert keyed == station_keys(stations, pipes, structures, 20.0)  # 고정 자료 = 같은 열쇠
    assert keyed_stations([{k: v for k, v in s.items() if k != "key"} for s in stations]) is None
    assert keyed_stations([*stations, {"chainage_m": 1.0}]) is None  # 섞이면(구조물 측점 덧) 파이썬
    route_save = (ROOT / "B05_Profile/B05_Profile_Route_Save.py").read_text(encoding="utf-8")
    assert route_save.count("keyed_stations(") == 2