diff --git a/resources/tester/test_346_3_route_save.py b/resources/tester/test_346_3_route_save.py index 50aa79fef..3a0618cb5 100644 --- a/resources/tester/test_346_3_route_save.py +++ b/resources/tester/test_346_3_route_save.py @@ -1,108 +1,561 @@ -"""346-3 화면 — [노선 반영] 없앰 · [저장]이 노선부터 · 추천 [적용] = as_initial 저장 · 저장 뒤 초안 넘김. +# -*- coding: utf-8 -*- +"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 오목 보충 · 이월 열쇠 · +초기값 규칙 · +배수유역 낡음 · 실패 때 옛 노선 되살림. -노선 이음(`B05_Profile_UI_RouteEdit_Apply`)은 Node 헬퍼 `helper_346_3_route_save.cjs` 가 서버 · 화면 창구만 -가짜로 두고 돌림 · 저장 차례(노선 먼저 · 안 바뀌면 안 보냄 · 실패 때 초안 남음)는 `test_design_store_save.py`. +순수 부품(`B05_Profile_Route_Save_Parts`)은 값으로 · 체인(`run_route_save_chain`)과 API +(`replan_route`)는 DB · 엔진을 가짜로 바꿔 차례 · 되살림 · 초기값만 잰다. """ from __future__ import annotations +import asyncio import json -import subprocess +import sys from pathlib import Path +from types import SimpleNamespace +from uuid import uuid4 +import numpy as np import pytest -HERE = Path(__file__).resolve().parent -ROOT = HERE.parents[1] +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from fastapi.responses import JSONResponse # noqa: E402 + +import B03_FileInput.B03_FileInput_Service_Chain as chain # noqa: E402 +import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed # noqa: E402 +import B05_Profile.B05_Profile_Route_Save as save # noqa: E402 +import B05_Profile.B05_Profile_Router as b05_router # noqa: E402 +import B05_Profile.B05_Profile_Router_Replan as replan # noqa: E402 +import B05_Profile.B05_Profile_Router_RouteInitial as route_initial # noqa: E402 +import B05_Profile.B05_Profile_Structures_Router as structures_router # noqa: E402 +import B06_Section.B06_Section_Repository as b06_repo # noqa: E402 +import B06_Section.B06_Section_Router as b06_router # noqa: E402 +import B06_Section.B06_Section_Router_Confirm as b06_confirm # noqa: E402 +import common_util.common_util_drainage_pipes as drainage_pipes # noqa: E402 +import common_util.common_util_initial_snapshot as snapshot # noqa: E402 +import common_util.common_util_workflow_state as workflow # noqa: E402 +from B05_Profile.B05_Profile_Route_Save_Parts import ( # noqa: E402 + add_crossing_pipes, + carry_cross_inputs, + concave_chainages, + fill_concave_pipes, + follow_pipes, + station_keys, + vertices_of, +) +from common_util.common_util_drainage_pipes import ( # noqa: E402 + PIPE_SOURCE_SPACING, + PIPE_SOURCE_STREAM, + PipePoint, + analysis_route_signature, + route_signature, +) + +OLD = [(0.0, 0.0), (100.0, 0.0), (200.0, 0.0), (300.0, 0.0), (400.0, 0.0)] +# 새 노선 = 100 ~ 300 m 를 북으로 휘게(앞 · 뒤는 옛 노선 위) +NEW = [(0.0, 0.0), (100.0, 0.0), (200.0, 60.0), (300.0, 0.0), (400.0, 0.0)] -def _read(rel: str) -> str: - return (ROOT / rel).read_text(encoding="utf-8") +# ── 관 따라감 · 교차점 · 오목 보충 ───────────────────────────────────────────── -@pytest.fixture(scope="module") -def out() -> dict: - run = subprocess.run( - ["node", str(HERE / "helper_346_3_route_save.cjs")], - capture_output=True, - text=True, - encoding="utf-8", - check=True, +def test_관은_지우지_않고_좌표를_따라감(): + old = [ + PipePoint(chainage_m=50.0, x=50.0, y=0.0, options={"pipe_kind": "흄관"}), + PipePoint(chainage_m=350.0, x=350.0, y=0.0, start_m=345.0, end_m=355.0), + ] + pairs = follow_pipes(old, vertices_of(NEW)) + assert [pipe_id for pipe_id, _ in pairs] == [0, 1] + assert pairs[0][1].options == {"pipe_kind": "흄관"} # 사용자 관 속성 그대로 + shift = pairs[1][1].chainage_m - 350.0 # 새 노선이 길어진 만큼 뒤 관이 밀림 + assert shift > 30 and pairs[1][1].start_m == pytest.approx(345.0 + shift) + assert old[1].chainage_m == 350.0 # 옛 목록은 안 고침 + + +def test_새_교차점에만_기본_관(): + pairs = follow_pipes([PipePoint(chainage_m=50.0, x=50.0, y=0.0)], vertices_of(NEW)) + crossings = [ + SimpleNamespace(chainage_m=20.0, x=20.0, y=0.0), # 옛 노선 위 — 사용자가 지운 관 안 되살림 + SimpleNamespace(chainage_m=150.4, x=150.0, y=30.0), # 새 자리 · 관 없음 → 더함 + SimpleNamespace(chainage_m=52.0, x=52.0, y=0.0), # 옛 자리 + ] + added = add_crossing_pipes(pairs, crossings, OLD, 456.0) + assert added == 1 + new = [pipe for pipe_id, pipe in pairs if pipe_id < 0] + assert new[0].chainage_m == 150.0 and new[0].source == PIPE_SOURCE_STREAM + # 같은 자리 다시 = 이미 관이 있음 + assert add_crossing_pipes(pairs, crossings, OLD, 456.0) == 0 + + +def test_오목_지점_보충은_새_자리의_오목만(): + vertices = vertices_of(NEW) + total = vertices[-1].chainage_m + chainage = np.arange(0.0, total, 2.0) + ground = 100.0 + np.abs(chainage - 220.0) * 0.05 # 220 m 에 골(새 자리) + ground[chainage < 60] = ( + 100.0 + np.abs(chainage[chainage < 60] - 30.0) * 0.05 + ) # 30 m 골(옛 자리) + concave = concave_chainages(chainage, ground) + assert any(abs(at - 220.0) < 3 for at, _ in concave) + pairs: list = [] + added = fill_concave_pipes(pairs, concave, vertices, OLD) + places = [pipe.chainage_m for _, pipe in pairs] + assert added >= 1 and all(pipe.source == PIPE_SOURCE_SPACING for _, pipe in pairs) + assert any(abs(at - 220.0) <= 3 for at in places) + assert not any(at < 60 for at in places) # 옛 노선 위 오목엔 안 둠 + + +def test_간격이_좁으면_보충_없음(): + vertices = vertices_of(NEW) + pairs = [(0, PipePoint(chainage_m=at)) for at in (0.0, 90.0, 180.0, 270.0, 360.0, 450.0)] + concave = [(220.0, 99.0)] + assert fill_concave_pipes(pairs, concave, vertices, OLD) == 0 + + +# ── 측점 열쇠 · 횡단 이월 ────────────────────────────────────────────────────── + + +def _structure(sid, type_id, anchor, start=None, end=None): + return SimpleNamespace( + structure_id=sid, + type_id=type_id, + chainage_m=anchor, + start_m=start, + end_m=end, + options={}, ) - return json.loads(run.stdout) -def test_노선_안_바뀜_안_보냄(out): - assert out["noneAtAll"] is None - assert out["unchanged"] is None # 편집기가 정함 — 가게에 초안이 남아 있어도 - assert out["storeOnly"]["draft"]["rev"] == 3 # 편집기 없으면 가게 routeDraft - - -def test_초안은_계약_RouteDraft_꼴(out): - draft = out["body"]["draft"] - assert set(draft) == { - "rev", - "nodes", - "curveOn", - "curveRadius", - "curveLock", - "curveArc", - "stationSnap", - "terrain", +def test_측점_열쇠_규칙(): + stations = [{"chainage_m": s} for s in (0.0, 20.0, 40.0, 50.0, 60.0, 77.0, 80.0, 95.5)] + pipes = [(3, PipePoint(chainage_m=50.0))] + structures = [_structure("w1", "retaining_wall", 77.0, 60.0, 80.0)] + keys = station_keys(stations, pipes, structures, 20.0) + assert keys == { + "end:BP": 0.0, + "grid:1": 20.0, + "grid:2": 40.0, + "pipe:3": 50.0, + "structure:w1:start": 60.0, + "structure:w1": 77.0, + "structure:w1:end": 80.0, + "end:EP": 95.5, } - n = len(draft["nodes"]) - assert all( - len(draft[name]) == n for name in ("curveOn", "curveRadius", "curveLock", "curveArc") + + +def test_이월은_열쇠로_누가거리가_밀려도(): + old_keys = { + "end:BP": 0.0, + "grid:1": 20.0, + "pipe:0": 50.0, + "structure:w1": 77.0, + "grid:5": 100.0, + } + new_keys = {"end:BP": 0.0, "grid:1": 20.0, "pipe:0": 57.0, "structure:w1": 84.0} + designs = { + 0.0: {"ground_type": "soil"}, + 20.0: {"ground_type": "rock", "berm": {"on": True}, "cut_area_m2": 9.0}, + 50.0: {"ditch_type": "u", "inlet_structure": "basin", "basin_adjust": {"d": 1}}, + 77.0: {"rock_boundary_offset_m": 0.8, "revet_adjust": {"h": 2}}, + 100.0: {"ground_type": "soil"}, + } + inputs, adjusts, dropped = carry_cross_inputs( + old_keys, + new_keys, + designs, + ("ground_type", "ditch_type", "berm", "rock_boundary_offset_m"), + ("inlet_structure", "basin_adjust", "revet_adjust"), ) - assert draft["curveOn"] == [True, False, True] and draft["curveRadius"] == [None, 25, None] - assert draft["stationSnap"] is True and draft["terrain"] == "sheet" - assert out["body"]["initial"] == {"mode": "follow"} and out["body"]["balanced_profile"] is True - assert "as_initial" not in out["body"] # 손으로 고친 노선 = 초기값 그대로 + assert inputs == { + 0.0: {"ground_type": "soil"}, + 20.0: {"ground_type": "rock", "berm": {"on": True}}, # 결과 칸(면적)은 안 넘김 + 57.0: {"ditch_type": "u"}, + 84.0: {"rock_boundary_offset_m": 0.8}, + } + # 구조물 조정은 관 · 구조물 열쇠만 따라감 + assert adjusts == { + 57.0: {"inlet_structure": "basin", "basin_adjust": {"d": 1}}, + 84.0: {"revet_adjust": {"h": 2}}, + } + assert dropped == 1 # grid:5 — 새 노선에 없음(노선 밖 고침) -def test_저장_뒤_가게_초안_비움_결과_한_줄(out): - assert out["savedRouteId"] == 9 - assert out["storeAfterSave"] is None - assert out["keptAfterSave"] == 1 - assert out["summary"] == ( - "노선 저장 — 관 지킴 3 · 새 교차 관 1 · 오목 보충 관 2 · 구조물 옮김 1 · 멀어짐 0 · " - "노선 밖 고침 3 건 · 배수유역 낡음" - ) # 서버 cross_dropped 1 + 화면 넘김 밖 2 = 한 줄 +# ── 노선 초안 요청 ──────────────────────────────────────────────────────────── -def test_초안_넘김_차례(out): - assert out["carryCalls"] == ["invalidate:9", "load:9", "keys:9:9", "carry:7->9", "crossSave:9"] - assert out["outside"] == 2 +def test_초안_모양을_옛_칸으로(): + body = save.RouteSaveRequest( + draft={ + "rev": 3, + "nodes": [[0, 0], [10, 5], [20, 0]], + "curveOn": [True, False, True], + "curveRadius": [None, 30.0, None], + "curveLock": [None, "radius", None], + "curveArc": [None, None, None], + "stationSnap": True, + "terrain": "sheet", + } + ) + request = body.resolved() + assert [(v.x, v.y, v.curve, v.radius_m) for v in request.vertices] == [ + (0, 0, True, None), + (10, 5, False, 30.0), + (20, 0, True, None), + ] + assert request.station_snap is True and request.terrain == "sheet" + with pytest.raises(ValueError): + save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]}).resolved() -def test_추천_적용은_as_initial_저장_뒤_다시_세움(out): - assert out["recommendOk"] is True - body = out["recommendBody"] - assert body["as_initial"] is True and body["initial"] == {"mode": "earthwork"} - assert body["draft"]["nodes"] == [[1, 1], [9, 9]] - assert out["recommendCalls"] == ["overlay:on", "replan", "overlay:off"] # 잠금 막 하나 - assert out["recommendDraftsLeft"] == {"route": None, "pipes": None} # 새 초기값에서 다시 - assert out["rebuilt"] == 1 +# ── 체인 차례 · 되살림 · 배수유역 낡음 ───────────────────────────────────────── -def test_추천_저장_실패는_한_줄_다시_안_세움(out): - assert out["recommendFail"] is False - errors = [c for c in out["recommendFailCalls"] if c.startswith("toast:error:")] - assert len(errors) == 1 and "체인 실패" in errors[0] - assert out["rebuiltAfterFail"] == 1 +class _Cursor: + def __init__(self, log): + self.log = log + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def execute(self, sql, params=None): + self.log.append(" ".join(sql.split())) -def test_노선_반영_단추_없앰_배선(): - chrome = _read("B05_Profile/B05_Profile_UI_RouteEdit_Chrome.ts") - assert 'data-act="apply"' not in chrome and "노선 반영" not in chrome - apply = _read("B05_Profile/B05_Profile_UI_RouteEdit_Apply.ts") - assert "addEventListener" not in apply # 단추 듣기 없음 — [저장]이 부름 - save = _read("A00_Common/design/design_save.ts") - assert save.index('"route",') < save.index('"pipes",') - assert "await pendingRouteSave(ctx.projectId)" in save - assert "carryAfterRouteSave(ctx.projectId, fromRouteId, routeId)" in save - initial = _read("B05_Profile/B05_Profile_UI_RouteEdit_Initial.ts") - at = initial.index("onApply: (picked) => {") - assert "saveRecommendedRoute(" in initial[at : at + 900] - replan = _read("B05_Profile/B05_Profile_Api_Replan.ts") - assert "body: JSON.stringify(body)" in replan and "as_initial?: boolean;" in replan +class _Connection: + def __init__(self, log): + self.log = log + + def cursor(self, *args): + return _Cursor(self.log) + + async def begin(self): + return None + + async def commit(self): + return None + + async def rollback(self): + return None + + +class _Pool: + def __init__(self, log): + self.log = log + + def acquire(self): + connection = _Connection(self.log) + + class _Acquire: + async def __aenter__(self): + return connection + + async def __aexit__(self, *exc): + return False + + return _Acquire() + + +def _async(value, order=None, name=None): + async def fake(*args, **kwargs): + if order is not None: + order.append(name) + return value + + return fake + + +@pytest.fixture +def chain_world(monkeypatch, tmp_path): + order: list[str] = [] + sql: list[str] = [] + root = tmp_path + route_file = root / "B05_Profile" / "route" / "route_main.geojson" + route_file.parent.mkdir(parents=True) + route_file.write_text( + json.dumps({"geometry": {"coordinates": [[*p, 0] for p in OLD]}}), "utf-8" + ) + long_dir = root / "B06_Section" / "longitudinal" + long_dir.mkdir(parents=True) + (long_dir / "longitudinal.json").write_text(json.dumps({"stations": [], "old": True}), "utf-8") + drain = root / "drain" + pipes_file = drain / "edits" / "pipe_points.json" + pipes_file.parent.mkdir(parents=True) + pipes_file.write_text( + json.dumps( + {"route_signature": "old", "points": [{"chainage_m": 50.0, "x": 50.0, "y": 0.0}]} + ), + "utf-8", + ) + (drain / "00_watershed_response.json").write_text( + json.dumps({"route_signature": "old"}), "utf-8" + ) + state = {"stages": [{"stage_no": 2, "params": {"points": {}, "options": {}}}]} + + def solve(project_id, request, *, generate_sections=True): + order.append("solve") + assert generate_sections is False + route_file.write_text( + json.dumps({"geometry": {"coordinates": [[*p, 0] for p in NEW]}}), "utf-8" + ) + return _async( + SimpleNamespace( + route_id=8, + total_length_m=456.0, + route_data_path="B05_Profile/route/route_main.geojson", + ) + )() + + async def regenerate(*args, **kwargs): + order.append("sections") + (long_dir / "longitudinal.json").write_text( + json.dumps({"stations": [], "new": True}), "utf-8" + ) + return SimpleNamespace() + + async def watershed_must_not_run(*args, **kwargs): + raise AssertionError("배수유역은 체인에서 뗌") + + async def prepare(project_id): + order.append("pipes") + return {"vertices": vertices_of(NEW), "streams": []} + + monkeypatch.setattr(save, "get_db_pool", lambda: _Pool(sql)) + monkeypatch.setattr(workflow, "get_workflow_state", _async(state)) + monkeypatch.setattr(workflow, "start_stage", _async(None, order, "restore_params")) + monkeypatch.setattr( + b06_repo, + "get_longitudinal_section", + _async( + {"data": {}, "longitudinal_file_path": "B06_Section/longitudinal/longitudinal.json"} + ), + ) + monkeypatch.setattr(b06_repo, "get_cross_section_designs", _async([])) + monkeypatch.setattr(b06_repo, "get_latest_section_options", _async({})) + monkeypatch.setattr( + structures_router, + "follow_structures_to_route", + lambda *a: order.append("structures") or {"moved": 0, "far": 0}, + ) + monkeypatch.setattr( + chain, + "redesign_solve_request", + lambda *a: b05_router.RouteSolveRequest( + filter_key="csf", + method="dtm", + smooth=False, + algorithm="as_planned", + bp={"x": 0, "y": 0}, + ep={"x": 400, "y": 0}, + ), + ) + monkeypatch.setattr(b05_router, "solve_route_record", solve) + monkeypatch.setattr(b05_router, "confirm_latest_route", _async({})) + monkeypatch.setattr(watershed, "get_primary_region", watershed_must_not_run) + monkeypatch.setattr(watershed, "_prepare", prepare) + monkeypatch.setattr(drainage_pipes, "clear_pipe_points", watershed_must_not_run) + monkeypatch.setattr(drainage_pipes, "drainage_dir", lambda stored: drain) + monkeypatch.setattr(save, "_ground_along", lambda *a: (np.zeros(0), np.zeros(0))) + monkeypatch.setattr(b06_router, "regenerate_route_sections", regenerate) + monkeypatch.setattr(save, "_carry_profile", _async(None, order, "profile")) + monkeypatch.setattr(save, "_carry_cross", _async(2, order, "cross")) + monkeypatch.setattr( + b06_confirm, "confirm_sections", _async(SimpleNamespace(), order, "confirm") + ) + monkeypatch.setattr(snapshot, "mark_designing", lambda root: None) + monkeypatch.setattr(snapshot, "clear_designing", lambda root: None) + return SimpleNamespace( + order=order, + sql=sql, + root=root, + route_file=route_file, + pipes_file=pipes_file, + long_dir=long_dir, + drain=drain, + ) + + +def _run(world, **kwargs): + old_route = {"id": 7, "route_data_path": "B05_Profile/route/route_main.geojson"} + return asyncio.run( + save.run_route_save_chain( + uuid4(), + world.root, + "p", + 1, + {}, + old_route, + [{"x": x, "y": y} for x, y in NEW], + {"old": 1}, + **kwargs, + ) + ) + + +def test_차례와_배수유역_뗌(chain_world): + result = _run(chain_world) + assert "error" not in result, result + assert chain_world.order == [ + "structures", + "solve", + "pipes", + "sections", + "profile", + "cross", + "confirm", + ] + assert result["cross_dropped"] == 2 and result["drainage_stale"] is True + assert result["pipes"] == {"kept": 1, "crossing": 0, "concave": 0} + # 관 정본 = 새 노선 지문 · 배수유역 분석 지문(옛) 과 달라 B04 가 「낡음」 + saved = json.loads(chain_world.pipes_file.read_text("utf-8")) + assert saved["route_signature"] == route_signature(vertices_of(NEW)) + assert analysis_route_signature("p") != saved["route_signature"] + assert len(saved["points"]) == 1 # 사용자 관 그대로(지우지 않음) + + +def test_실패하면_옛_노선을_되살림(chain_world, monkeypatch): + monkeypatch.setattr( + b06_confirm, "confirm_sections", _async(JSONResponse(status_code=500, content={})) + ) + before_pipes = chain_world.pipes_file.read_text("utf-8") + result = _run(chain_world) + assert result["error"].startswith("종횡단 확정 실패") + assert json.loads(chain_world.route_file.read_text("utf-8"))["geometry"]["coordinates"] == [ + [*p, 0] for p in OLD + ] + assert chain_world.pipes_file.read_text("utf-8") == before_pipes + assert json.loads((chain_world.long_dir / "longitudinal.json").read_text("utf-8")) == { + "stations": [], + "old": True, + } + assert any( + sql.startswith("DELETE FROM routes WHERE project_id = %s AND id > %s") + for sql in chain_world.sql + ) + assert chain_world.order[-1] == "restore_params" + assert not list(chain_world.root.glob("route_save_*")) # 손에 쥔 사본 치움 + + +def test_추천_종단_반영이면_균형_종단(chain_world, monkeypatch): + monkeypatch.setattr( + route_initial, "apply_profile_target", _async(None, chain_world.order, "balanced") + ) + _run(chain_world, profile_target=([0.0, 456.0], [100.0, 110.0])) + assert "balanced" in chain_world.order and "profile" not in chain_world.order + + +# ── 초기값 규칙 ─────────────────────────────────────────────────────────────── + + +@pytest.fixture +def api_world(monkeypatch, tmp_path): + calls: list = [] + monkeypatch.setattr(replan, "_project_paths", _async((tmp_path, "p"))) + monkeypatch.setattr(replan, "_ensure_expected_route", lambda root: None) + monkeypatch.setattr(replan, "_plan_criteria", _async((20.0, 0.0, 0.0))) + monkeypatch.setattr(replan, "_ensure_planned_initial", lambda *a: None) + monkeypatch.setattr( + replan, + "_write_planned_polyline", + lambda *a, **k: {"vertices": 3, "curves": 0, "violations": 0}, + ) + monkeypatch.setattr(replan, "_recompute", _async({"route_id": 9, "total_length_m": 10.0})) + monkeypatch.setattr(replan, "read_initial_choice", lambda root: None) + monkeypatch.setattr( + route_initial, "refresh_initial_snapshot", lambda *a: calls.append(a) or _async(None)() + ) + return calls + + +def _body(**kwargs): + return save.RouteSaveRequest(vertices=[{"x": 0, "y": 0}, {"x": 5, "y": 5}], **kwargs) + + +def test_손으로_고친_노선_저장은_초기값_그대로(api_world): + result = asyncio.run(replan.replan_route(uuid4(), _body())) + assert result["initial_refreshed"] is False and api_world == [] + + +def test_추천_적용은_초기값_다시(api_world): + result = asyncio.run(replan.replan_route(uuid4(), _body(as_initial=True))) + assert result["initial_refreshed"] is True and len(api_world) == 1 + + +def test_초안_칸이_어긋나면_400(api_world): + body = save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]}) + response = asyncio.run(replan.replan_route(uuid4(), body)) + assert response.status_code == 400 + + +def _longitudinal(total: float) -> dict: + from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options + from B05_Profile.B05_Profile_Engine_Grade_Profile import design_ground_following_profile + + samples = [ + {"chainage_m": float(s), "elevation_m": 100.0 + 0.05 * s + 2.0 * np.sin(s / 17.0)} + for s in range(int(total) + 1) + ] + stations = [{"chainage_m": float(s)} for s in range(0, int(total) + 1, 20)] + longitudinal = {"samples": samples, "stations": stations} + alignment, profile = design_ground_following_profile( + longitudinal, resolve_grade_options("trunk") + ) + return {**longitudinal, "profile_alignment": alignment, "design_profiles": [profile]} + + +def test_계획고_옮김은_앞_그대로_뒤_밂(tmp_path, monkeypatch): + new_xy = [list(p) for p in NEW] + total = float(vertices_of(NEW)[-1].chainage_m) + old = _longitudinal(400.0) + old["profile_alignment"]["edits"] = { + "station_offsets": {"40.000": 0.7, "200.000": -0.4, "380.000": 0.3}, + "curve_radii": {}, + } + path = tmp_path / "B06_Section" / "longitudinal" / "longitudinal.json" + path.parent.mkdir(parents=True) + path.write_text(json.dumps(_longitudinal(total)), "utf-8") + import B05_Profile.B05_Profile_Repository as b05_repo + + monkeypatch.setattr(save, "get_db_pool", lambda: _Pool([])) + monkeypatch.setattr(b05_repo, "update_longitudinal_grade_summary", _async(None)) + asyncio.run( + save._carry_profile( + tmp_path, + "B06_Section/longitudinal/longitudinal.json", + 8, + old["profile_alignment"], + [list(p) for p in OLD], + new_xy, + ) + ) + edits = json.loads(path.read_text("utf-8"))["profile_alignment"]["edits"]["station_offsets"] + delta = total - 400.0 + assert edits["40.000"] == 0.7 # 바뀐 구간 앞 — 그대로 + assert "200.000" not in edits # 바뀐 구간 안 — 새 지반 + assert edits[f"{380.0 + delta:.3f}"] == 0.3 # 뒤 — 길이 차만큼 밂 + + +def test_저장_확정이_옛_횡단_파일을_지움(tmp_path): + from B05_Profile.B05_Profile_Engine_Sections import cross_filename + + long_path = tmp_path / "B06_Section" / "longitudinal" / "longitudinal.json" + cross_dir = tmp_path / "B06_Section" / "cross_sections" + long_path.parent.mkdir(parents=True) + cross_dir.mkdir(parents=True) + long_path.write_text(json.dumps({"stations": [{"chainage_m": 0.0}, {"chainage_m": 20.0}]})) + for at in (0.0, 20.0, 37.0): + (cross_dir / cross_filename(at)).write_text("{}") + b06_confirm._prune_cross_files(tmp_path, "B06_Section/longitudinal/longitudinal.json") + assert sorted(p.name for p in cross_dir.glob("cross_*.json")) == sorted( + cross_filename(at) for at in (0.0, 20.0) + ) + source = (ROOT / "B06_Section" / "B06_Section_Router_Confirm.py").read_text("utf-8") + for name in ("async def save_sections(", "async def confirm_sections("): + body = source[source.index(name) :] + assert body.index("_stand_patch_stations(") < body.index("_prune_cross_files,") diff --git a/resources/tester/test_346_3_route_save_screen.py b/resources/tester/test_346_3_route_save_screen.py new file mode 100644 index 000000000..154d25cb4 --- /dev/null +++ b/resources/tester/test_346_3_route_save_screen.py @@ -0,0 +1,109 @@ +"""346-3 화면 — [노선 반영] 없앰 · [저장]이 노선부터 · 추천 [적용] = as_initial 저장 · 저장 뒤 초안 넘김. + +노선 이음(`B05_Profile_UI_RouteEdit_Apply`)은 Node 헬퍼 `helper_346_3_route_save.cjs` 가 서버 · 화면 +창구만 가짜로 두고 돌림 · 서버 길 시험은 `test_346_3_route_save.py` · 저장 차례(노선 먼저 · 안 바뀌면 +안 보냄 · 실패 때 초안 남음)는 `test_design_store_save.py`. +""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parents[1] + + +def _read(rel: str) -> str: + return (ROOT / rel).read_text(encoding="utf-8") + + +@pytest.fixture(scope="module") +def out() -> dict: + run = subprocess.run( + ["node", str(HERE / "helper_346_3_route_save.cjs")], + capture_output=True, + text=True, + encoding="utf-8", + check=True, + ) + return json.loads(run.stdout) + + +def test_노선_안_바뀜_안_보냄(out): + assert out["noneAtAll"] is None + assert out["unchanged"] is None # 편집기가 정함 — 가게에 초안이 남아 있어도 + assert out["storeOnly"]["draft"]["rev"] == 3 # 편집기 없으면 가게 routeDraft + + +def test_초안은_계약_RouteDraft_꼴(out): + draft = out["body"]["draft"] + assert set(draft) == { + "rev", + "nodes", + "curveOn", + "curveRadius", + "curveLock", + "curveArc", + "stationSnap", + "terrain", + } + n = len(draft["nodes"]) + assert all( + len(draft[name]) == n for name in ("curveOn", "curveRadius", "curveLock", "curveArc") + ) + assert draft["curveOn"] == [True, False, True] and draft["curveRadius"] == [None, 25, None] + assert draft["stationSnap"] is True and draft["terrain"] == "sheet" + assert out["body"]["initial"] == {"mode": "follow"} and out["body"]["balanced_profile"] is True + assert "as_initial" not in out["body"] # 손으로 고친 노선 = 초기값 그대로 + + +def test_저장_뒤_가게_초안_비움_결과_한_줄(out): + assert out["savedRouteId"] == 9 + assert out["storeAfterSave"] is None + assert out["keptAfterSave"] == 1 + assert out["summary"] == ( + "노선 저장 — 관 지킴 3 · 새 교차 관 1 · 오목 보충 관 2 · 구조물 옮김 1 · 멀어짐 0 · " + "노선 밖 고침 3 건 · 배수유역 낡음" + ) # 서버 cross_dropped 1 + 화면 넘김 밖 2 = 한 줄 + + +def test_초안_넘김_차례(out): + assert out["carryCalls"] == ["invalidate:9", "load:9", "keys:9:9", "carry:7->9", "crossSave:9"] + assert out["outside"] == 2 + + +def test_추천_적용은_as_initial_저장_뒤_다시_세움(out): + assert out["recommendOk"] is True + body = out["recommendBody"] + assert body["as_initial"] is True and body["initial"] == {"mode": "earthwork"} + assert body["draft"]["nodes"] == [[1, 1], [9, 9]] + assert out["recommendCalls"] == ["overlay:on", "replan", "overlay:off"] # 잠금 막 하나 + assert out["recommendDraftsLeft"] == {"route": None, "pipes": None} # 새 초기값에서 다시 + assert out["rebuilt"] == 1 + + +def test_추천_저장_실패는_한_줄_다시_안_세움(out): + assert out["recommendFail"] is False + errors = [c for c in out["recommendFailCalls"] if c.startswith("toast:error:")] + assert len(errors) == 1 and "체인 실패" in errors[0] + assert out["rebuiltAfterFail"] == 1 + + +def test_노선_반영_단추_없앰_배선(): + chrome = _read("B05_Profile/B05_Profile_UI_RouteEdit_Chrome.ts") + assert 'data-act="apply"' not in chrome and "노선 반영" not in chrome + apply = _read("B05_Profile/B05_Profile_UI_RouteEdit_Apply.ts") + assert "addEventListener" not in apply # 단추 듣기 없음 — [저장]이 부름 + save = _read("A00_Common/design/design_save.ts") + assert save.index('"route",') < save.index('"pipes",') + assert "await pendingRouteSave(ctx.projectId)" in save + assert "carryAfterRouteSave(ctx.projectId, fromRouteId, routeId)" in save + initial = _read("B05_Profile/B05_Profile_UI_RouteEdit_Initial.ts") + at = initial.index("onApply: (picked) => {") + assert "saveRecommendedRoute(" in initial[at : at + 900] + replan = _read("B05_Profile/B05_Profile_Api_Replan.ts") + assert "body: JSON.stringify(body)" in replan and "as_initial?: boolean;" in replan