Aislo/resources/tester/test_445_save_patch.py

"""PLAN 445 — [저장] 횡단 patch 바탕 · 되돌림 · 옮긴 측점 · 집수정 범위 · 버린 수.

요구 268-3 · 분석 `docs/raw/조사/2026-10-18_442_요구268_저장누락.md`
(H2 · H3 · M2 · H5 · M1 · M5 · M7).

못박는 것(조정창으로 고침 → 횡단 보기 [저장] 본문 → 서버 병합 → DB 설계)
  - H3 — 새 탭에서 유출 벽만 고쳐도 유입 벽 · 집수 쪽 단 수가 남음(바탕 = 캐시 설계 통째).
  - M2 — 새 탭 ↺ · 추가 벽 0 · 빈 전체 폭이 저장분을 지움 · 저장 전 새로고침에도 되돌림이 남음.
  - M1 — 집수정 내공 2.1 → 서버 범위 2.0 으로 잘림(422 없음).
  - H2 — 정본 행이 없는 자리(옮긴 관)는 지반 등 입력까지 실음 · 서버 기본 설계가 그 입력으로 섬.
  - M5 — 행 없는 측점 patch 를 버린 수 · 응답 칸.
  - H5 · M7 — 노선 저장 뒤 본문 · kindreset 옮김(소스 배선).
짝: `helper_445_save_patch.cjs`.
"""

from __future__ import annotations

import asyncio
import json
import subprocess
import sys
from pathlib import Path
from uuid import uuid4

import pytest
from pydantic import ValidationError

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

from B06_Section import B06_Section_Router_Confirm as confirm  # noqa: E402
from B06_Section.B06_Section_Repository_Bulk import (  # noqa: E402
    merge_cross_section_design_batches,
)
from B06_Section.B06_Section_Schema import (  # noqa: E402
    CrossSectionPatch,
    SectionConfirmRequest,
    SectionConfirmResponse,
)

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

AT = 100.0
SAVED = {
    "ground_type": "soil",
    "section_mode": "left_cut",
    "structure_note": "메모",
    "display_half_width_m": 8.0,
    "revet_adjust": {
        "inlet": {"x": 1.0, "d": None, "h": None, "m": None},
        "outlet": {"x": 2.0, "d": None, "h": None, "m": None},
    },
    "extra_wall_counts": {"outlet": 1, "basin": 2},
    "basin_adjust": {"innerWidthM": 1.5, "innerHeightM": 1.2, "lateralM": 0, "slopeM": 0},
}
MOVED = {
    "chainage_m": 130.0,
    "kind": "irregular",
    "design": {
        "ground_type": "soil",
        "ditch_type": "l_type",
        "two_stage_slope": True,
        "rock_boundary_offset_m": -1.5,
        "structure_note": "옮긴 관 메모",
    },
}


@pytest.fixture(scope="module")
def r(tmp_path_factory) -> dict:
    work = tmp_path_factory.mktemp("b445")
    job = {
        "at": AT,
        "detail": {
            "longitudinal": {"length_m": 500, "stations": []},
            "cross_sections": [{"chainage_m": AT, "kind": "irregular", "design": SAVED}],
        },
        "moved": MOVED,
    }
    path = work / "job.json"
    path.write_text(json.dumps(job, ensure_ascii=False), encoding="utf-8")
    proc = subprocess.run(  # noqa: S603 — 고정 실행 파일
        ["node", str(HERE / "helper_445_save_patch.cjs"), str(path)],
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=120,
    )
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


class _Cursor:
    def __init__(self, connection):
        self.connection = connection

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False

    async def execute(self, sql, params=None):
        self.connection.log.append((" ".join(sql.split()), list(params or [])))

    async def fetchall(self):
        return self.connection.rows


class _Connection:
    def __init__(self, rows):
        self.rows = rows
        self.log: list = []

    def cursor(self, *args):
        return _Cursor(self)


