eomsangdonandClaude Opus 5.5
4b6800e387
- POST /design/save 에 route(노선 초안) · carry(옛 노선 세션 칸) 칸 · 서버 = 노선 저장 → Node facility.rebase(rebasePipeDraft · rebaseStructureDraft) → 6-A 차례(새 노선) → Node stations.carry(routeCarryPatches) → 횡단 저장 - 되돌림 범위 = 노선 행 · 자식 표 · 단계 · 배수 폴더 · 노선 파일 · 지표면 확정(설계 지형 바꿈) - 손 상단측은 안 넘김(요구 274) · 노선 바뀐 [확정] = [저장](새 노선 본 뒤 다시) - 브라우저 = 노선 초안 + 사용자 칸만 한 본문 · 옛 차례 길(saveRouteDraft 뒤 flush · carryAfterRouteSave · keepDraftsOnNewRoute) · 부름 0 된 flush 창구 여섯 · saveCachedCrossPatches 걷음 - 등록부 R9 stations.carry · F10 facility.rebase · 옛 차례 단언 시험 한 본문 꼴로 · test_472_6_route_save(옛 길 = 새 길 · 실패 되돌림) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
412 lines
17 KiB
Python
412 lines
17 KiB
Python
"""PLAN 448(442-2 · 요구 269-1) — B05 페이지 설정 [저장] · [확정] 영구저장.
|
|
|
|
- 병합 — stage 2 params 에 사용자 값만 얹음(`options` 칸끼리 · 상태 · STALE 안 건드림)
|
|
- 정책 갱신 — 종단 경고 기준(최대 기울기 · 종단곡선 R · 평면 R)만 · 기하(base_pvi · 편집) 그대로
|
|
- 작업임도 30 → 20 해석 · params 엔 원값
|
|
- E1 — 노선 바꾼 저장 · B04 재확정 요청이 설계속도 · 최소고 강제 · 시작 측점을 옮김
|
|
- E2 — B06 재생성 지형 기본 = 특수(B05 기본)
|
|
- E3 — 초기값 스냅샷이 stage 2 params 를 뜨고 [초기화]가 되돌림 · 옛 스냅샷은 설정 칸만 걷음
|
|
- 저장 차례 · 초안 내보내기 · 끝 측점 편집 — Node 헬퍼 `helper_442_2_page_settings.cjs`
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_442_2_page_settings.py -q
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from B03_FileInput.B03_FileInput_Service_Chain import redesign_solve_request
|
|
from B05_Profile import B05_Profile_PageSettings as page
|
|
from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options
|
|
from B05_Profile.B05_Profile_Engine_Grade_Profile import design_alignment_profile
|
|
from B05_Profile.B05_Profile_Schema import PageSettingsRequest
|
|
from B06_Section.B06_Section_Router_Regenerate_Options import _regeneration_grade_options
|
|
from common_util import common_util_initial_snapshot as snapshot
|
|
from common_util.common_util_workflow_state import merge_stage_params, merged_params
|
|
|
|
HERE = Path(__file__).parent
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def out():
|
|
proc = subprocess.run(
|
|
["node", str(HERE / "helper_442_2_page_settings.cjs")],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
timeout=60,
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
return json.loads(proc.stdout)
|
|
|
|
|
|
PARAMS = {
|
|
"points": {"bp": {"x": 0, "y": 0}, "ep": {"x": 1, "y": 1}},
|
|
"options": {"grade_class": "trunk", "terrain_type": "special", "paved": False},
|
|
"station_interval_m": 20,
|
|
"max_grade_pct": 13.0,
|
|
"start_elevation_offset_m": 0.4,
|
|
}
|
|
|
|
|
|
# ── 병합 ─────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_병합_options_칸끼리_나머지_갈음():
|
|
request = PageSettingsRequest(
|
|
design_speed_kph=30,
|
|
terrain_type="normal",
|
|
max_grade_pct=None,
|
|
min_vertical_radius_m=300,
|
|
station_display={"station": 12.4, "cumulative": 240},
|
|
)
|
|
merged = merged_params(PARAMS, page.params_patch(request))
|
|
assert merged["options"] == {
|
|
"grade_class": "trunk",
|
|
"terrain_type": "normal",
|
|
"paved": False,
|
|
"design_speed_kph": 30,
|
|
}
|
|
assert merged["max_grade_pct"] is None # 비운 칸 = 법정값
|
|
assert merged["min_vertical_radius_m"] == 300
|
|
assert merged["station_display"] == {"station": 12.0, "cumulative": 240.0}
|
|
assert merged["points"] == PARAMS["points"] and merged["start_elevation_offset_m"] == 0.4
|
|
assert PARAMS["options"]["terrain_type"] == "special" # 원본 안 바뀜
|
|
|
|
|
|
class _Cursor:
|
|
def __init__(self, params):
|
|
self.params = params
|
|
self.sql: list[str] = []
|
|
|
|
async def execute(self, sql, args=None):
|
|
self.sql.append(" ".join(sql.split()))
|
|
if sql.strip().upper().startswith("UPDATE"):
|
|
self.params = args[0]
|
|
|
|
async def fetchone(self):
|
|
return {"params": self.params}
|
|
|
|
|
|
def test_merge_stage_params_상태_안_건드림():
|
|
cursor = _Cursor(json.dumps(PARAMS))
|
|
merged = asyncio.run(merge_stage_params(cursor, "p1", 2, {"options": {"design_speed_kph": 40}}))
|
|
assert merged["options"]["design_speed_kph"] == 40
|
|
assert json.loads(cursor.params)["options"]["grade_class"] == "trunk"
|
|
assert "FOR UPDATE" in cursor.sql[0]
|
|
assert len(cursor.sql) == 2 and cursor.sql[1].startswith("UPDATE project_workflow_stages SET")
|
|
assert all("state" not in sql.lower() and "projects" not in sql for sql in cursor.sql[1:])
|
|
|
|
|
|
def test_요청_검증():
|
|
with pytest.raises(ValueError):
|
|
PageSettingsRequest(design_speed_kph=50)
|
|
with pytest.raises(ValueError):
|
|
PageSettingsRequest(terrain_type="flat")
|
|
with pytest.raises(ValueError):
|
|
PageSettingsRequest(max_grade_pct=0)
|
|
with pytest.raises(ValueError):
|
|
PageSettingsRequest(extra=1)
|
|
|
|
|
|
def test_작업임도_30은_20으로_params_는_원값():
|
|
request = PageSettingsRequest(design_speed_kph=30, terrain_type="normal")
|
|
params = merged_params({"options": {}}, page.params_patch(request))
|
|
assert params["options"]["design_speed_kph"] == 30 # 원값 저장(해석값 금지)
|
|
work = page.grade_options_of("work", params)
|
|
assert (work.design_speed_kph, work.max_grade_pct, work.min_vertical_radius_m) == (20, 9, 100)
|
|
trunk = page.grade_options_of("main", params) # B02 main = 간선
|
|
assert (trunk.design_speed_kph, trunk.max_grade_pct, trunk.min_vertical_radius_m) == (
|
|
30,
|
|
8,
|
|
250,
|
|
)
|
|
typed = page.grade_options_of("main", {**params, "max_grade_pct": 11.0})
|
|
assert typed.max_grade_pct == 11.0 and typed.warnings # 법정 8 초과 = 경고(막지 않음)
|
|
|
|
|
|
# ── 정책 갱신 ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _longitudinal() -> dict:
|
|
"""앞 200 m 11 % 오르막 · 뒤 200 m 2 % — 특수 20(14 %)엔 위반 없음 · 일반 30(8 %)엔 위반."""
|
|
chainage = np.arange(0.0, 400.5, 1.0)
|
|
ground = np.where(chainage <= 200, 100 + 0.11 * chainage, 122 + 0.02 * (chainage - 200))
|
|
longitudinal = {
|
|
"samples": [
|
|
{"chainage_m": float(s), "elevation_m": float(z)} for s, z in zip(chainage, ground)
|
|
],
|
|
"stations": [
|
|
{"station_id": f"s{i}", "chainage_m": float(s)}
|
|
for i, s in enumerate(np.arange(0, 401, 20))
|
|
],
|
|
}
|
|
options = resolve_grade_options("trunk", terrain_type="special", design_speed_kph=20)
|
|
alignment, profile = design_alignment_profile(longitudinal, options)
|
|
edits = {"station_offsets": {"100.000": 0.8}, "curve_radii": {}}
|
|
longitudinal["profile_alignment"] = alignment
|
|
longitudinal["design_profiles"] = [profile]
|
|
from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile
|
|
|
|
alignment, profile = rebuild_alignment_profile(longitudinal, edits)
|
|
longitudinal["profile_alignment"] = alignment
|
|
longitudinal["design_profiles"] = [profile]
|
|
return longitudinal
|
|
|
|
|
|
def test_정책_갱신_기하_그대로():
|
|
before = _longitudinal()
|
|
after = copy.deepcopy(before)
|
|
strict = resolve_grade_options("trunk", terrain_type="normal", design_speed_kph=30)
|
|
alignment, summary = page.retarget_longitudinal(after, strict)
|
|
policy = alignment["policy"]
|
|
assert (policy["max_grade_pct"], policy["min_vertical_radius_m"]) == (8, 250)
|
|
assert policy["min_plan_radius_m"] == 30 # 30 km/h 일반지형
|
|
criteria = after["design_profiles"][0]["criteria"]
|
|
assert (criteria["design_speed_kph"], criteria["terrain_type"]) == (30, "normal")
|
|
assert not before["profile_alignment"]["violations"]
|
|
assert after["profile_alignment"]["violations"] # 같은 선 · 엄한 기준 = 위반 뜸
|
|
assert summary["id"] == after["design_profiles"][0]["id"]
|
|
# 기하 그대로 — 자동 선형 · 편집 · 계획고 표본
|
|
assert alignment["base_pvi"] == before["profile_alignment"]["base_pvi"]
|
|
assert alignment["edits"] == before["profile_alignment"]["edits"]
|
|
z0 = [s["elevation_m"] for s in before["profile_alignment"]["samples"]]
|
|
z1 = [s["elevation_m"] for s in alignment["samples"]]
|
|
assert np.allclose(z0, z1)
|
|
|
|
|
|
def test_retarget_file_선형_없으면_그대로(tmp_path):
|
|
options = resolve_grade_options("trunk", terrain_type="normal")
|
|
rel = "B06_Section/longitudinal/longitudinal.json"
|
|
path = tmp_path / rel
|
|
path.parent.mkdir(parents=True)
|
|
path.write_text(json.dumps({"samples": []}), encoding="utf-8")
|
|
assert page.retarget_file(tmp_path, rel, options) is None
|
|
assert json.loads(path.read_text(encoding="utf-8")) == {"samples": []}
|
|
assert page.retarget_file(tmp_path, "nope.json", options) is None
|
|
with pytest.raises(ValueError):
|
|
page.retarget_file(tmp_path, "../outside.json", options)
|
|
path.write_text(json.dumps(_longitudinal()), encoding="utf-8")
|
|
summary = page.retarget_file(tmp_path, rel, options)
|
|
stored = json.loads(path.read_text(encoding="utf-8"))
|
|
assert summary is not None and stored["profile_alignment"]["policy"]["max_grade_pct"] == 9
|
|
|
|
|
|
# ── E1 · E2 ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_E1_재설계_요청이_설정을_옮김():
|
|
params = {
|
|
**PARAMS,
|
|
"options": {**PARAMS["options"], "design_speed_kph": 30},
|
|
"enforce_pipe_clearance": True,
|
|
"station_display": {"station": 12, "cumulative": 240},
|
|
}
|
|
request = redesign_solve_request(params, {"source_filter": "csf"}, 5)
|
|
assert request.design_speed_kph == 30
|
|
assert request.enforce_pipe_clearance is True
|
|
assert request.station_display == {"station": 12, "cumulative": 240}
|
|
assert request.terrain_type == "special"
|
|
|
|
|
|
def test_E2_B06_재생성_지형_기본_특수():
|
|
no_terrain = _regeneration_grade_options({"options": {"grade_class": "trunk"}})
|
|
assert no_terrain.terrain_type == "special" and no_terrain.max_grade_pct == 14
|
|
saved = _regeneration_grade_options(
|
|
{"options": {"grade_class": "trunk", "terrain_type": "normal", "design_speed_kph": 30}}
|
|
)
|
|
assert (saved.terrain_type, saved.max_grade_pct) == ("normal", 8)
|
|
|
|
|
|
def test_노선_추천_설정은_options_에서_읽음(monkeypatch):
|
|
from B05_Profile import B05_Profile_Router_Replan as replan
|
|
|
|
stage = {
|
|
"stage_no": 2,
|
|
"params": {"options": {"design_speed_kph": 30, "terrain_type": "normal"}},
|
|
}
|
|
|
|
class _Cur:
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_):
|
|
return False
|
|
|
|
async def execute(self, *_):
|
|
return None
|
|
|
|
async def fetchone(self):
|
|
return ("main",)
|
|
|
|
class _Conn:
|
|
def cursor(self, *_):
|
|
return _Cur()
|
|
|
|
class _Acquire:
|
|
async def __aenter__(self):
|
|
return _Conn()
|
|
|
|
async def __aexit__(self, *_):
|
|
return False
|
|
|
|
monkeypatch.setattr(replan, "get_db_pool", lambda: SimpleNamespace(acquire=_Acquire))
|
|
|
|
async def fake_state(*_):
|
|
return {"stages": [stage]}
|
|
|
|
import common_util.common_util_workflow_state as workflow
|
|
|
|
monkeypatch.setattr(workflow, "get_workflow_state", fake_state)
|
|
assert asyncio.run(replan._road_settings("p1")) == ("main", 30, "normal")
|
|
|
|
|
|
# ── E3 스냅샷 ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class _SnapCursor:
|
|
def __init__(self, state):
|
|
self.state = state
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_):
|
|
return False
|
|
|
|
async def execute(self, sql, args=None):
|
|
self.state["sql"].append(sql)
|
|
if sql.strip().upper().startswith("UPDATE"):
|
|
self.state["params"] = args[0]
|
|
|
|
async def fetchone(self):
|
|
return (self.state["params"],)
|
|
|
|
|
|
class _SnapConn:
|
|
def __init__(self, state):
|
|
self.state = state
|
|
|
|
def cursor(self, *_):
|
|
return _SnapCursor(self.state)
|
|
|
|
|
|
def test_E3_스냅샷_stage2_왕복():
|
|
initial = {**PARAMS, "options": {**PARAMS["options"], "design_speed_kph": 20}}
|
|
state = {"params": json.dumps(initial), "sql": []}
|
|
dumped = asyncio.run(snapshot._stage2_params(_SnapConn(state), "p1"))
|
|
assert dumped == initial
|
|
state["params"] = json.dumps(
|
|
{**initial, "max_grade_pct": 7, "options": {"design_speed_kph": 40}}
|
|
)
|
|
asyncio.run(snapshot._restore_stage2_params(_SnapConn(state), "p1", {"stage2_params": dumped}))
|
|
assert json.loads(state["params"]) == initial
|
|
|
|
|
|
def test_E3_옛_스냅샷은_설정_칸만_걷음():
|
|
edited = {
|
|
**PARAMS,
|
|
"options": {**PARAMS["options"], "design_speed_kph": 40, "terrain_type": "normal"},
|
|
"min_vertical_radius_m": 500,
|
|
"enforce_pipe_clearance": True,
|
|
"station_display": {"station": 3, "cumulative": 60},
|
|
}
|
|
state = {"params": json.dumps(edited), "sql": []}
|
|
asyncio.run(snapshot._restore_stage2_params(_SnapConn(state), "p1", {"routes": []}))
|
|
restored = json.loads(state["params"])
|
|
assert restored["options"] == {"grade_class": "trunk", "paved": False}
|
|
for key in page.PAGE_TOP_KEYS:
|
|
assert key not in restored
|
|
assert restored["points"] == PARAMS["points"] and restored["start_elevation_offset_m"] == 0.4
|
|
|
|
|
|
def test_E3_뜰_params_없으면_안_씀():
|
|
state = {"params": None, "sql": []}
|
|
asyncio.run(snapshot._restore_stage2_params(_SnapConn(state), "p1", {"stage2_params": None}))
|
|
assert not any(sql.strip().upper().startswith("UPDATE") for sql in state["sql"])
|
|
|
|
|
|
# ── 화면(Node) ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_저장_차례_settings_는_노선_뒤_관_앞(out):
|
|
# 472-6 — 차례 = 서버 한 창구(노선 맨 앞 · 6-B)
|
|
server = (HERE.parents[1] / "B06_Section" / "B06_Section_Router_DesignSave.py").read_text(
|
|
"utf-8"
|
|
)
|
|
names = ("route", "settings", "pipes", "alignment")
|
|
at = [re.search(r'step\(\s*"' + name + '"', server).start() for name in names]
|
|
assert at == sorted(at)
|
|
|
|
|
|
def test_노선_바꾼_저장에도_settings_기본_창구(out):
|
|
assert out["routeSaveDone"] is True
|
|
# 한 본문에 노선 · settings 기본 조각 · 화면 몫(own) · 화면 횡단 조각 아님 · 다 저장한 뒤 정리
|
|
assert out["routeSaveCalls"] == ["post:route:settings", "settings:p1"]
|
|
|
|
|
|
def test_초안_내보내기(out):
|
|
assert out["putsWithoutDraft"] == 0
|
|
assert out["boom"] == "서버 500" and out["draftAfterBoom"] == out["draft"]
|
|
assert out["draftAfterOk"] is None and out["puts"] == 2
|
|
params = out["latestAfterOk"]["route_params"]
|
|
assert params["options"] == {
|
|
"grade_class": "trunk",
|
|
"design_speed_kph": 30,
|
|
"terrain_type": "normal",
|
|
}
|
|
assert params["max_grade_pct"] is None and params["min_vertical_radius_m"] == 300
|
|
assert params["station_display"] == {"station": 12, "cumulative": 240}
|
|
|
|
|
|
def test_초안_되살림_값(out):
|
|
restore = out["restoreFromDraft"]
|
|
assert restore["designSpeed"] == 30 and restore["maxGradePct"] is None
|
|
assert restore["stationDisplay"] == {"station": 12, "cumulative": 240}
|
|
|
|
|
|
def test_시종점_조정은_끝_측점_계획고_편집(out):
|
|
assert out["endsEmpty"] == {"start": 0, "end": 0}
|
|
assert out["endsRaised"] == {"start": 1.25, "end": 0}
|
|
assert out["endsBoth"] == {"start": 1.25, "end": -0.5}
|
|
assert out["planStart"] == pytest.approx(101.25) and out["planEnd"] == pytest.approx(107.5)
|
|
assert out["backToZero"]["start"] == 0
|
|
|
|
|
|
def test_시작_측점은_열쇠_다른_페이지에_안_섞임():
|
|
root = HERE.parents[1]
|
|
hits = subprocess.run(
|
|
["git", "grep", "-l", "station_display", "--", "*.py", "*.ts"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=root,
|
|
encoding="utf-8",
|
|
).stdout.split()
|
|
allowed = {
|
|
"B03_FileInput/B03_FileInput_Service_Chain.py", # E1 옮김
|
|
"B05_Profile/B05_Profile_Api_Fetch.ts",
|
|
"B05_Profile/B05_Profile_Api_PageSettings.ts",
|
|
"B05_Profile/B05_Profile_PageSettings.py",
|
|
"B05_Profile/B05_Profile_Router.py",
|
|
"B05_Profile/B05_Profile_Schema.py",
|
|
"B05_Profile/B05_Profile_UI_Page_Latest.ts",
|
|
}
|
|
assert {hit for hit in hits if not hit.startswith("resources/")} <= allowed
|
|
|
|
|
|
def test_저장_정책을_다시_읽어도_평면_R_유지():
|
|
from B05_Profile.B05_Profile_Engine_Grade_Alignment import AlignmentPolicy
|
|
|
|
policy = AlignmentPolicy.from_dict({"min_plan_radius_m": 30, "max_grade_pct": 8})
|
|
assert policy.min_plan_radius_m == 30 and policy.hairpin_min_radius_m == 10
|
|
assert AlignmentPolicy.from_dict({}).min_plan_radius_m == 0 # 옛 정책 = 판정 안 함 그대로
|