Aislo/resources/tester/test_351_1_recommend_gaps.py

"""351-1 노선 추천 저장 빈틈 — 재확정 체인 · 기본값 지킴 · 저장된 제약 · B01 사용 범위.

    ./venv/Scripts/python.exe -m pytest resources/tester/test_351_1_recommend_gaps.py -q

체인 뒤 첫 누름 = 저장본(기본값 · 저장된 제약 둘 다) · 「최근 2벌」 정리가 기본값을 안 밀어냄 ·
사용 범위를 바꾸면 예상노선을 원본에서 다시 자르고 미리 셈.
"""

import asyncio
import json
import time
from pathlib import Path
from uuid import uuid4

import pytest
from test_224_3_recommend_store import _ask, wired  # noqa: F401 — 픽스처

from B05_Profile import B05_Profile_Router_RouteInitial as router
from B05_Profile.B05_Profile_Engine_RouteInitial import write_initial_choice
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
    DEFAULT_MODES,
    is_default,
    store_dir,
    write_stored,
)
from common_util.common_util_route_geometry import (
    expected_route_csv_path,
    load_design_route,
    planned_route_working_path,
    write_route_csv,
)


def _saved(root: Path) -> list[dict]:
    return [json.loads(p.read_text(encoding="utf-8")) for p in store_dir(root).glob("*.json")]


def test_저장된_제약으로도_미리_셈_첫_누름_저장본(wired) -> None:  # noqa: F811
    criteria = {"min_radius_m": 15.0, "max_grade_pct": 12.0}
    write_initial_choice(
        wired.root, "weighted", {"expected": 40, "earthwork": 30, "grade": 30}, criteria
    )
    asyncio.run(router.precompute_defaults(uuid4()))
    assert wired.count == 2 * len(DEFAULT_MODES)
    assert all(item["default"] for item in _saved(wired.root))
    # 모달은 저장된 제약 · 비중으로 열림 — 셈 없이 저장본
    hit = _ask(
        mode="weighted", weights={"expected": 40, "earthwork": 30, "grade": 30}, criteria=criteria
    )
    gentle = _ask(mode="gentle", criteria=criteria)
    plain = _ask(mode="earthwork")
    assert hit["stored"] and gentle["stored"] and plain["stored"]
    assert wired.count == 10
    # 다시 불러도 셈 0
    asyncio.run(router.precompute_defaults(uuid4()))
    assert wired.count == 10


def test_저장된_제약_없으면_기본값만(wired) -> None:  # noqa: F811
    write_initial_choice(wired.root, "gentle", None, None)
    asyncio.run(router.precompute_defaults(uuid4()))
    assert wired.count == len(DEFAULT_MODES)


def test_기본값은_최근_두_벌_정리에서_안_밀림(tmp_path: Path) -> None:
    request = {"mode": "earthwork", "terrain": None}
    write_stored(tmp_path, "d", "p", request, {}, True)
    for name in ("a", "b", "c"):
        time.sleep(0.02)
        write_stored(tmp_path, name, "p", request, {}, False)
    names = sorted(path.stem for path in store_dir(tmp_path).glob("*.json"))
    assert names == ["b", "c", "d"]


def test_기본값_판정() -> None:
    saved = {
        "criteria": {"min_radius_m": 15.0},
        "weights": {"expected": 40.0, "earthwork": 30.0, "grade": 30.0},
    }
    assert is_default("gentle", None, None, None)
    assert is_default("gentle", None, {"min_radius_m": None}, None)
    assert is_default("gentle", None, None, "las")  # 지형 안 따짐(509-5 · 지형마다 미리 셈)
    assert not is_default("gentle", None, {"min_radius_m": 15.0}, None)
    assert is_default("gentle", None, {"min_radius_m": 15.0}, None, saved)
    assert not is_default("gentle", None, {"min_radius_m": 16.0}, None, saved)
    weights = {"expected": 40.0, "earthwork": 30.0, "grade": 30.0}
    assert is_default("weighted", weights, {"min_radius_m": 15.0}, None, saved)
    assert not is_default("weighted", None, {"min_radius_m": 15.0}, None, saved)


