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


def test_창구_측점_목록에_서버_정의_붙임(monkeypatch):
    from fastapi import FastAPI
    from fastapi.testclient import TestClient

    from B06_Section import B06_Section_Router_Calc as calc_router

    seen = []
    monkeypatch.setattr(
        calc_router, "run_calc_served", lambda run, inputs: seen.append(inputs) or {"results": {}}
    )
    app = FastAPI()
    app.include_router(calc_router.router)
    url = "/api/projects/11111111-2222-3333-4444-555555555555/calc"
    body = {"run": ["stations.list"], "inputs": {"stations": {"line": [], "definition": {"x": 1}}}}
    assert TestClient(app).post(url, json=body).status_code == 200
    definition = seen[0]["stations"]["definition"]
    assert definition != {"x": 1} and definition["types"]  # 브라우저가 보낸 정의 안 씀


def test_판_stations_줄_열림_기본_both():
    from common_util import common_util_calc_sites as calc_sites

    group = next(g for g in calc_sites.GROUPS if g["key"] == "stations")
    assert group["switchable"] and group["fixed"] is None
    assert calc_sites.read_sites()["sites"]["stations"] == "both"
    page = (ROOT / "A00_Common/design/design_pipeline_page.ts").read_text(encoding="utf-8")
    assert "remote: stationsOnServer" in page


# ── 화면 — 계산 자리 stations 「서버」 측점 = 화면 셈 측점(검증_sub3 · Playwright · 저장 안 함) ──

VERIFY = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"  # 검증_sub3(OWNERS.md)

JS_STATIONS = r"""async () => {
  const urls = performance.getEntriesByType('resource').map((e) => e.name);
  const page = await import(urls.find((n) => /design_pipeline_page\.ts/.test(n)));
  const pid = localStorage.getItem('frd_current_project_id');
  const pipe = page.pagePipeline(pid);
  await page.reconcilePagePipeline(pid);
  await pipe.settled();
  const calls = (window.__aisloCalcServer?.calls ?? []).filter((c) => c.id === 'stations.list');
  return { stations: JSON.parse(JSON.stringify(pipe.output('stations'))),
           calls: calls.map((c) => ({ round: c.roundMs, node: c.nodeMs, bytes: c.bytes })),
           runs: (window.__aisloPipeline?.runs ?? []).slice(-1).map((r) => r.outcome) };
}"""


@pytest.fixture(scope="module")
def screens():
    from test_511_b1_calc_node_stages import BASE, JS_DRAFT, _UP, _started

    if not BASE or _UP is None or _UP != _started(BASE.replace("5178", "8005")):
        pytest.skip("이 폴더 서버 없음")
    sync = pytest.importorskip("playwright.sync_api")
    cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
    shots = {}
    with sync.sync_playwright() as play:
        chrome = play.chromium.launch(channel="msedge", headless=True)
        try:
            for mode in ("both", "server"):
                context = chrome.new_context(viewport={"width": 1600, "height": 1000})
                login = {k: cred[k] for k in ("email", "password")}
                r = context.request.post(
                    f"{BASE}/api/auth/login/request", data=login, timeout=60000
                )
                assert r.ok, f"로그인 실패 {r.status}"  # 비밀번호는 안 찍음
                if mode == "server":  # 판은 안 고침 — 이 창이 받는 판만 stations = server

                    def board(route):
                        body = route.fetch().json()
                        body["sites"]["stations"] = "server"
                        route.fulfill(json=body)

                    context.route("**/api/dashboard/calc-sites", board)
                page = context.new_page()
                page.add_init_script(
                    f"localStorage.setItem('frd_current_project_id', '{VERIFY}');"
                    " performance.setResourceTimingBufferSize(20000);"
                )
                page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
                page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
                page.wait_for_timeout(9000)
                shots[mode] = {"saved": page.evaluate(JS_STATIONS)}
                page.evaluate(JS_DRAFT)  # 교각점 하나 옮김(세션 초안 · 창과 함께 버림)
                page.wait_for_timeout(3000)
                shots[mode]["draft"] = page.evaluate(JS_STATIONS)
                context.close()
        finally:
            chrome.close()
    (ROOT / "tmp" / "511b2_screens.json").write_text(json.dumps(shots), encoding="utf-8")
    return shots


import json  # noqa: E402


@pytest.mark.parametrize("case", ["saved", "draft"])
def test_화면_서버_측점_화면_셈과_같음(screens, case):
    both, server = screens["both"][case], screens["server"][case]
    assert both["runs"] == server["runs"] == ["done"]
    assert len(both["stations"]) > 10
    hit = _first_diff(both["stations"], server["stations"])
    assert hit is None, hit
    assert both["calls"] == [] and server["calls"]  # server 판만 서버를 부름