def _server_save(patches: list[dict], rows=None) -> tuple[dict, list[float]]:
    """화면 patch → 스키마 → 서버 병합(실제 `merge_cross_section_design_batches`) → 저장된 설계."""
    request = SectionConfirmRequest(cross_patches=patches)
    rows = rows or [(1, AT, json.dumps({"design": SAVED}, ensure_ascii=False))]
    connection = _Connection(rows)
    missed = asyncio.run(confirm._apply_section_edits(connection, 7, request, []))
    updates = [params for sql, params in connection.log if sql.startswith("UPDATE")]
    if not updates:
        return SAVED, missed
    return json.loads(updates[-1][1])["design"], missed


def test_H3_한_벽만_고쳐도_다른_벽_단_수_남음(r):
    design, _ = _server_save(r["h3"]["patches"])
    assert design["revet_adjust"]["outlet"]["x"] == 3.0
    assert design["revet_adjust"]["inlet"]["x"] == 1.0  # 예전: 유입 0(세션 몫만 보내 통째 교체)
    assert design["extra_wall_counts"] == {"outlet": 1, "basin": 2}
    assert design["structure_note"] == "메모" and design["display_half_width_m"] == 8.0
    # 세션 몫만으로는 유입이 빠짐 — 바탕이 메운 것
    assert set(r["h3"]["overlay"][0]["revet_adjust"]) == {"outlet"}
    old, _ = _server_save(r["h3"]["overlay"])  # 옛 길(세션 몫만) = 유입이 사라짐
    assert "inlet" not in old["revet_adjust"]


def test_M2_되돌림이_저장분을_지움(r):
    design, _ = _server_save(r["m2"]["patches"])
    assert set(design["revet_adjust"]) == {"outlet"}  # ↺ 유입 = 지움
    assert design["extra_wall_counts"] == {"outlet": 0, "basin": 2}  # 새 탭 − = 0 · 집수 쪽 그대로
    assert design["display_half_width_m"] == 0  # 빈 전체 폭 = 개별 폭 없음(읽는 쪽 0 이하 = 없음)
    assert r["m2"]["countShown"] == 0


def test_M2_저장_전_새로고침에도_되돌림이_남음(r):
    reload = r["m2reload"]
    assert reload["cachedInlet"] is None  # 세션 null 이 정본 벽을 다시 뺌
    assert reload["cachedCounts"] == {"outlet": 0, "basin": 2}
    design, _ = _server_save(reload["patches"])
    assert set(design["revet_adjust"]) == {"outlet"}
    assert design["extra_wall_counts"] == {"outlet": 0, "basin": 2}
    assert design["display_half_width_m"] == 0


def test_M1_집수정_내공_서버_범위로_잘림(r):
    basin = next(p for p in r["m2"]["patches"] if "basin_adjust" in p)["basin_adjust"]
    assert basin["innerWidthM"] == 2.0 and basin["innerHeightM"] == 1.0
    design, _ = _server_save(r["m2"]["patches"])
    assert design["basin_adjust"]["innerWidthM"] == 2.0
    with pytest.raises(ValidationError):  # 서버 범위 그대로(화면이 맞춤)
        CrossSectionPatch(chainage_m=AT, basin_adjust={**SAVED["basin_adjust"], "innerWidthM": 2.1})


def test_H2_옮긴_측점은_지반_등_입력까지(r):
    moved = next(p for p in r["h2"] if abs(p["chainage_m"] - MOVED["chainage_m"]) < 0.01)
    for key in ("ground_type", "ditch_type", "two_stage_slope", "rock_boundary_offset_m"):
        assert moved[key] == MOVED["design"][key], key
    assert moved["structure_note"] == "옮긴 관 메모"
    kept = next(p for p in r["h2"] if abs(p["chainage_m"] - AT) < 0.01)
    assert (
        "ground_type" not in kept
    )  # 정본 행이 있는 측점은 사용자 칸만(결과 칸 · 입력은 정본 그대로)
    # 서버 — 행 없는 자리의 기본 설계를 그 입력으로 세움(`compute_default_designs(carried=…)`)
    request = SectionConfirmRequest(cross_patches=r["h2"])
    carried = confirm._carried_inputs(request, [130.004])
    assert carried == {
        130.004: {
            "ground_type": "soil",
            "ditch_type": "l_type",
            "two_stage_slope": True,
            "rock_boundary_offset_m": -1.5,
        }
    }