def test_두_체인_끝에서_미리_셈_예약() -> None:
    source = Path("B03_FileInput/B03_FileInput_Service_Chain.py").read_text(encoding="utf-8")
    start = source.index("async def run_auto_design_chain(")
    upload = source[start : source.index("async def run_preprocess_chain(")]
    preprocess = source[source.index("async def run_preprocess_chain(") :]
    # 두 체인 모두 finally — 실패 · 건너뜀 길에도 예약(763-3 · 실행 시험 = test_763_3) ·
    # 옛 재확정 체인(run_redesign_chain)은 걷음(765-3) · 두 벌 체인이 자리를 이음
    for body in (upload, preprocess):
        assert "schedule_precompute(project_id)" in body.split("    finally:", 1)[1]
    b01 = Path("B01_Dashboard/B01_Dashboard_Router.py").read_text(encoding="utf-8")
    assert "refresh_expected_route(" in b01


def test_사용_범위_비교() -> None:
    from decimal import Decimal

    from B01_Dashboard.B01_Dashboard_Router import _range_of

    assert _range_of({"route_start_m": Decimal("10.0"), "route_end_m": None}) == (10.0, None)
    assert _range_of({"route_start_m": 10.0, "route_end_m": None}) == _range_of(
        {"route_start_m": Decimal("10"), "route_end_m": None}
    )
    assert _range_of({"route_start_m": 10.0}) != _range_of({"route_start_m": 20.0})


def _source_project(root: Path) -> None:
    write_route_csv(
        root / "B03_FileInput/input/route.csv",
        [{"x": float(x), "y": 0.0} for x in range(0, 101, 10)],
    )
    # 설계 정본(수정본) · 예상노선이 따로 있음 — 원본에서 다시 읽어야 함
    write_route_csv(planned_route_working_path(root), [{"x": 0.0, "y": 5.0}, {"x": 9.0, "y": 5.0}])
    write_route_csv(expected_route_csv_path(root), [{"x": 0.0, "y": 0.0}, {"x": 100.0, "y": 0.0}])


def test_원본에서_다시_읽기(tmp_path: Path) -> None:
    _source_project(tmp_path)
    surface = {"source_filter": "csf", "method": "dtm", "smooth": False}
    stored = load_design_route(tmp_path, surface, (20.0, 80.0))
    assert [v.y for v in stored.vertices] == [5.0, 5.0]  # 정본이 먼저
    fresh = load_design_route(tmp_path, surface, (20.0, 80.0), from_source=True)
    xs = [round(v.x, 3) for v in fresh.vertices]
    assert xs[0] == 20.0 and xs[-1] == 80.0


def test_사용_범위_바꾸면_예상노선_다시_쓰고_미리_셈(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import B03_FileInput.B03_FileInput_Repository as repo
    import common_util.common_util_storage as storage
    import common_util.common_util_surface_confirmation as surface

    _source_project(tmp_path)
    calls: list[str] = []

    class _Cursor:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_):
            return None

        async def execute(self, *_):
            return None

        async def fetchone(self):
            return (30.0, 70.0)

    class _Connection:
        def cursor(self):
            return _Cursor()

    class _Acquire:
        async def __aenter__(self):
            return _Connection()

        async def __aexit__(self, *_):
            return None

    class _Pool:
        def acquire(self):
            return _Acquire()

    async def storage_path(*_):
        return "x"

    async def surface_params(*_):
        return {"source_filter": "csf", "method": "dtm", "smooth": False}

    async def precompute(_pid):
        calls.append("precompute")

    monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
    monkeypatch.setattr(repo, "get_project_storage_relative_path", storage_path)
    monkeypatch.setattr(storage, "resolve_stored_project_path", lambda _p: str(tmp_path))
    monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params)
    monkeypatch.setattr(router, "precompute_defaults", precompute)
    asyncio.run(router.refresh_expected_route(uuid4()))
    rows = expected_route_csv_path(tmp_path).read_text(encoding="utf-8").splitlines()
    assert rows[1].split(",")[1:] == ["30.0", "0.0"] and rows[-1].split(",")[1:] == ["70.0", "0.0"]
    assert calls == ["precompute"]
    # 체인 전(예상노선 없음)은 건너뜀
    expected_route_csv_path(tmp_path).unlink()
    asyncio.run(router.refresh_expected_route(uuid4()))
    assert calls == ["precompute"] and not expected_route_csv_path(tmp_path).exists()