From 7ae7515773d637be78793cd38d66145d5b8b719f Mon Sep 17 00:00:00 2001 From: umsangdon Date: Wed, 21 Oct 2026 13:28:06 +0900 Subject: [PATCH] =?UTF-8?q?fix(505-1=20=EB=8D=A7=20=C2=B7=20505-2=205d=20B?= =?UTF-8?q?05):=20=EC=B6=94=EC=B2=9C=20=EC=A0=91=EA=B7=BC=20=3D=20?= =?UTF-8?q?=EC=97=94=EC=A7=84=20=EC=A7=80=EB=AC=B8=20=EB=8B=AC=EB=9D=BC?= =?UTF-8?q?=EB=8F=84=20=EC=A0=80=EC=9E=A5=EB=B3=B8=20=EC=9D=BD=EC=9D=8C=20?= =?UTF-8?q?=C2=B7=20=EB=AA=A8=EB=8D=B8=20=EC=97=86=EB=8A=94=20sheet=20?= =?UTF-8?q?=EA=B3=A0=EB=A5=B4=EA=B8=B0=20=EC=88=A8=EA=B9=80=20=C2=B7=20409?= =?UTF-8?q?=20=EC=82=AC=EC=9C=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 추천 저장본: 프로젝트 접근(화면 기본 갈래)은 같은 판 · 같은 입력이면 그 파일 소스 지문으로 키를 다시 내 찾음(read_stored_any) · 지문 견줌 · 다시 셈은 전처리 체인만 - 초기 종단 저장본: 접근은 코드 지문 빼고 입력만 견줌 · 파일에 코드 지문 칸 · 지문 칸 없는 옛 파일은 그대로 읽음 - 분석 지형 고르기 목록: 설계 지형 아닌 쪽은 DB 완료 모델도 있어야 둠(없으면 숨김) - 노선 [저장] 설계 지형 바꿈에 모델 없음(LookupError) = 되돌림 + 409 사유 한 줄 - 시험 test_309_3 둘 더함 · test_505_2_terrain_sheet 새로 · test_133_1 DB 모델 맞춤 Co-Authored-By: Claude Opus 5.5 --- .../B05_Profile_Engine_InitialProfile.py | 33 ++++-- .../B05_Profile_Engine_RouteInitial_Store.py | 35 +++++- B05_Profile/B05_Profile_Router_Replan.py | 8 +- .../B05_Profile_Router_RouteInitial.py | 19 +++- ...B05_Profile_Router_RouteInitial_Initial.py | 42 ++++---- B05_Profile/B05_Profile_Router_Terrain.py | 17 ++- .../tester/test_133_1_analysis_terrain.py | 7 ++ .../test_309_3_recommend_preprocess_only.py | 41 +++++++ resources/tester/test_505_2_terrain_sheet.py | 101 ++++++++++++++++++ 9 files changed, 264 insertions(+), 39 deletions(-) create mode 100644 resources/tester/test_505_2_terrain_sheet.py diff --git a/B05_Profile/B05_Profile_Engine_InitialProfile.py b/B05_Profile/B05_Profile_Engine_InitialProfile.py index 5e373e0f8..7ecc75a21 100644 --- a/B05_Profile/B05_Profile_Engine_InitialProfile.py +++ b/B05_Profile/B05_Profile_Engine_InitialProfile.py @@ -8,8 +8,8 @@ 「종단 반영」 켬이면 1 m 균형 종단을 꼭짓점에 얹음(`seat_base_pvi` → `rebuild_alignment_profile`). [적용]은 이 값을 받아 적지 않음 — 서버가 체인에서 같은 조각으로 다시 셈(CLAUDE.md 5). -저장본 = `B05_Profile/route_recommend/initial/<결과 키>.json` `{fingerprint, profile}` — 결과 정리 -(`*.json` glob)가 안 보는 하위 폴더 · 결과 짝 없는 파일은 쓸 때 지움 · 지문 다르면 다시 셈. +저장본 = `B05_Profile/route_recommend/initial/<결과 키>.json` `{fingerprint, source, profile}` — +결과 정리(`*.json` glob)가 안 보는 하위 폴더 · 결과 짝 없는 파일은 쓸 때 지움 · 지문 다르면 다시 셈. 응답 칸 `initial_profile`(화면 476-2) = `{stations_m, ground_m, plan_m, line: [[s, z]…], pvi_count, curve_count, straightened, max_grade_pct, curve_length_m, warnings, with_profile, seconds}` · 실패 = `{error}`. @@ -209,8 +209,11 @@ def stat_marks(paths: list[Path]) -> list[str]: return marks -def fingerprint(parts: dict[str, Any]) -> str: - text = json.dumps({**parts, "source": source_fingerprint()}, sort_keys=True, default=str) +def fingerprint(parts: dict[str, Any], source: str | None = None) -> str: + """입력 `parts` + 코드 지문(`source` 없으면 지금 코드).""" + text = json.dumps( + {**parts, "source": source or source_fingerprint()}, sort_keys=True, default=str + ) return hashlib.blake2b(text.encode(), digest_size=16).hexdigest() @@ -218,21 +221,35 @@ def _path(project_root: Path, key: str) -> Path: return Path(project_root) / INITIAL_DIR / f"{key}.json" -def read_initial(project_root: Path, key: str, mark: str) -> dict[str, Any] | None: - """저장본 — 없거나 지문 다르면 None.""" +def read_initial( + project_root: Path, key: str, mark: str, parts: dict[str, Any] | None = None +) -> dict[str, Any] | None: + """저장본 — 없거나 지문 다르면 None. `parts` 있음 = 프로젝트 접근(505-1 덧) — 코드 지문은 + 안 견주고 입력만(그 파일의 코드 지문으로 다시 냄) · 코드 지문 없는 옛 파일은 그대로 꺼냄.""" try: data = json.loads(_path(project_root, key).read_text(encoding="utf-8")) except (OSError, ValueError): return None + if not isinstance(data, dict): + return None + if data.get("fingerprint") == mark: + return data.get("profile") + if parts is None: + return None + source = data.get("source") return ( - data.get("profile") if isinstance(data, dict) and data.get("fingerprint") == mark else None + data.get("profile") + if source is None or fingerprint(parts, source) == data.get("fingerprint") + else None ) def write_initial(project_root: Path, key: str, mark: str, profile: dict[str, Any]) -> None: """한 결과에 하나(덮어씀) · 결과 저장본 짝 없는 파일 지움.""" path = _path(project_root, key) - atomic_write_json(path, {"fingerprint": mark, "profile": profile}) + atomic_write_json( + path, {"fingerprint": mark, "source": source_fingerprint(), "profile": profile} + ) results = path.parent.parent for other in path.parent.glob("*.json"): if not (results / other.name).exists(): diff --git a/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py b/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py index 760e2b416..5ded181e0 100644 --- a/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py +++ b/B05_Profile/B05_Profile_Engine_RouteInitial_Store.py @@ -11,7 +11,9 @@ 기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · 표준 횡단면 · `ALGO_VERSION` · 엔진 소스 지문. 관 지점은 뺀다 — 덮개 수는 꺼낼 때 다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 다른 판 · 지문 파일은 안 읽고 남기며 30일 지난 -것만 미리 셈 때 지운다(420-2 · storage 를 두 PC · 여러 워크트리가 같이 씀). +것만 미리 셈 때 지운다(420-2 · storage 를 두 PC · 여러 워크트리가 같이 씀). 프로젝트 접근(화면 +요청)은 같은 판 · 같은 입력이면 다른 소스 지문 저장본도 꺼냄(`read_stored_any` · 505-1 덧) — 지문 +견줌 · 다시 셈은 전처리 체인에서만. 기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1). """ @@ -25,7 +27,7 @@ import time from contextlib import asynccontextmanager from datetime import datetime from pathlib import Path -from typing import Any, AsyncIterator +from typing import Any, AsyncIterator, Callable from B05_Profile.B05_Profile_Engine_RouteInitial import ( DEFAULT_WEIGHTS, @@ -91,8 +93,11 @@ def request_fingerprint( with_profile: bool, design_speed: int, standard: dict[str, Any] | None = None, + source: str | None = None, ) -> tuple[str, str]: - """(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키.""" + """(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키. + + `source` = 그 엔진 소스 지문의 키(없으면 지금 코드 · `read_stored_any`).""" pool = _pool_key( expected, terrain, @@ -109,7 +114,7 @@ def request_fingerprint( int(design_speed), json.dumps(standard, sort_keys=True, default=str) if standard else None, ALGO_VERSION, - source_fingerprint(), + source or source_fingerprint(), ) key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest() return key, pool @@ -156,6 +161,28 @@ def read_stored(project_root: Path, key: str) -> dict[str, Any] | None: return data if _current(data) else None +def read_stored_any( + project_root: Path, pool: str, key_of: Callable[[str], str] +) -> tuple[str, dict[str, Any]] | None: + """(키, 저장본) — 같은 판 · 묶음이고 그 파일의 소스 지문으로 다시 낸 키(`key_of`)가 파일 이름과 + 같은 것 중 최근 것 · 없으면 None. 프로젝트 접근 전용(505-1 덧 · 엔진 고쳐도 안내 안 쏟아짐).""" + found: list[tuple[float, str, dict[str, Any]]] = [] + for path in store_dir(project_root).glob("*.json"): + try: + data = json.loads(path.read_text(encoding="utf-8")) + mtime = path.stat().st_mtime + except (OSError, ValueError): + continue + if not isinstance(data, dict) or data.get("version") != ALGO_VERSION: + continue + if data.get("pool") == pool and key_of(str(data.get("source"))) == path.stem: + found.append((mtime, path.stem, data)) + if not found: + return None + _, key, data = max(found, key=lambda item: item[0]) + return key, data + + def write_stored( project_root: Path, key: str, diff --git a/B05_Profile/B05_Profile_Router_Replan.py b/B05_Profile/B05_Profile_Router_Replan.py index 404b2f779..57f9320a7 100644 --- a/B05_Profile/B05_Profile_Router_Replan.py +++ b/B05_Profile/B05_Profile_Router_Replan.py @@ -63,6 +63,8 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/projects", tags=["B05 Route Design"]) _PROJECT_PATH_MISSING = {"status": "error", "message": "프로젝트 저장 경로를 찾을 수 없습니다."} +# 고른 분석 지형의 지표면 모델이 DB 에 없음(505-2 5d) +TERRAIN_MODEL_MISSING = "고른 분석 지형의 지표면 모델이 없습니다 — 전처리에서 만든 뒤 다시 하세요." async def _project_paths(project_id: UUID) -> tuple[Path, str] | None: @@ -336,7 +338,11 @@ async def _recompute( pool = get_db_pool() async with pool.acquire() as connection: await connection.begin() # 고른 지형 = 설계 지형 — 뒤 단계가 모두 이 확정값을 읽음 - switched = await switch_design_terrain(connection, project_id, terrain) + try: + switched = await switch_design_terrain(connection, project_id, terrain) + except LookupError: # 고른 지형의 DB 모델 없음 — 409 사유 한 줄(505-2 5d) + await connection.rollback() + return {"error": TERRAIN_MODEL_MISSING} await connection.commit() selection = await get_surface_confirmation_params(connection, str(project_id)) latest = await get_latest_route(connection, project_id) diff --git a/B05_Profile/B05_Profile_Router_RouteInitial.py b/B05_Profile/B05_Profile_Router_RouteInitial.py index b97f0137b..56682676f 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial.py @@ -49,7 +49,8 @@ async def compute_initial( 응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음). `recompute` = [재계산] — 묶음 잠금 안에서 이 묶음 저장본 · 메모리 묶음을 버리고 새로 셈(420). 미리 셈 갈래(기본 · 저장 제약 · 종단 켬 · 설계 지형)는 `precompute`(전처리 체인)만 셈 — 화면 - 요청은 저장본만 · 없으면 409 `NOT_PREPARED`(505-1). 사용자가 바꾼 값 · [재계산]은 그 자리 셈. + 요청은 저장본만(엔진 소스 지문이 달라도 · 505-1 덧) · 없으면 409 `NOT_PREPARED`(505-1). + 사용자가 바꾼 값 · [재계산]은 그 자리 셈. """ from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct from B05_Profile.B05_Profile_Engine_RouteInitial import ( @@ -67,6 +68,7 @@ async def compute_initial( is_default, pool_lock, read_stored, + read_stored_any, request_fingerprint, write_stored, ) @@ -154,9 +156,20 @@ async def compute_initial( forget_pool_on_standard(pool_key, inputs["standard"]) stored = await asyncio.to_thread(read_stored, project_root, key) if stored is None and not may_compute: - return JSONResponse( - status_code=409, content={"status": "error", "message": NOT_PREPARED} + # 프로젝트 접근 — 엔진 코드 판이 달라도 같은 입력 저장본을 꺼냄(505-1 덧) + found = await asyncio.to_thread( + read_stored_any, + project_root, + pool_key, + lambda source: request_fingerprint( + route, terrain, streams, **inputs, source=source + )[0], ) + if found is None: + return JSONResponse( + status_code=409, content={"status": "error", "message": NOT_PREPARED} + ) + key, stored = found if stored is not None: result = stored["result"] else: diff --git a/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py b/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py index 09b689127..85309c903 100644 --- a/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py +++ b/B05_Profile/B05_Profile_Router_RouteInitial_Initial.py @@ -94,27 +94,29 @@ async def initial_profile_of( crs = f"EPSG:{epsg}" if epsg is not None else None surface = {k: selection.get(k) for k in ("source_filter", "method", "smooth")} models = project_root / _MODELS_SUBDIR - mark = ip.fingerprint( - { - "key": key, - "nodes": result["nodes"], - "curves": result.get("curves"), - "radius": radius_m, - "surface": surface, - "models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))), - "streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]), - # 저장 노선 파일은 안 넣음(505-1) — 초기화(492) 셈은 노선 파일 · 관 · 구조물 · - # 추가 측점을 안 읽음(선 = 후보 노드) · 넣으면 노선 [저장]마다 다시 셈 - "options": asdict(options), - "grade": grade_options.as_dict() if grade_options else None, - "crs": crs, - "with_profile": request.with_profile, - "criteria": request.criteria.model_dump() if request.criteria else None, - "standard": await project_standard(project_id), - "road": await replan._road_settings(project_id), - } + parts = { + "key": key, + "nodes": result["nodes"], + "curves": result.get("curves"), + "radius": radius_m, + "surface": surface, + "models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))), + "streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]), + # 저장 노선 파일은 안 넣음(505-1) — 초기화(492) 셈은 노선 파일 · 관 · 구조물 · + # 추가 측점을 안 읽음(선 = 후보 노드) · 넣으면 노선 [저장]마다 다시 셈 + "options": asdict(options), + "grade": grade_options.as_dict() if grade_options else None, + "crs": crs, + "with_profile": request.with_profile, + "criteria": request.criteria.model_dump() if request.criteria else None, + "standard": await project_standard(project_id), + "road": await replan._road_settings(project_id), + } + mark = ip.fingerprint(parts) + # 프로젝트 접근(compute 거짓) = 코드 지문 안 견줌(505-1 덧) + stored = await asyncio.to_thread( + ip.read_initial, project_root, key, mark, None if compute else parts ) - stored = await asyncio.to_thread(ip.read_initial, project_root, key, mark) if stored is not None: return stored if not compute: diff --git a/B05_Profile/B05_Profile_Router_Terrain.py b/B05_Profile/B05_Profile_Router_Terrain.py index bbf439ac5..b17af5d7e 100644 --- a/B05_Profile/B05_Profile_Router_Terrain.py +++ b/B05_Profile/B05_Profile_Router_Terrain.py @@ -116,7 +116,11 @@ async def read_route_elevations(project_id: UUID, request: ElevationRequest) -> @router.get("/{project_id}/route/terrain-sources", response_model=None) async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse: - """편집 모달이 고를 수 있는 분석 지형 — 둘이면 고르기를 띄움 · 하나면 숨김.""" + """편집 모달이 고를 수 있는 분석 지형 — 둘이면 고르기를 띄움 · 하나면 숨김. + + 설계 지형 아닌 쪽은 DB 완료 모델이 있어야 둠 — [확인]이 그 모델로 바꾸므로(505-2 5d).""" + from B04_PreProcess.B04_PreProcess_Service import find_surface_model_for_selection + pool = get_db_pool() async with pool.acquire() as connection: stored = await get_project_storage_relative_path(connection, project_id) @@ -127,8 +131,15 @@ async def read_terrain_sources(project_id: UUID) -> dict | JSONResponse: ) params = await get_surface_confirmation_params(connection, str(project_id)) confirmed = await _confirmed_model(connection, project_id) - models_dir = Path(resolve_stored_project_path(stored)) / _MODELS_SUBDIR - sources = await asyncio.to_thread(analysis_terrain_sources, params, models_dir) + models_dir = Path(resolve_stored_project_path(stored)) / _MODELS_SUBDIR + sources = await asyncio.to_thread(analysis_terrain_sources, params, models_dir) + for source in [s for s in sources if s != design_terrain(params)]: + try: + await find_surface_model_for_selection( + connection, project_id, analysis_terrain_params(params, source) + ) + except LookupError: + sources.remove(source) return { "status": "success", "project_id": str(project_id), diff --git a/resources/tester/test_133_1_analysis_terrain.py b/resources/tester/test_133_1_analysis_terrain.py index 896998426..b278a66e1 100644 --- a/resources/tester/test_133_1_analysis_terrain.py +++ b/resources/tester/test_133_1_analysis_terrain.py @@ -94,6 +94,13 @@ def test_라이다_있는_프로젝트_두_지형_다_고름(tmp_path, monkeypat models = root / "B04_PreProcess" / "models" assert analysis_terrain_sources(LAS_PARAMS, models) == ["las", "sheet"] _patch_db(monkeypatch, root, LAS_PARAMS) + import B04_PreProcess.B04_PreProcess_Service as service + + async def models_in_db(*args): # sheet 고르기 = DB 완료 모델도 있어야(505-2 5d) + params = {"source_filter": SHEET_ANALYSIS_FILTER} + return [{"id": 2, "status": "COMPLETE", "model_type": "DTM", "generation_params": params}] + + monkeypatch.setattr(service, "list_surface_models", models_in_db) answer = asyncio.run(terrain_router.read_terrain_sources(PID)) assert answer["sources"] == ["las", "sheet"] diff --git a/resources/tester/test_309_3_recommend_preprocess_only.py b/resources/tester/test_309_3_recommend_preprocess_only.py index a8a859e65..e9cc4f9d5 100644 --- a/resources/tester/test_309_3_recommend_preprocess_only.py +++ b/resources/tester/test_309_3_recommend_preprocess_only.py @@ -139,3 +139,44 @@ def test_초기_종단_지문에_노선_파일_없음(gather, monkeypatch) -> No doc["properties"] = {**(doc.get("properties") or {}), "saved": "노선 [저장]"} route.write_text(json.dumps(doc), "utf-8") # 노선 [저장] = 노선 파일만 바뀜 assert gather["run"]("k1", False) == first + + +def test_엔진_지문_다른_저장본_접근은_읽음_체인은_다시_셈(wired, monkeypatch) -> None: # noqa: F811 + """505-1 덧 — 엔진 코드를 고쳐도(소스 지문 바뀜) 화면은 옛 저장본을 꺼냄 · 체인만 다시 셈.""" + import B05_Profile.B05_Profile_Engine_RouteInitial_Store as store + + asyncio.run(router.run_precompute(uuid4())) + assert wired.count == 5 + monkeypatch.setattr(store, "source_fingerprint", lambda: "엔진 고침") + for mode in DEFAULT_MODES: + assert _ask(False, mode=mode)["stored"] is True + assert wired.count == 5 # 접근 = 셈 0 · 안내 없음 + # 입력이 다르면(제약 바꿈 · 사용자 셈 갈래) 옛 지문을 끌어오지 않음 + assert _ask(False, mode="gentle", criteria={"min_radius_m": 15.0})["stored"] is False + assert wired.count == 6 + asyncio.run(router.run_precompute(uuid4())) # 체인 = 지문 견줌 · 다시 셈 + assert wired.count == 11 + sources = { + json.loads(p.read_text("utf-8"))["source"] for p in store_dir(wired.root).glob("*.json") + } + assert "엔진 고침" in sources and len(sources) == 2 # 옛 판은 30일 정리 몫 + + +def test_초기_종단_접근은_코드_지문_안_견줌(gather, monkeypatch) -> None: # noqa: F811 + first = gather["run"]() + assert "error" not in first, first + + def broken(*args, **kwargs): + raise RuntimeError("다시 셈") + + monkeypatch.setattr(ip, "initial_longitudinal", broken) + monkeypatch.setattr(ip, "source_fingerprint", lambda: "엔진 고침") + assert gather["run"]("k1", False) == first # 접근 = 그대로 읽음 + assert "다시 셈" in gather["run"]("k1", True)["error"] # 체인 = 다시 셈 + path = gather["root"] / ip.INITIAL_DIR / "k1.json" + data = json.loads(path.read_text("utf-8")) + path.write_text(json.dumps({k: v for k, v in data.items() if k != "source"}), "utf-8") + assert gather["run"]("k1", False) == first # 코드 지문 없는 옛 파일도 읽음 + path.write_text(json.dumps(data), "utf-8") + gather["state"]["stages"][0]["params"]["max_grade_pct"] = 7.5 # 입력이 바뀜 + assert gather["run"]("k1", False) == {"error": router.NOT_PREPARED} diff --git a/resources/tester/test_505_2_terrain_sheet.py b/resources/tester/test_505_2_terrain_sheet.py new file mode 100644 index 000000000..2464e7f5b --- /dev/null +++ b/resources/tester/test_505_2_terrain_sheet.py @@ -0,0 +1,101 @@ +"""PLAN 505-2 5d — 수치지형도 sheet 고르기 = DB 모델 있을 때만 · 없으면 [확인] 409 사유 한 줄. + + ./venv/Scripts/python.exe -m pytest resources/tester/test_505_2_terrain_sheet.py -q + +고르기 목록(`/route/terrain-sources`)은 npz 파일 + DB 완료 모델을 함께 봄 — 모델 없는 sheet 는 숨김 +(화면은 고르기가 숨으면 분석 지형을 안 실음). 바꿈(`switch_design_terrain`)이 모델을 못 찾으면 +`_recompute` 가 `{error}` → 노선 되돌리고 409(전엔 LookupError 가 저장 통째 되돌림). +""" + +from __future__ import annotations + +import asyncio +import sys +from contextlib import asynccontextmanager +from pathlib import Path +from uuid import uuid4 + +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +import B04_PreProcess.B04_PreProcess_Service as service # noqa: E402 +import B05_Profile.B05_Profile_Router_Replan as replan # noqa: E402 +import B05_Profile.B05_Profile_Router_Terrain as terrain # noqa: E402 +from common_util.common_util_surface_confirmation import SHEET_ANALYSIS_FILTER # noqa: E402 + +LAS = {"source_filter": "classification", "method": "dtm", "smooth": True} + + +class _Pool: + @asynccontextmanager + async def acquire(self): + yield _Connection() + + +class _Connection: + async def begin(self): + pass + + async def rollback(self): + _Connection.rolled = True + + async def commit(self): + raise AssertionError("모델 없으면 확정하지 않음") + + +def _async(value): + async def inner(*args, **kwargs): + return value + + return inner + + +def _model(source_filter: str) -> dict: + return { + "id": 7, + "status": "COMPLETE", + "model_type": "DTM", + "generation_params": {"source_filter": source_filter}, + } + + +@pytest.fixture +def sources(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + models = tmp_path / terrain._MODELS_SUBDIR + models.mkdir(parents=True) + (models / f"dtm_{SHEET_ANALYSIS_FILTER}.npz").write_bytes(b"") # sheet 파일은 있음 + monkeypatch.setattr(terrain, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(terrain, "get_project_storage_relative_path", _async("p")) + monkeypatch.setattr(terrain, "get_surface_confirmation_params", _async(dict(LAS))) + monkeypatch.setattr(terrain, "_confirmed_model", _async(None)) + monkeypatch.setattr(terrain, "resolve_stored_project_path", lambda stored: tmp_path) + monkeypatch.setattr(terrain, "_design_grid", lambda *args: None) + + def run(models_in_db: list[dict]) -> list[str]: + monkeypatch.setattr(service, "list_surface_models", _async(models_in_db)) + return asyncio.run(terrain.read_terrain_sources(uuid4()))["sources"] + + return run + + +def test_모델_없는_sheet_고르기_숨김(sources) -> None: + assert sources([_model("classification")]) == ["las"] + + +def test_모델_있는_sheet_고르기_보임(sources) -> None: + assert sources([_model(SHEET_ANALYSIS_FILTER)]) == ["las", "sheet"] + + +def test_모델_없으면_재계산_409_사유(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + import common_util.common_util_surface_confirmation as surface + + _Connection.rolled = False + monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(service, "get_surface_confirmation_params", _async(dict(LAS))) + monkeypatch.setattr(service, "list_surface_models", _async([_model("classification")])) + monkeypatch.setattr(surface, "get_surface_confirmation_params", _async(dict(LAS))) + got = asyncio.run(replan._recompute(uuid4(), tmp_path, "p", terrain="sheet")) + assert got == {"error": replan.TERRAIN_MODEL_MISSING} and _Connection.rolled