"""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(ctx.projectId, routeId, edits.crossPatches)" in persist 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]