Aislo/resources/tester/test_224_3_recommend_store.py

"""계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3) — `B05_Profile_Engine_RouteInitial_Store`.

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

같은 입력 = 같은 키 · 하나만 바뀌어도 다른 키 · 두 번째 요청 셈 0회 · 덮어쓰기 없음 ·
[초기화](스냅샷 복원) 뒤에도 남음 · 자료 교체 때 지워짐 · 덮개 수는 꺼낼 때 지금 관으로.
엔드포인트 몸통(`compute_initial`)은 DB · 지형 읽기만 갈아 끼우고 그대로 돌린다.
"""

import asyncio
import json
import shutil
from contextlib import asynccontextmanager
from pathlib import Path
from uuid import uuid4

import numpy as np
import pytest

from B05_Profile import B05_Profile_Router_Replan as replan
from B05_Profile import B05_Profile_Router_RouteInitial as router
from B05_Profile.B05_Profile_Engine_RouteInitial import STREAM_OFFSET_M, Terrain
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
    DEFAULT_MODES,
    request_fingerprint,
    store_dir,
    write_stored,
)
from common_util.common_util_initial_snapshot import (
    _FILE_TREES,
    restore_snapshot_files,
    snapshot_dir,
)
from common_util.common_util_project_reset import purge_output_folders


def _terrain(lift: float = 0.0) -> Terrain:
    x = np.arange(0.0, 40.0, 2.0)
    y = np.arange(0.0, 20.0, 2.0)
    xx, yy = np.meshgrid(x, y)
    return Terrain(x, y, 0.1 * xx + lift, np.ones(xx.shape, dtype=bool))


EXPECTED = [(0.0, 10.0), (20.0, 10.0), (38.0, 10.0)]
STREAMS = [[(5.0, 0.0), (5.0, 18.0)]]
BASE = {
    "mode": "weighted",
    "weights": None,
    "max_grade": 0.14,
    "min_radius_m": 12.0,
    "stream_offset_m": None,
    "grade_smooth_m": None,
    "target_grade": None,
    "with_profile": True,
    "design_speed": 20,
}


def _key(expected=EXPECTED, terrain=None, streams=STREAMS, **changes) -> str:
    return request_fingerprint(expected, terrain or _terrain(), streams, **{**BASE, **changes})[0]


def test_same_input_same_key() -> None:
    assert _key() == _key()
    # 비운 값 = 기본값 · 비중 None = 50/25/25 — 화면이 채워 보내도 같은 키
    assert _key() == _key(stream_offset_m=STREAM_OFFSET_M)
    assert _key() == _key(weights={"expected": 50.0, "earthwork": 25.0, "grade": 25.0})
    # 비중 갈래가 아니면 비중은 갈래가 정함 — 보낸 비중은 키에 안 듦
    assert _key(mode="earthwork") == _key(mode="earthwork", weights={"expected": 1.0})


@pytest.mark.parametrize(
    "change",
    [
        {"mode": "earthwork"},
        {"weights": {"expected": 40.0, "earthwork": 30.0, "grade": 30.0}},
        {"max_grade": 0.12},
        {"min_radius_m": 15.0},
        {"stream_offset_m": 50.0},
        {"grade_smooth_m": 3.0},
        {"target_grade": 0.08},
        {"with_profile": False},
        {"design_speed": 30},
        {"expected": [(0.0, 10.0), (20.0, 11.0), (38.0, 10.0)]},
        {"terrain": _terrain(lift=0.5)},
        {"streams": []},
    ],
    ids=lambda change: next(iter(change)),
)
def test_one_change_other_key(change: dict) -> None:
    assert _key(**change) != _key()


def test_no_overwrite(tmp_path: Path) -> None:
    assert write_stored(tmp_path, "k", "p", {}, {"n": 1}, True)
    assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False)
    data = json.loads((store_dir(tmp_path) / "k.json").read_text(encoding="utf-8"))
    assert data["result"] == {"n": 1} and data["default"] is True
    assert set(data) == {"version", "key", "pool", "default", "created_at", "request", "result"}


def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None:
    (tmp_path / "B05_Profile/route").mkdir(parents=True)
    (tmp_path / "B05_Profile/route/initial_choice.json").write_text("{}", encoding="utf-8")
    write_stored(tmp_path, "k", "p", {}, {}, True)
    # 스냅샷 = 작업본 트리 사본(`save_initial_snapshot` 과 같은 자리)
    for tree in _FILE_TREES:
        source = tmp_path / tree
        if source.is_dir():
            shutil.copytree(source, snapshot_dir(tmp_path) / tree.replace("/", "__"))
    restore_snapshot_files(tmp_path)
    assert (store_dir(tmp_path) / "k.json").is_file()
    purge_output_folders(tmp_path)  # 자료 교체
    assert not store_dir(tmp_path).exists()


