# -*- coding: utf-8 -*- """493-1 · 요구 289 · 290 — 손 고침 [저장] = 보이는 화면 그대로 · 추천 [적용] = 초기화 + 초기값 다시 뜸. ① 교점 끌기 [저장](손 고침) — 초기화 아님 · 초기값 그대로 · 체인 = 구조물 · 추가 측점 · 사용자 관 (옵션) 그대로 따라감 · 새 세류선 교차 관만 더함 · `route_reset` 칸 없음 ② 추천 [적용] — 초기화(균형 종단 갈래도) · 사용자 구조물 · 추가 측점 · 관 버림 · `route_reset` ③ 스냅 · 측점 간격만 — 초기화 아님 · 초기값 그대로 ④ 검증_sub8 자료(노선 522 → 628 · 관 p35100) — 손 고침 = 관 p35100(옵션) 지킴 + 교차 388 더함 · 추천 [적용] = 교차 388 하나 ⑤ 한 본문 손 고침 — 노선 차례 뒤 보낸 시설 · 넘김 차례 그대로 씀 ⑥ 추천 [적용] 실패 — 초기값 사본(`initial_snapshot`)까지 되돌림 ⑦ 떠 두기 중 뒤 미리 셈이 파일을 지움(`_derived` 원자 쓰기 · 정리) — 건너뜀(옛 = 500 FileNotFoundError) 492 코드 = ① ④(손 고침) ⑤ 실패 · 옛 떠 두기 = ⑦ 실패 · 새 교차 관 = B04 새 관 기본값(494-2). 체인 · API 는 346-3 가짜 세계(`chain_world` · `api_world`)를 빌려 씀 — 파일은 임시 폴더만. ./venv/Scripts/python.exe -m pytest resources/tester/test_283_route_change_reset.py -q """ from __future__ import annotations import asyncio import json import sys from pathlib import Path from types import SimpleNamespace from uuid import uuid4 import pytest HERE = Path(__file__).resolve().parent ROOT = HERE.parents[1] for path in (ROOT, HERE): if str(path) not in sys.path: sys.path.insert(0, str(path)) import test_346_3_route_save as base # noqa: E402 from test_346_3_route_save import _async, api_world, chain_world # noqa: E402, F401 import B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze as analyze # noqa: E402 import B05_Profile.B05_Profile_Route_Save as save # noqa: E402 import B05_Profile.B05_Profile_Router_Replan as replan # noqa: E402 import B06_Section.B06_Section_Repository as b06_repo # noqa: E402 from B04_PreProcess.B04_PreProcess_Router_Basins import _apply_new_defaults # noqa: E402 from B05_Profile.B05_Profile_Plus_Stations import load_plus_stations, save_plus_stations # noqa: E402 from B05_Profile.B05_Profile_Structures_Repository import load_structures # noqa: E402 from common_util.common_util_drainage_pipes import PipePoint # noqa: E402 from common_util.common_util_json import atomic_write_json # noqa: E402 FIXTURE = HERE / "fixtures" / "req283_sub8_route.json" WALL = { "structure_id": "wall", "type_id": "retaining_wall", "placement": "interval", "chainage_m": 250.0, "start_m": 245.0, "end_m": 255.0, "options": {"wall_form": "콘크리트", "length_m": 10, "height_m": 2}, "memo": "", "placement_source": "manual", "status": "draft", "revision": 0, "geometry": None, "summary_item": None, "group_items": None, "x": 250.0, "y": 0.0, } # 새 노선(NEW) 100 ~ 200 m 토막 위 세류선 교차 — 옛 노선 밖 CROSSING = SimpleNamespace(chainage_m=158.31, x=150.0, y=30.0) def _new_pipe_options() -> dict: """새 교차 관 옵션 = B04 유역 길의 새 관 기본값(494-2 · 54-39).""" pipe = PipePoint(chainage_m=0.0, source="stream") _apply_new_defaults([pipe]) return pipe.options def _side(xy, x: float, y: float) -> float: from shapely.geometry import LineString, Point return LineString(xy).distance(Point(x, y)) def _seed(world, pipe_options: dict | None = None) -> None: """사용자 시설 — 관(옵션) · 구조물 · 추가 측점 · 옛 횡단 손 칸.""" world.pipes_file.write_text( json.dumps( { "route_signature": "old", "points": [ { "chainage_m": 50.0, "x": 50.0, "y": 0.0, "source": "user", "options": pipe_options or {"pipe_kind": "흄관"}, } ], } ), "utf-8", ) atomic_write_json( world.root / "B05_Profile" / "route" / "structures.json", {"revision": 1, "structures": [WALL], "definition_version": None}, ) save_plus_stations(world.root, [350], base.OLD) @pytest.fixture def crossings(monkeypatch): """세류선 교차 = 고정 목록(도엽 없이).""" found = [CROSSING] monkeypatch.setattr(analyze, "_base_pipes", lambda vertices, streams: list(found)) return found @pytest.fixture def cross_seen(monkeypatch): seen: list = [] async def carry(*args): seen.append(args[5]) # 옛 횡단 설계 — 넘길 손 입력 return 0 monkeypatch.setattr(save, "_carry_cross", carry) monkeypatch.setattr( b06_repo, "get_cross_section_designs", _async([{"chainage_m": 50.0, "design": {"inlet_basin_length_m": 3}}]), ) return seen def _assert_reset(world, result: dict, cross_seen: list) -> None: assert "error" not in result, result assert result["route_reset"] == { "reset": True, "dropped": {"structures": 1, "pipes": 1, "plus_stations": 1}, } assert load_structures(str(world.root))[1] == [] assert load_plus_stations(world.root) == [] points = json.loads(world.pipes_file.read_text("utf-8"))["points"] assert len(points) == 1, points # 옛 사용자 관 버림 · 새 교차 관만 assert points[0]["source"] == "stream" and points[0]["options"] == _new_pipe_options() assert _side(base.NEW, points[0]["x"], points[0]["y"]) < 1e-6 assert "profile" not in world.order # 계획고 옮김 없음 = 1차 계획선 assert cross_seen == [[]] # 횡단 손 입력 안 넘김 # ── ① 교점 끌기 [저장](손 고침) ───────────────────────────────────────────── def _replan_world(monkeypatch, old_line, new_line) -> list: calls: list = [] async def recompute( project_id, root, stored, balanced=None, terrain=None, lines=None, initial=False ): calls.append(("recompute", initial, balanced is not None)) return {"route_id": 9, "total_length_m": 10.0} monkeypatch.setattr(replan, "_recompute", recompute) monkeypatch.setattr(replan, "_planned_line", lambda root: old_line) monkeypatch.setattr(replan, "_vertices_of", lambda path: new_line) return calls def test_교점_끌기_저장은_초기화_아님_초기값_그대로(api_world, monkeypatch): calls = _replan_world(monkeypatch, [[0.0, 0.0], [4.0, 4.0]], [[0.0, 0.0], [25.0, 5.0]]) result = asyncio.run(replan.replan_route(uuid4(), base._body(balanced_profile=True))) assert calls == [("recompute", False, False)] # 균형 종단 · 초기화는 추천 [적용]만 assert result["initial_refreshed"] is False and api_world == [] def test_교점_끌기_체인은_시설_따라감(chain_world, crossings, cross_seen): _seed(chain_world) result = base._run(chain_world) assert "error" not in result, result assert "route_reset" not in result assert len(load_structures(str(chain_world.root))[1]) == 1 assert len(load_plus_stations(chain_world.root)) == 1 points = json.loads(chain_world.pipes_file.read_text("utf-8"))["points"] assert len(points) == 2, points # 사용자 관 그대로 + 새 교차 관 assert any(p.get("options") == {"pipe_kind": "흄관"} for p in points) assert result["pipes"]["crossing"] == 1 assert "profile" in chain_world.order # 계획고 옮김 assert cross_seen != [[]] # 횡단 손 입력 넘김 # ── ② 추천 [적용] ───────────────────────────────────────────────────────────── def test_추천_적용은_초기화(api_world, chain_world, crossings, cross_seen, monkeypatch): import B05_Profile.B05_Profile_Router_RouteInitial as route_initial same = [[0.0, 0.0], [5.0, 5.0]] calls = _replan_world(monkeypatch, same, same) result = asyncio.run( replan.replan_route(uuid4(), base._body(as_initial=True, balanced_profile=True)) ) assert calls == [("recompute", True, True)] and result["initial_refreshed"] is True _seed(chain_world) monkeypatch.setattr(route_initial, "apply_profile_target", _async(None)) chained = base._run(chain_world, initial=True, profile_target=([0.0, 456.0], [1.0, 2.0])) _assert_reset(chain_world, chained, cross_seen) def test_추천_적용_체인은_사용자_시설_버림(chain_world, crossings, cross_seen): _seed(chain_world) result = base._run(chain_world, initial=True) _assert_reset(chain_world, result, cross_seen) assert result["pipes"]["crossing"] == 1 # ── ③ 스냅 · 간격만 ─────────────────────────────────────────────────────────── def test_스냅만_바꾼_저장은_초기화_아님(api_world, monkeypatch): same = [[0.0, 0.0], [5.0, 5.0]] calls = _replan_world(monkeypatch, same, [list(p) for p in same]) body = base._body(station_snap=True, station_snap_step=10.0) result = asyncio.run(replan.replan_route(uuid4(), body)) assert calls == [("recompute", False, False)] assert result["initial_refreshed"] is False and api_world == [] # ── ④ 검증_sub8 자료 ───────────────────────────────────────────────────────── @pytest.fixture def sub8(monkeypatch): data = json.loads(FIXTURE.read_text("utf-8")) monkeypatch.setattr(base, "OLD", [tuple(p) for p in data["old_xy"]]) monkeypatch.setattr(base, "NEW", [tuple(p) for p in data["new_xy"]]) found = [SimpleNamespace(**c) for c in data["crossings"]] monkeypatch.setattr(analyze, "_base_pipes", lambda vertices, streams: list(found)) return data def _sub8_pipes(world, data) -> None: world.pipes_file.write_text( json.dumps({"route_signature": "old", "points": [data["old_pipe"]]}), "utf-8" ) def test_검증_sub8_손_고침은_관_지킴_교차_더함(sub8, chain_world, cross_seen): _sub8_pipes(chain_world, sub8) result = base._run(chain_world) assert "error" not in result, result points = json.loads(chain_world.pipes_file.read_text("utf-8"))["points"] assert len(points) == 2, points kept = [p for p in points if p.get("id") == "p35100"] assert kept and kept[0]["options"] == sub8["old_pipe"]["options"] # 화면 그대로 assert any(round(p["chainage_m"]) == 388 for p in points) # 새 교차 def test_검증_sub8_추천_적용은_교차_하나(sub8, chain_world, cross_seen): _sub8_pipes(chain_world, sub8) result = base._run(chain_world, initial=True) assert "error" not in result, result points = json.loads(chain_world.pipes_file.read_text("utf-8"))["points"] assert [round(p["chainage_m"]) for p in points] == [388] assert _side(sub8["new_xy"], points[0]["x"], points[0]["y"]) < 0.01 assert points[0]["options"] == _new_pipe_options() # 494-2 새 교차 관 = 기본값 assert result["route_reset"]["dropped"]["pipes"] == 1 # ── ⑤ 한 본문 손 고침 ───────────────────────────────────────────────────────── def test_한_본문_손_고침은_보낸_칸을_씀(monkeypatch): import B04_PreProcess.B04_PreProcess_Router_Basins as basins import B05_Profile.B05_Profile_Router as b05_router import B05_Profile.B05_Profile_Router_Lifecycle as lifecycle import B05_Profile.B05_Profile_Structures_Router as structures_router import B06_Section.B06_Section_Router_Confirm as confirm import B06_Section.B06_Section_Router_DesignSave as design_save import B06_Section.B06_Section_Router_DesignSave_Route as moving import B06_Section.B06_Section_Router_Stations as stations called: list[str] = [] def record(name, value=None): async def fake(*args, **kwargs): called.append(name) return {} if value is None else value return fake monkeypatch.setattr(replan, "replan_route", record("route", {"route_id": 9})) monkeypatch.setattr(moving, "saved_facilities", record("before")) monkeypatch.setattr(moving, "rebase_drafts", record("rebase")) monkeypatch.setattr(moving, "carry_cross", record("carry", {"patches": [], "outside": 0})) monkeypatch.setattr(lifecycle, "save_page_settings", record("settings")) monkeypatch.setattr(basins, "put_pipe_points", record("pipes")) monkeypatch.setattr(basins, "get_pipe_points", record("culvert", {"pipe_points": []})) monkeypatch.setattr(structures_router, "write_structures", record("structures")) monkeypatch.setattr(stations, "put_plus_stations", record("plus")) monkeypatch.setattr(b05_router, "save_profile_alignment", record("alignment")) monkeypatch.setattr(confirm, "save_sections", record("cross")) request = design_save.DesignSaveRequest.model_construct( route_id=7, confirm=False, settings=SimpleNamespace(), pipes={"points": [{"chainage_m": 50.0}]}, structures=SimpleNamespace(), plus_stations=SimpleNamespace(), alignment=SimpleNamespace(model_copy=lambda update: SimpleNamespace(**update)), culvert_options=None, uphill=None, cross=None, route=SimpleNamespace(), carry=SimpleNamespace(), ) out = asyncio.run(design_save._run_steps(uuid4(), request, None)) assert called == [ "before", "route", "rebase", "settings", "pipes", "structures", "plus", "alignment", "carry", "cross", ], called assert "route_reset" not in out["route"] # ── ⑥ 추천 [적용] 실패 ───────────────────────────────────────────────────────── def test_추천_적용_실패면_초기값_사본까지_되돌림(tmp_path, monkeypatch): import B06_Section.B06_Section_Router_DesignSave as design_save import common_util.common_util_save_guard as guard from fastapi.responses import JSONResponse copy = tmp_path / "initial_snapshot" / "B04_PreProcess__drainage" / "edits" copy.mkdir(parents=True) (copy / "pipe_points.json").write_text("old", encoding="utf-8") async def fail(project_id, body): (copy / "pipe_points.json").write_text("new", encoding="utf-8") # 초기값 다시 뜸 return JSONResponse(status_code=409, content={"status": "error", "message": "x"}) monkeypatch.setattr(replan, "_project_paths", _async((tmp_path, "p"))) monkeypatch.setattr(replan, "_save_route", fail) monkeypatch.setattr(replan, "get_db_pool", lambda: base._Pool([])) monkeypatch.setattr(design_save, "get_db_pool", lambda: base._Pool([])) monkeypatch.setattr(guard, "snapshot_rows", _async({})) monkeypatch.setattr(guard, "restore_rows", _async(None)) response = asyncio.run(replan.replan_route(uuid4(), base._body(as_initial=True))) assert response.status_code == 409 assert (copy / "pipe_points.json").read_text(encoding="utf-8") == "old" # ── ⑦ 떠 두기 중 파일 사라짐 ─────────────────────────────────────────────────── def test_떠_두기_중_지워진_파일은_건너뜀(tmp_path, monkeypatch): import common_util.common_util_save_guard as guard derived = tmp_path / "_derived" / "route_initial" derived.mkdir(parents=True) (derived / "a.json").write_text("a", encoding="utf-8") ghost = derived / "b.json.tmp" # 목록 뒤 · 사본 전에 원자 쓰기가 옮김 · 정리가 지움 real_files = guard._files def listed(root, scope): return {**real_files(root, scope), "_derived/route_initial/b.json.tmp": ghost} monkeypatch.setattr(guard, "_files", listed) snap = guard.snapshot_storage(tmp_path, ("_derived",)) monkeypatch.setattr(guard, "_files", real_files) try: assert set(snap.stamps) == {"_derived/route_initial/a.json"} (derived / "a.json").write_text("changed", encoding="utf-8") (derived / "c.json").write_text("new", encoding="utf-8") snap.restore() assert sorted(p.name for p in derived.iterdir()) == ["a.json"] assert (derived / "a.json").read_text(encoding="utf-8") == "a" finally: snap.discard()