feat(노선추천): 346-3 추천 [적용] = as_initial 로 서버 노선 저장 — 초기값 = 추천 노선 · 잠금 막 · 초안 비움 · 새로고침 없이 다시 세움 · 실패 = 오류 한 줄 · 평면에 실린 채 남음

- _RouteEdit_Initial onApply → 평면 되돌리기 한 칸 그대로 + saveRecommendedRoute
- 시험: test_346_3_route_save 7(안 바뀜 안 보냄 · 계약 꼴 · 저장 뒤 가게 비움 · 결과 한 줄 · 넘김 차례 · as_initial · 실패) · 관련 242 통과 · typecheck 통과

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VmC34fsxxbweEG3RUjs3Mn
This commit is contained in:
2026-10-16 12:14:22 +09:00
co-authored by Claude Opus 5.5
parent 47366d3740
commit 6086f30982
3 changed files with 269 additions and 538 deletions
@@ -1,14 +1,15 @@
/* ============================================================================= /* =============================================================================
* B05_Profile_UI_RouteEdit_Initial.ts * B05_Profile_UI_RouteEdit_Initial.ts
* 계획노선 편집의 **노선 추천 이음**(계획서 20 · 23-2) — 좌측 「노선 추천」 칸(345-4) [노선 추천] 단추 → * 계획노선 편집의 **노선 추천 이음**(계획서 20 · 23-2) — 좌측 「노선 추천」 칸(345-4) [노선 추천] 단추 →
* 모달(`_Recommend`) · [초기화] · 「현재 추천: ○○」 · [노선 반영]에 실을 지금 추천. 칸 요소는 * 모달(`_Recommend`) · [초기화] · 「현재 추천: ○○」 · [저장]에 실을 지금 추천. 칸 요소는
* 밖(`_Panel_Recommend`)에서 받는다. * 밖(`_Panel_Recommend`)에서 받는다.
* *
* 모달의 알약이 서버에 갈래를 묻고(`/route/initial`) 미리 보기만 한다. [적용]을 눌러야 받은 노드를 * 모달의 알약이 서버에 갈래를 묻고(`/route/initial`) 미리 보기만 한다. [적용]을 눌러야 받은 노드를
* **편집값에 싣고**(되돌리기 한 걸음) 캐시(`routeedit-initial`)에 적는다. [초기화]는 서버를 다시 * **편집값에 싣고**(되돌리기 한 걸음) 캐시(`routeedit-initial`)에 적는다. [초기화]는 서버를 다시
* 셈하지 않고 **전처리 때 만든 초기값 노선**(예상노선 추적)을 놓는다 — 그 노드는 `GET /route/plan` 의 * 셈하지 않고 **전처리 때 만든 초기값 노선**(예상노선 추적)을 놓는다 — 그 노드는 `GET /route/plan` 의
* `initial_route` 한 곳에서 읽는다. 종단 · 배수유역 · 횡단 다시 계산은 여기서 안 돈다 — * `initial_route` 한 곳에서 읽는다. 손으로 고친 노선의 다시 계산은 페이지 [저장]에서(`_Apply` · 346-3) —
* 노선이 바뀐 뒤 [확인]에서만(`_Apply`). 비교값 · 기준값은 서버가 준 것을 보이기만 한다. * 추천 [적용]만 그 자리에서 서버 노선을 `as_initial` 로 저장한다(초기값 = 추천 노선 · 사용자 확정 · 잠금 막 ·
* 끝나면 페이지 다시 세움). 비교값 · 기준값은 서버가 준 것을 보이기만 한다.
* *
* 공통 제약 네 칸(종단기울기 % · 최소 반지름 m · 계류 이격 m · 지반 고름 m, 계획서 20 확정) * 공통 제약 네 칸(종단기울기 % · 최소 반지름 m · 계류 이격 m · 지반 고름 m, 계획서 20 확정)
* — 비우면 서버 기본값을 쓴다. 채운 칸만 요청에 실어 보내고, 그 「채운 칸」만 캐시 · * — 비우면 서버 기본값을 쓴다. 채운 칸만 요청에 실어 보내고, 그 「채운 칸」만 캐시 ·
@@ -41,6 +42,7 @@ import {
type InitialChoice, type InitialChoice,
} from "./B05_Profile_UI_RouteEdit_Initial_Model"; } from "./B05_Profile_UI_RouteEdit_Initial_Model";
import { openRecommendModal } from "./B05_Profile_UI_RouteEdit_Recommend"; import { openRecommendModal } from "./B05_Profile_UI_RouteEdit_Recommend";
import { saveRecommendedRoute } from "./B05_Profile_UI_RouteEdit_Apply";
export interface InitialPanelParams { export interface InitialPanelParams {
/** 단추 · 글 요소 — 좌측 「노선 추천」 칸 것(345-4). */ /** 단추 · 글 요소 — 좌측 「노선 추천」 칸 것(345-4). */
@@ -61,7 +63,7 @@ export interface InitialPanelParams {
export interface InitialPanel { export interface InitialPanel {
/** 노선을 읽었다 — 창을 연 편집값 · 「고친 노선인가」 · 서버가 적어 둔 현재 초기값(`current_initial`). */ /** 노선을 읽었다 — 창을 연 편집값 · 「고친 노선인가」 · 서버가 적어 둔 현재 초기값(`current_initial`). */
ready: (opening: RouteEditSnapshot, edited: boolean, server?: unknown) => void; ready: (opening: RouteEditSnapshot, edited: boolean, server?: unknown) => void;
/** [확인]에 실어 보낼 지금 추천. */ /** [저장]에 실어 보낼 지금 추천. */
current: () => { current: () => {
mode: RouteInitialLogic; mode: RouteInitialLogic;
weights: RouteInitialWeights | null; weights: RouteInitialWeights | null;
@@ -160,7 +162,18 @@ export function createInitialPanel(params: InitialPanelParams): InitialPanel {
render(); render();
params.apply( params.apply(
copy(picked.snapshot), copy(picked.snapshot),
`노선 추천 「${choiceLabel(picked.logic, picked.weights)}」 적용 — 다시 계산은 [확인]에서.`, `노선 추천 「${choiceLabel(picked.logic, picked.weights)}」 적용 — 서버에 저장합니다.`,
);
// 초기값 = 추천 노선(346-3) — 바로 저장 · 실패하면 평면에 실린 채 남아 [저장]이 다시 보냄.
void saveRecommendedRoute(
projectId,
picked.snapshot,
{
mode: picked.logic,
weights: picked.weights,
criteria: picked.criteriaOverrides,
},
withProfileOf(picked.logic),
); );
}, },
}); });
@@ -193,7 +206,7 @@ export function createInitialPanel(params: InitialPanelParams): InitialPanel {
writeInitialChoice(projectId, choice); writeInitialChoice(projectId, choice);
params.apply( params.apply(
copy(snapshot), copy(snapshot),
"초기값(예상노선 추적)으로 되돌렸습니다 — 다시 계산은 [확인]에서.", "초기값(예상노선 추적)으로 되돌렸습니다 — 다시 계산은 [저장]에서.",
); );
} finally { } finally {
busy = false; busy = false;
@@ -0,0 +1,171 @@
/* 346-3 노선 저장 이음 헬퍼 — `B05_Profile_UI_RouteEdit_Apply.ts` 를 트랜스파일해 서버 · 화면 창구는 가짜로,
* 가게 · 캐시(`design_store` · `b_page_state`)는 진짜로 돌림. 보낸 본문 · 부른 차례를 적어 JSON 한 벌로 냄.
*/
const fs = require("fs");
const path = require("path");
const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript"));
const ROOT = path.join(__dirname, "..", "..");
function memoryStorage() {
const map = new Map();
return {
getItem: (key) => (map.has(key) ? map.get(key) : null),
setItem: (key, value) => map.set(key, String(value)),
removeItem: (key) => map.delete(key),
key: (index) => [...map.keys()][index] ?? null,
get length() {
return map.size;
},
};
}
const session = memoryStorage();
globalThis.window = { sessionStorage: session, localStorage: memoryStorage() };
globalThis.sessionStorage = session;
const calls = [];
const bodies = [];
let failReplan = false;
const STUBS = {
"@ui/ui_template_elements": {
showToast: (text, kind) => calls.push(`toast:${kind}:${text}`),
showLoadingOverlay: () => calls.push("overlay:on"),
hideLoadingOverlay: () => calls.push("overlay:off"),
},
"../A00_Common/router": {
currentRoute: () => "b05-profile",
remountRoute: () => calls.push("remount"),
},
"../B06_Section/B06_Section_Section_Store": {
invalidateSectionDetail: (_p, routeId) => calls.push(`invalidate:${routeId ?? "all"}`),
loadSectionDetail: async (_p, routeId) => (calls.push(`load:${routeId}`), { tag: routeId }),
saveCachedCrossPatches: async (_p, routeId) => (calls.push(`crossSave:${routeId}`), 1),
},
"../B06_Section/B06_Section_Station_Keys": {
useStationKeys: (_p, routeId, detail) => calls.push(`keys:${routeId}:${detail.tag}`),
carryStationDrafts: (_p, from, to) => (calls.push(`carry:${from}->${to}`), 2),
},
"./B05_Profile_Api_Fetch": { clearRouteLatestCache: () => calls.push("latest:clear") },
"./B05_Profile_Api_Replan": {
replanRoute: async (_p, body) => {
bodies.push(JSON.parse(JSON.stringify(body)));
calls.push("replan");
if (failReplan) throw new Error("체인 실패");
return {
route_id: 9,
pipes: { kept: 3, crossing: 1, concave: 2 },
structures: { moved: 1, far: 0 },
cross_dropped: 1,
drainage_stale: true,
};
},
},
};
const cache = new Map();
function loadTs(file) {
const full = path.resolve(file);
if (cache.has(full)) return cache.get(full);
const js = ts.transpileModule(fs.readFileSync(full, "utf8"), {
compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 },
}).outputText;
const box = { exports: {} };
cache.set(full, box.exports);
const localRequire = (request) => {
if (STUBS[request]) return STUBS[request];
if (request.startsWith(".")) return loadTs(path.join(path.dirname(full), `${request}.ts`));
return require(request);
};
new Function("exports", "module", "require", js)(box.exports, box, localRequire);
return box.exports;
}
const { useDesignStore } = loadTs(path.join(ROOT, "A00_Common", "design", "design_store.ts"));
const { readState, writeState } = loadTs(path.join(ROOT, "A00_Common", "b_page_state.ts"));
const apply = loadTs(path.join(ROOT, "B05_Profile", "B05_Profile_UI_RouteEdit_Apply.ts"));
(async () => {
const out = {};
const draft = {
rev: 3,
nodes: [
[0, 0],
[5, 5],
[10, 0],
],
curveOn: [true, true, true],
};
// 편집기 없음 — 가게 routeDraft 를 보냄 · 없으면 null
out.noneAtAll = await apply.pendingRouteSave("p1");
useDesignStore("p1", null).set("routeDraft", draft);
out.storeOnly = await apply.pendingRouteSave("p1");
// 편집기 등록 — 안 바뀌었으면 null(가게에 초안이 있어도 편집기가 정함) · 바뀌면 계약 꼴
let changed = false;
let kept = 0;
let rebuilt = 0;
apply.bindRouteApply({
projectId: "p1",
nodes: () => ({
planned: [
[0, 0],
[5, 5],
[10, 0],
],
curveOn: [true, false],
curveRadius: [null, 25],
}),
changed: () => changed,
terrain: () => "sheet",
stationSnap: () => true,
initial: { current: () => ({ mode: "follow" }), keep: () => (kept += 1) },
profileShown: async () => true,
onApplied: () => (rebuilt += 1),
});
out.unchanged = await apply.pendingRouteSave("p1");
changed = true;
const body = await apply.pendingRouteSave("p1");
out.body = body;
// 저장 — 가게 초안 비움 · 결과 캐시 비움 · 추천 다시 적음
writeState("design-stations", [{ k: 1 }], "p1");
const saved = await apply.saveRouteDraft("p1", body);
out.savedRouteId = saved.route_id;
out.storeAfterSave = useDesignStore("p1", null).get("routeDraft");
out.keptAfterSave = kept;
out.summary = apply.routeSaveSummary(saved, 2);
// 저장 뒤 횡단 초안 넘김 차례 — 새 상세 → 측점 열쇠 → 넘김 → 넘긴 고침 저장
calls.length = 0;
out.outside = await apply.carryAfterRouteSave("p1", 7, 9);
out.carryCalls = [...calls];
// 추천 [적용] — as_initial · 잠금 막 · 초안 비움 · 다시 세움
useDesignStore("p1", null).set("routeDraft", draft);
writeState("pipes", [{ chainage_m: 1 }], "p1");
calls.length = 0;
const snapshot = {
planned: [
[1, 1],
[9, 9],
],
curveOn: [true, true],
curveRadius: [null, null],
};
out.recommendOk = await apply.saveRecommendedRoute("p1", snapshot, { mode: "earthwork" }, true);
out.recommendBody = bodies[bodies.length - 1];
out.recommendCalls = calls.filter((c) => !c.startsWith("toast:"));
out.recommendDraftsLeft = {
route: useDesignStore("p1", null).get("routeDraft"),
pipes: readState("pipes", "p1"),
};
out.rebuilt = rebuilt;
// 추천 저장 실패 — false · 오류 한 줄 · 다시 안 세움
failReplan = true;
calls.length = 0;
out.recommendFail = await apply.saveRecommendedRoute("p1", snapshot, null, false);
out.recommendFailCalls = [...calls];
out.rebuiltAfterFail = rebuilt;
console.log(JSON.stringify(out));
})();
+79 -532
View File
@@ -1,561 +1,108 @@
# -*- coding: utf-8 -*- """346-3 화면 — [노선 반영] 없앰 · [저장]이 노선부터 · 추천 [적용] = as_initial 저장 · 저장 뒤 초안 넘김.
"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 오목 보충 · 이월 열쇠 ·
초기값 규칙 ·
배수유역 낡음 · 실패 때 옛 노선 되살림.
순수 부품(`B05_Profile_Route_Save_Parts`)은 값으로 · 체인(`run_route_save_chain`)과 API 노선 이음(`B05_Profile_UI_RouteEdit_Apply`)은 Node 헬퍼 `helper_346_3_route_save.cjs` 가 서버 · 화면 창구만
(`replan_route`)는 DB · 엔진을 가짜로 바꿔 차례 · 되살림 · 초기값만 잰다. 가짜로 두고 돌림 · 저장 차례(노선 먼저 · 안 바뀌면 안 보냄 · 실패 때 초안 남음)는 `test_design_store_save.py`.
""" """
from __future__ import annotations from __future__ import annotations
import asyncio
import json import json
import sys import subprocess
from pathlib import Path from pathlib import Path
from types import SimpleNamespace
from uuid import uuid4
import numpy as np
import pytest import pytest
ROOT = Path(__file__).resolve().parents[2] HERE = Path(__file__).resolve().parent
if str(ROOT) not in sys.path: ROOT = HERE.parents[1]
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")
def test_관은_지우지_않고_좌표를_따라감(): @pytest.fixture(scope="module")
old = [ def out() -> dict:
PipePoint(chainage_m=50.0, x=50.0, y=0.0, options={"pipe_kind": "흄관"}), run = subprocess.run(
PipePoint(chainage_m=350.0, x=350.0, y=0.0, start_m=345.0, end_m=355.0), ["node", str(HERE / "helper_346_3_route_save.cjs")],
] capture_output=True,
pairs = follow_pipes(old, vertices_of(NEW)) text=True,
assert [pipe_id for pipe_id, _ in pairs] == [0, 1] encoding="utf-8",
assert pairs[0][1].options == {"pipe_kind": "흄관"} # 사용자 관 속성 그대로 check=True,
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_측점_열쇠_규칙(): def test_노선_안_바뀜_안_보냄(out):
stations = [{"chainage_m": s} for s in (0.0, 20.0, 40.0, 50.0, 60.0, 77.0, 80.0, 95.5)] assert out["noneAtAll"] is None
pipes = [(3, PipePoint(chainage_m=50.0))] assert out["unchanged"] is None # 편집기가 정함 — 가게에 초안이 남아 있어도
structures = [_structure("w1", "retaining_wall", 77.0, 60.0, 80.0)] assert out["storeOnly"]["draft"]["rev"] == 3 # 편집기 없으면 가게 routeDraft
keys = station_keys(stations, pipes, structures, 20.0)
assert keys == {
"end:BP": 0.0, def test_초안은_계약_RouteDraft_꼴(out):
"grid:1": 20.0, draft = out["body"]["draft"]
"grid:2": 40.0, assert set(draft) == {
"pipe:3": 50.0, "rev",
"structure:w1:start": 60.0, "nodes",
"structure:w1": 77.0, "curveOn",
"structure:w1:end": 80.0, "curveRadius",
"end:EP": 95.5, "curveLock",
"curveArc",
"stationSnap",
"terrain",
} }
n = len(draft["nodes"])
assert all(
def test_이월은_열쇠로_누가거리가_밀려도(): len(draft[name]) == n for name in ("curveOn", "curveRadius", "curveLock", "curveArc")
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 inputs == { assert draft["curveOn"] == [True, False, True] and draft["curveRadius"] == [None, 25, None]
0.0: {"ground_type": "soil"}, assert draft["stationSnap"] is True and draft["terrain"] == "sheet"
20.0: {"ground_type": "rock", "berm": {"on": True}}, # 결과 칸(면적)은 안 넘김 assert out["body"]["initial"] == {"mode": "follow"} and out["body"]["balanced_profile"] is True
57.0: {"ditch_type": "u"}, assert "as_initial" not in out["body"] # 손으로 고친 노선 = 초기값 그대로
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_초안_모양을_옛_칸으로(): def test_초안_넘김_차례(out):
body = save.RouteSaveRequest( assert out["carryCalls"] == ["invalidate:9", "load:9", "keys:9:9", "carry:7->9", "crossSave:9"]
draft={ assert out["outside"] == 2
"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
class _Cursor: def test_추천_저장_실패는_한_줄_다시_안_세움(out):
def __init__(self, log): assert out["recommendFail"] is False
self.log = log errors = [c for c in out["recommendFailCalls"] if c.startswith("toast:error:")]
assert len(errors) == 1 and "체인 실패" in errors[0]
async def __aenter__(self): assert out["rebuiltAfterFail"] == 1
return self
async def __aexit__(self, *exc):
return False
async def execute(self, sql, params=None):
self.log.append(" ".join(sql.split()))
class _Connection: def test_노선_반영_단추_없앰_배선():
def __init__(self, log): chrome = _read("B05_Profile/B05_Profile_UI_RouteEdit_Chrome.ts")
self.log = log assert 'data-act="apply"' not in chrome and "노선 반영</button>" not in chrome
apply = _read("B05_Profile/B05_Profile_UI_RouteEdit_Apply.ts")
def cursor(self, *args): assert "addEventListener" not in apply # 단추 듣기 없음 — [저장]이 부름
return _Cursor(self.log) save = _read("A00_Common/design/design_save.ts")
assert save.index('"route",') < save.index('"pipes",')
async def begin(self): assert "await pendingRouteSave(ctx.projectId)" in save
return None assert "carryAfterRouteSave(ctx.projectId, fromRouteId, routeId)" in save
initial = _read("B05_Profile/B05_Profile_UI_RouteEdit_Initial.ts")
async def commit(self): at = initial.index("onApply: (picked) => {")
return None assert "saveRecommendedRoute(" in initial[at : at + 900]
replan = _read("B05_Profile/B05_Profile_Api_Replan.ts")
async def rollback(self): assert "body: JSON.stringify(body)" in replan and "as_initial?: boolean;" in replan
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,")