Aislo/resources/tester/test_283_route_change_reset.py

# -*- coding: utf-8 -*-
"""492-1 · 요구 283 · 284 · 285 — 노선 모양이 바뀐 [저장] · 추천 [적용] = 초기화 + 초기값 다시 뜸.

  ① 교점 끌기 [저장] — 판정(저장 · 새 계획 폴리라인) = 초기화 · 초기값 다시 뜸 · 체인 = 구조물 0 ·
     추가 측점 0 · 관 = 새 교차만(stream · 옵션 없음 · 옆 0 m) · 계획고 옮김 없음 · 횡단 손 칸 0
  ② 추천 [적용] — 같은 초기화(균형 종단 갈래도)
  ③ 스냅 · 측점 간격만(선 그대로) — 초기화 아님 · 시설 따라감 · 초기값 그대로
  ④ 검증_sub8 자료(노선 522 → 628 · 관 p35100) — 관 = 새 교차 388 하나(옛 = 388 + 394 · 34.44 m 옆)
  ⑤ 한 본문 노선 초기화 — 보낸 관 · 구조물 · 추가 측점 · 계획고 · 관 옵션 · 넘김 칸 안 씀
  ⑥ 한 본문 실패 — 초기값 사본(`initial_snapshot`)까지 되돌림
옛 코드(492 전) = ① ② ④ ⑤ ⑥ 실패.

체인 · 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 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_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 _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 not points[0].get("options")
    assert _side(base.NEW, points[0]["x"], points[0]["y"]) < 1e-6
    assert "profile" not in world.order  # 계획고 옮김 없음 = 1차 계획선
    assert cross_seen == [[]]  # 횡단 손 입력 안 넘김


# ── ① 교점 끌기 [저장] ─────────────────────────────────────────────────────────


def test_판정_선이_바뀌면_참_같으면_거짓():
    from B05_Profile.B05_Profile_Engine_Grade_Carry import route_shape_changed

    line = [(0.0, 0.0), (100.0, 0.0), (200.0, 0.0)]
    assert route_shape_changed(line, list(line)) is False
    assert (
        route_shape_changed(line, [(0.0, 0.0), (100.0, 0.0005), (200.0, 0.0)]) is False
    )  # 1 mm 안
    assert route_shape_changed(line, [(0.0, 0.0), (100.0, 20.0), (200.0, 0.0)]) is True
    assert route_shape_changed(line, [(0.0, 0.0), (200.0, 0.0)]) is True  # 교점 지움
    assert route_shape_changed(line, [*line, (300.0, 0.0)]) is True  # 교점 더함


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], [5.0, 5.0]])
    result = asyncio.run(replan.replan_route(uuid4(), base._body()))
    assert calls == [("recompute", True, False)]  # 균형 종단은 추천 [적용]만
    assert result["initial_refreshed"] is True and len(api_world) == 1


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, 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_스냅만_바꾼_저장은_초기화_아님(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 == []


def test_스냅만_체인은_시설_따라감(chain_world, crossings, cross_seen):
    _seed(chain_world)
    result = base._run(chain_world)
    assert result["route_reset"]["reset"] is False
    assert result["route_reset"]["dropped"] == {"structures": 0, "pipes": 0, "plus_stations": 0}
    assert len(load_structures(str(chain_world.root))[1]) == 1
    assert load_plus_stations(chain_world.root)
    points = json.loads(chain_world.pipes_file.read_text("utf-8"))["points"]
    assert any(p.get("options") == {"pipe_kind": "흄관"} for p in points)  # 사용자 관 그대로


# ── ④ 검증_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 test_검증_sub8_노선_522_628_관은_교차_하나(sub8, chain_world, cross_seen):
    chain_world.pipes_file.write_text(
        json.dumps({"route_signature": "old", "points": [sub8["old_pipe"]]}), "utf-8"
    )
    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]  # 옛 = [388, 394]
    assert _side(sub8["new_xy"], points[0]["x"], points[0]["y"]) < 0.01  # 옛 394 = 34.44 m
    assert not points[0].get("options")
    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

    routed = {
        "route_id": 9,
        "route_reset": {
            "reset": True,
            "dropped": {"structures": 1, "pipes": 2, "plus_stations": 1},
        },
    }
    monkeypatch.setattr(replan, "replan_route", record("route", routed))
    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={"50.00": {"pipe_kind": "흄관"}},
        uphill=None,
        cross=None,
        route=SimpleNamespace(),
        carry=SimpleNamespace(),
    )
    out = asyncio.run(design_save._run_steps(uuid4(), request, None))
    assert called == ["before", "route", "settings"], called
    assert out["route"]["route_reset"]["reset"] is True


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 B06_Section.B06_Section_Router_DesignSave_Route import StepFailed

    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 steps(project_id, request, session):
        (copy / "pipe_points.json").write_text("new", encoding="utf-8")  # 초기값 다시 뜸
        raise StepFailed("route", "뒤 차례 실패")

    monkeypatch.setattr(design_save, "get_db_pool", lambda: base._Pool([]))
    monkeypatch.setattr(design_save, "get_project_storage_relative_path", _async("storage/x"))
    monkeypatch.setattr(design_save, "resolve_stored_project_path", lambda stored: str(tmp_path))
    monkeypatch.setattr(guard, "snapshot_rows", _async({}))
    monkeypatch.setattr(guard, "restore_rows", _async(None))
    monkeypatch.setattr(design_save, "_run_steps", steps)
    request = design_save.DesignSaveRequest.model_construct(
        route_id=7, confirm=False, pipes=None, culvert_options=None, route=SimpleNamespace()
    )
    response = asyncio.run(design_save.save_design(uuid4(), request, None))
    assert response.status_code == 409
    assert (copy / "pipe_points.json").read_text(encoding="utf-8") == "old"