eomsangdonandClaude Opus 5.5
52553a6c6a
- followed_pipes 에서 새로 더한 관(짝 -1)만 _apply_new_defaults · 옛 빈 관 · 사용자 관 그대로 - 시험 test_494_2(옛 코드 빨강) · test_283 새 교차 관 옵션 기대 = 기본값 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
390 lines
17 KiB
Python
390 lines
17 KiB
Python
# -*- 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()
|