From f48be66c9a02304e14042993d74b382cd201725e Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sun, 1 Nov 2026 11:50:22 +0900 Subject: [PATCH] =?UTF-8?q?fix(637-4):=20[=EC=B4=88=EA=B8=B0=ED=99=94]=20?= =?UTF-8?q?=EB=95=8C=20=EC=84=A4=EA=B3=84=20=EC=A7=80=ED=98=95=EB=8F=84=20?= =?UTF-8?q?=EC=B4=88=EA=B8=B0=EA=B0=92=EC=9C=BC=EB=A1=9C=20=E2=80=94=20?= =?UTF-8?q?=EC=B4=88=EA=B8=B0=EA=B0=92=EC=97=90=20stage=201=20=EC=A7=80?= =?UTF-8?q?=ED=98=95=20=EA=B3=A0=EB=A6=84=20=EB=8B=B4=EC=9D=8C=20=C2=B7=20?= =?UTF-8?q?=EB=90=98=EB=8F=8C=EB=A6=B4=20=EB=95=8C=20stage=201=20=ED=99=95?= =?UTF-8?q?=EC=A0=95=EA=B0=92=20=C2=B7=20CONFIRMED=20=EB=AA=A8=EB=8D=B8(?= =?UTF-8?q?=EB=85=B8=EC=84=A0=20=ED=96=89=EC=9D=B4=20=EA=B0=80=EB=A6=AC?= =?UTF-8?q?=ED=82=A4=EB=8A=94=20=EA=B2=83)=20=ED=95=A8=EA=BB=98=20=C2=B7?= =?UTF-8?q?=20=EC=98=9B=20=EC=B4=88=EA=B8=B0=EA=B0=92=20=3D=20=EC=A7=80?= =?UTF-8?q?=ED=98=95=20=EA=B7=B8=EB=8C=80=EB=A1=9C=20+=20=EB=A1=9C?= =?UTF-8?q?=EA=B7=B8=20=ED=95=9C=20=EC=A4=84(=EC=9A=94=EA=B5=AC=20546)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 재현(검증_sub4 B · 라이다 → 수치지형도 → [초기화]): stage 1 = 수치지형도 · 노선 행 = 라이다 모델 · 횡단 지반선 대 stage 1 지형 평균 3.78 m · 최대 13.67 m - 고친 뒤 같은 길(API · ORCA 칩 → [초기화]): 세 자리 라이다 · 어긋남 0 - las_selection · 등고선 간격 등 지형 밖 칸 안 건드림 · stage 1 행 없으면 안 씀 - 시험 test_637_4_reset_terrain 새로 Co-Authored-By: Claude Opus 5.5 --- common_util/common_util_initial_snapshot.py | 45 ++++++++ resources/tester/test_637_4_reset_terrain.py | 111 +++++++++++++++++++ 2 files changed, 156 insertions(+) create mode 100644 resources/tester/test_637_4_reset_terrain.py diff --git a/common_util/common_util_initial_snapshot.py b/common_util/common_util_initial_snapshot.py index fcaa00f55..b6ee4fe88 100644 --- a/common_util/common_util_initial_snapshot.py +++ b/common_util/common_util_initial_snapshot.py @@ -11,12 +11,15 @@ CLAUDE.md 5장(조작·데이터 흐름 정책)의 **초기값** 층이다. 자 import csv import json +import logging import shutil from pathlib import Path from typing import Any import aiomysql +logger = logging.getLogger(__name__) + # 스냅샷 폴더는 워크플로우 단계가 아니므로 PROJECT_STORAGE_LAYOUT_V2에 넣지 않는다. SNAPSHOT_DIRNAME = "initial_snapshot" _DB_DUMP_NAME = "db.json" @@ -50,6 +53,9 @@ _LEGACY_DRAINAGE_TREE = "B04_PreProcess__drainage__edits" # 번호 없는 이름으로 두었다가 복원 때 새 번호로 되돌린다. _CORRIDOR_NAME = "initial_corridor.json" +# 설계 지형 고름 — stage 1 확정값 중 지형 칸(PLAN 637-4 · 요구 546) +_TERRAIN_KEYS = ("source_filter", "method", "smooth") + # `routes.id`를 참조하는 표는 스키마상 이 넷이 전부다(001_create_schema.sql:539~556). _CHILD_TABLES = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections") @@ -220,6 +226,8 @@ async def save_initial_snapshot( dump[table] = _json_safe(await _dump_rows(connection, table, route_id)) # stage 2 params(제어점 · 페이지 설정)도 — [초기화]가 함께 되돌림(448 E3) dump["stage2_params"] = await _stage2_params(connection, str(route["project_id"])) + # 설계 지형 고름도 — 초기값은 그 지형으로 센 값이라 [초기화]가 함께 되돌림(637-4) + dump["stage1_terrain"] = await _stage1_terrain(connection, str(route["project_id"])) (target / _DB_DUMP_NAME).write_text(json.dumps(dump, ensure_ascii=False), encoding="utf-8") @@ -325,9 +333,46 @@ async def restore_initial_snapshot( child["project_id"] = project_id await _insert_row(connection, table, child) await _restore_stage2_params(connection, project_id, dump) + await _restore_stage1_terrain(connection, project_id, dump, route.get("surface_model_id")) return new_id +async def _stage1_terrain(connection: aiomysql.Connection, project_id: str) -> dict[str, Any]: + from common_util.common_util_surface_confirmation import get_surface_confirmation_params + + params = await get_surface_confirmation_params(connection, project_id) + return {key: params[key] for key in _TERRAIN_KEYS} + + +async def _restore_stage1_terrain( + connection: aiomysql.Connection, project_id: str, dump: dict[str, Any], model_id: Any +) -> None: + """뜬 설계 지형으로 stage 1 확정값 · CONFIRMED 모델(노선 행이 가리키는 것)을 되돌림(637-4). + + 옛 초기값(지형 칸 없음)은 지형 그대로 · 라이다 적어 둔 값(`las_selection`)은 안 건드림. + """ + terrain = dump.get("stage1_terrain") + if not isinstance(terrain, dict): + logger.info( + "[초기화] 옛 초기값 — 설계 지형 칸 없음 · 지형 그대로: project_id=%s", project_id + ) + return + from B04_PreProcess.B04_PreProcess_Repository import confirm_surface_model + from common_util.common_util_surface_confirmation import ( + get_surface_confirmation_params, + merge_surface_confirmation_params, + ) + + params = await get_surface_confirmation_params(connection, project_id) + try: + await merge_surface_confirmation_params(connection, project_id, {**params, **terrain}) + except LookupError: # stage 1 행 없음 — 되돌릴 자리 없음(아무것도 안 씀) + logger.warning("[초기화] stage 1 행 없음 · 설계 지형 그대로: project_id=%s", project_id) + return + if model_id is not None: + await confirm_surface_model(connection, project_id, int(model_id)) + + async def _stage2_params(connection: aiomysql.Connection, project_id: str) -> Any: async with connection.cursor() as cursor: await cursor.execute( diff --git a/resources/tester/test_637_4_reset_terrain.py b/resources/tester/test_637_4_reset_terrain.py new file mode 100644 index 000000000..aaae2145a --- /dev/null +++ b/resources/tester/test_637_4_reset_terrain.py @@ -0,0 +1,111 @@ +"""PLAN 637-4 — 초기값에 설계 지형 고름을 담고 [초기화]가 함께 되돌림(요구 546). + +지형 칩으로 설계 지형을 바꾼 뒤 [초기화] = 옛 지형으로 센 횡단 · 노선 행이 돌아오는데 +stage 1 확정값 · CONFIRMED 모델은 새것 그대로였음. 가짜 커넥션으로 뜨기 · 되돌리기 규약만 봄. +""" + +import asyncio +import json +import logging + +from common_util.common_util_initial_snapshot import ( + restore_initial_snapshot, + save_initial_snapshot, + snapshot_dir, +) + +LAS = {"source_filter": "classification", "method": "dtm", "smooth": True} +STASH = {"source_filter": "classification", "method": "dtm", "smooth": True} + + +class _Cursor: + def __init__(self, db): + self.db = db + self.rows: list = [] + self.rowcount = 0 + self.lastrowid = 0 + + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + return False + + async def execute(self, sql, params=None): + text = " ".join(sql.split()) + head = text.split()[0].upper() + self.rows = [] + if head == "INSERT": + self.db["next_id"] += 1 + self.lastrowid = self.db["next_id"] + elif head == "UPDATE": + self.db["updates"].append((text, params)) + if "project_workflow_stages" in text and "stage_no = 1" in text: + self.db["stage1"] = json.loads(params[0]) + self.rowcount = 1 + elif "FROM routes" in text: + self.rows = [{"id": 7, "project_id": "p1", "surface_model_id": 642}] + elif "stage_no = 1" in text: + self.rows = [{"params": json.dumps(self.db["stage1"])}] + + async def fetchone(self): + return self.rows[0] if self.rows else None + + async def fetchall(self): + return self.rows + + +class _Connection: + def __init__(self, db): + self.db = db + + def cursor(self, *_args, **_kwargs): + return _Cursor(self.db) + + +def _db(stage1): + return {"stage1": dict(stage1), "updates": [], "next_id": 100} + + +def test_뜰_때_설계_지형_고름을_담음(tmp_path): + db = _db({**LAS, "contour_interval_m": 1.0, "input_file_id": "9"}) + asyncio.run(save_initial_snapshot(_Connection(db), tmp_path, 7)) + dump = json.loads((snapshot_dir(tmp_path) / "db.json").read_text(encoding="utf-8")) + assert dump["stage1_terrain"] == LAS + + +def _write_dump(root, **extra): + folder = snapshot_dir(root) + folder.mkdir(parents=True) + dump = {"routes": [{"id": 7, "project_id": "p1", "surface_model_id": 642}], **extra} + (folder / "db.json").write_text(json.dumps(dump), encoding="utf-8") + + +def test_초기화가_설계_지형과_확정_모델을_되돌림(tmp_path): + _write_dump(tmp_path, stage1_terrain=LAS) + # 칩으로 수치지형도로 바꾼 뒤 — 라이다 값은 las_selection 에 적혀 있음 + sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True} + db = _db({**sheet, "contour_interval_m": 2.0, "las_selection": STASH, "input_file_id": "9"}) + new_id = asyncio.run(restore_initial_snapshot(_Connection(db), tmp_path, "p1")) + + assert new_id == 101 + assert {key: db["stage1"][key] for key in LAS} == LAS + # 지형 밖 칸 · 라이다 적어 둔 값은 그대로 + assert db["stage1"]["contour_interval_m"] == 2.0 + assert db["stage1"]["las_selection"] == STASH + assert db["stage1"]["input_file_id"] == "9" + # CONFIRMED = 노선 행이 가리키는 모델(642) + confirmed = [p for sql, p in db["updates"] if "SET status = 'CONFIRMED'" in sql] + assert confirmed == [(642, "p1")] + + +def test_옛_초기값은_지형_그대로_로그_한_줄(tmp_path, caplog): + _write_dump(tmp_path) # stage1_terrain 칸 없음 + sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True} + db = _db(sheet) + with caplog.at_level(logging.INFO, logger="common_util.common_util_initial_snapshot"): + asyncio.run(restore_initial_snapshot(_Connection(db), tmp_path, "p1")) + + assert db["stage1"] == sheet + assert not [sql for sql, _ in db["updates"] if "surface_models" in sql] + assert sum("옛 초기값" in r.getMessage() for r in caplog.records) == 1