class _Calls:
    def __init__(self) -> None:
        self.count = 0


@pytest.fixture
def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
    """`compute_initial` 의 DB · 파일 읽기를 갈아 끼움 · 셈은 횟수만 세는 가짜."""
    import B05_Profile.B05_Profile_Engine_Grade as grade
    import B05_Profile.B05_Profile_Engine_RouteInitial as engine
    import B05_Profile.B05_Profile_Engine_RouteInitial_Profile as profile
    import common_util.common_util_surface_confirmation as surface

    calls = _Calls()

    async def project_paths(_pid):
        return tmp_path, "x"

    async def plan_criteria(_pid):
        return 12.0, 12.0, 5.0

    async def road_settings(_pid):
        return "간선", 20, "일반"

    async def surface_params(_connection, _pid):
        return {}

    class _Pool:
        @asynccontextmanager
        async def acquire(self):
            yield None

    def generate(expected, terrain, streams, *, pipes, **_inputs):
        assert pipes is None  # 관은 키 밖 — 덮개 없이 셈해 남김
        calls.count += 1
        return {
            "planned": [[x, y] for x, y in expected],
            "metrics": {
                "length_m": 38.0,
                "offset_avg_m": 0.0,
                "profile": {
                    "stations_m": [0.0, 38.0],
                    "plan_m": [1.0, 1.0],
                    "ground_m": [0.0, 0.0],
                },
            },
        }

    monkeypatch.setattr(replan, "_project_paths", project_paths)
    monkeypatch.setattr(replan, "_plan_criteria", plan_criteria)
    monkeypatch.setattr(replan, "_road_settings", road_settings)
    monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED])
    monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
    monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params)
    monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
    monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
    monkeypatch.setattr(engine, "load_terrain_and_streams", lambda *_a: (_terrain(), STREAMS))
    monkeypatch.setattr(engine, "generate_initial_route", generate)
    # 덮개 여유 2 m 관 하나가 노선 위 — 계획고 − 지반고 1 m 라 모자람 1
    monkeypatch.setattr(profile, "project_pipes", lambda _root: [(19.0, 10.0, 2.0)])
    calls.root = tmp_path
    return calls


def _ask(**fields) -> dict:
    request = replan.RouteInitialRequest(**fields)
    return asyncio.run(router.compute_initial(uuid4(), request))


def test_second_request_computes_nothing(wired: _Calls) -> None:
    first = _ask(mode="earthwork")
    assert wired.count == 1 and first["stored"] is False
    second = _ask(mode="earthwork")
    assert wired.count == 1 and second["stored"] is True
    assert second["planned"] == first["planned"]
    # 덮개 수는 꺼낼 때마다 지금 관으로 · 저장본에는 없음
    assert second["metrics"]["profile"]["cover_short"] == 1
    files = list(store_dir(wired.root).glob("*.json"))
    assert len(files) == 1
    saved = json.loads(files[0].read_text(encoding="utf-8"))
    assert "cover_short" not in saved["result"]["metrics"]["profile"]
    assert saved["default"] is True and saved["request"]["mode"] == "earthwork"
    # 제약 하나 바꾸면 파일 하나 더 · 옛 파일 그대로 · 기본값 표시 꺼짐
    changed = _ask(mode="earthwork", criteria={"min_radius_m": 15.0})
    assert changed["stored"] is False and wired.count == 2
    assert len(list(store_dir(wired.root).glob("*.json"))) == 2
    assert json.loads(files[0].read_text(encoding="utf-8")) == saved


def test_same_pool_in_flight_waits(wired: _Calls) -> None:
    async def both() -> list[dict]:
        request = replan.RouteInitialRequest(mode="gentle")
        project = uuid4()
        return await asyncio.gather(
            router.compute_initial(project, request), router.compute_initial(project, request)
        )

    results = asyncio.run(both())
    assert wired.count == 1
    assert sorted(result["stored"] for result in results) == [False, True]


def test_precompute_defaults_then_hits(wired: _Calls) -> None:
    asyncio.run(router.precompute_defaults(uuid4()))
    assert wired.count == len(DEFAULT_MODES) == 5
    saved = [
        json.loads(path.read_text(encoding="utf-8"))
        for path in store_dir(wired.root).glob("*.json")
    ]
    assert sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES)
    assert all(item["default"] for item in saved)
    # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 추적은 종단 끔) — 셈 없이 저장본
    hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25})
    follow = _ask(mode="follow", with_profile=False)
    assert hit["stored"] and follow["stored"] and wired.count == 5