def test_M5_행_없는_측점_patch_버린_수():
    missed: list[float] = []
    connection = _Connection([(1, AT, json.dumps({"design": SAVED}))])
    asyncio.run(
        merge_cross_section_design_batches(
            connection,
            route_id=7,
            batches=[
                ([(AT, {"structure_note": "x"}), (250.0, {"structure_note": "y"})], False, None)
            ],
            missed=missed,
        )
    )
    assert missed == [250.0]
    assert confirm._default_failed([10.0, 20.0], [(10.004, {})]) == [20.0]
    fields = SectionConfirmResponse.model_fields
    assert {"dropped_patch_chainages", "default_failed_chainages", "stations_synced"} <= set(fields)
    api = (ROOT / "B06_Section" / "B06_Section_Api_Fetch.ts").read_text(encoding="utf-8")
    assert "result.dropped_patch_chainages" in api and "stations_synced === false" in api


def test_M5_새_행은_버린_수에_안_셈():
    missed: list[float] = []
    connection = _Connection([])
    asyncio.run(
        merge_cross_section_design_batches(
            connection,
            route_id=7,
            batches=[([(300.0, {"ground_type": "soil"})], True, uuid4())],
            missed=missed,
        )
    )
    assert missed == []


def test_H5_노선_저장_뒤_본문_M7_kindreset_옮김():
    store = (ROOT / "B06_Section" / "B06_Section_Section_Store.ts").read_text(encoding="utf-8")
    body = store[store.index("export async function saveCachedCrossPatches") :]
    for needle in (
        "sessionOverlay(projectId, routeId)",
        "readStandardCrossSession(projectId)",
        "balloon_offsets: balloonOffsetsPayload()",
    ):
        assert needle in body, needle
    overlay = store[store.index("function sessionOverlay") :]
    for needle in ("crossDesignChoices(", "readRockBoundarySession(", "readCutSlopeSession("):
        assert needle in overlay, needle
    persist = (ROOT / "B06_Section" / "B06_Section_UI_Page_Persist.ts").read_text(encoding="utf-8")
    assert (
        "cachedCrossPatches(projectId, routeId, edits.crossPatches)" in persist
    )  # 472-6 횡단 조각
    keys = (ROOT / "B06_Section" / "B06_Section_Station_Keys.ts").read_text(encoding="utf-8")
    carry = keys[keys.index("export function carryChainageStates") :]
    assert 'writeState("kindreset", rekey("revetx", resets), projectId)' in carry
    idle = keys[keys.index("function moveIdleStates") :]
    assert 'writeState("kindreset", move(resets), projectId)' in idle


def test_H2_B06_관_폼_초안이_관_id_를_실음():
    """B06 관 폼이 초안에 관 id 를 빠뜨리면 측점 열쇠가 `pipe:p…` → `pipe:pipe-356.00` 으로 바뀌어
    옮긴 측점이 앞 설계(지반 · 메모 · 조절)를 잃음 — 화면 실측(관 351 → 356)."""
    panel = (ROOT / "B06_Section" / "B06_Section_UI_Page_Structures_Panel.ts").read_text(
        encoding="utf-8"
    )
    load = panel[panel.index("pipeFacilities = (pendingPipes ?? pipeResponse.pipe_points)") :]
    assert "id: pipe.id" in load[:200]
    persist = panel[panel.index("function persistPipes") :]
    assert "...(pipe.id ? { id: pipe.id } : {})" in persist[:400]