fix(505-1 덧 · 505-2 5d B05): 추천 접근 = 엔진 지문 달라도 저장본 읽음 · 모델 없는 sheet 고르기 숨김 · 409 사유
- 추천 저장본: 프로젝트 접근(화면 기본 갈래)은 같은 판 · 같은 입력이면 그 파일 소스 지문으로 키를 다시 내 찾음(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 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -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():
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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"]
|
||||
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user