Aislo/resources/tester/test_309_3_recommend_preprocess_only.py

"""요구 309-3 · 310 · PLAN 505-1 — 추천 노선 미리 셈 = 전처리 체인에서만.

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

B05 열기(GET /route/plan) · 노선 [저장] · [적용] 뒤 다시 세움 = 미리 셈 0 · 저장본 읽기만(파일
안 건드림) · 미리 셈 갈래가 저장본에 없으면 409 「전처리를 다시 하면 생김」 · 사용자가 바꾼 값 ·
[재계산]은 그 자리 셈 · 체인 셈(`run_precompute`)이 만들면 화면이 꺼냄 · 초기 종단 지문에 노선
파일 없음.
"""

from __future__ import annotations

import asyncio
import json
import os
import re
import sys
from pathlib import Path
from uuid import uuid4

ROOT = Path(__file__).resolve().parents[2]
TESTER = Path(__file__).resolve().parent
for path in (ROOT, TESTER):
    if str(path) not in sys.path:
        sys.path.insert(0, str(path))

import helper_472_1_paths as P  # noqa: E402
from test_224_2_station_snap import _NODES, _patch_replan, _working  # noqa: E402
from test_224_3_recommend_store import _ask, wired  # noqa: E402, F401 — 가짜 세계 fixture
from test_476_1_recommend_initial import gather  # noqa: E402, F401

from B05_Profile import B05_Profile_Engine_InitialProfile as ip  # noqa: E402
from B05_Profile import B05_Profile_Router_RouteInitial as router  # noqa: E402
from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (  # noqa: E402
    DEFAULT_MODES,
    read_stored,
    store_dir,
    write_stored,
)


def test_B05_열기_미리_셈_0(tmp_path: Path, monkeypatch) -> None:
    import B03_FileInput.B03_FileInput_Router_Helpers as helpers

    scheduled: list[str] = []
    monkeypatch.setattr(
        helpers,
        "_schedule_background_task",
        lambda coroutine, task_name: (scheduled.append(task_name), coroutine.close()),
    )
    replan = _patch_replan(monkeypatch, tmp_path)
    replan._write_planned_polyline(_working(tmp_path), _NODES, 30.0, simplify=False)
    body = asyncio.run(replan.read_route_plan(uuid4()))
    assert body["status"] == "success" and scheduled == []
    assert not store_dir(tmp_path).exists()


def test_미리_셈_부르는_곳은_체인만() -> None:
    """노선 [저장] · [적용] · B05 열기 길에는 미리 셈 호출이 없음 — 체인 둘 · B01 사용 범위만."""
    callers = set()
    for path in ROOT.glob("B0*/*.py"):
        text = path.read_text(encoding="utf-8")
        if re.search(r"\b(schedule_precompute|run_precompute|refresh_expected_route)\(", text):
            callers.add(path.name)
    assert callers == {
        "B03_FileInput_Service_Chain.py",  # 업로드 체인 · B04 재확정 체인
        "B01_Dashboard_Router.py",  # 사용 범위 바꿈 = 예상노선 다시 씀
        "B05_Profile_Router_RouteInitial.py",  # 정의
    }


def test_화면_기본_갈래_저장본_없으면_409_셈_0(wired) -> None:  # noqa: F811
    for mode in DEFAULT_MODES:
        response = _ask(False, mode=mode)
        assert response.status_code == 409
        assert "전처리를 다시 하면 생깁니다" in json.loads(response.body)["message"]
    assert wired.count == 0 and not list(store_dir(wired.root).glob("*.json"))


def test_체인이_만들면_화면이_꺼냄(wired) -> None:  # noqa: F811
    asyncio.run(router.run_precompute(uuid4()))  # 체인 끝 예약이 도는 한 줄
    assert wired.count == 5
    for mode in DEFAULT_MODES:
        assert _ask(False, mode=mode)["stored"] is True
    assert wired.count == 5


def test_화면이_실은_분석_지형이_설계_지형이면_기본_갈래(wired) -> None:  # noqa: F811
    """창 지형 — 설계 지형이든 다른 지형이든 기본 갈래는 저장본만(509-5 · 지형마다 미리 셈)."""
    assert _ask(False, mode="gentle", terrain="las").status_code == 409
    assert _ask(False, mode="gentle", terrain="sheet").status_code == 409
    assert wired.count == 0


def test_사용자가_바꾼_값_재계산은_셈(wired) -> None:  # noqa: F811
    assert _ask(False, mode="earthwork", criteria={"min_radius_m": 15.0})["stored"] is False
    assert _ask(False, mode="earthwork", with_profile=False)["stored"] is False
    assert wired.count == 2
    assert _ask(False, mode="gentle", recompute=True)["stored"] is False
    assert wired.count == 3


def test_초기_종단_화면은_저장본만(wired, monkeypatch) -> None:  # noqa: F811
    import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer

    flags: list[bool] = []

    async def fake(project_id, root, stored, result, request, selection, key, compute=True):
        flags.append(compute)
        return {"pvi_count": 3}

    monkeypatch.setattr(gatherer, "initial_profile_of", fake)
    _ask(True, mode="earthwork")  # 체인
    _ask(False, mode="earthwork")  # 화면 기본 갈래
    _ask(False, mode="earthwork", criteria={"min_radius_m": 15.0})  # 사용자가 바꾼 값
    assert flags == [True, False, True]


def test_꺼내기는_파일_안_건드림(tmp_path: Path) -> None:
    write_stored(tmp_path, "k", "p", {"mode": "gentle"}, {"nodes": []}, True)
    path = store_dir(tmp_path) / "k.json"
    os.utime(path, ns=(1_000_000_000, 1_000_000_000))
    assert read_stored(tmp_path, "k") is not None
    assert path.stat().st_mtime_ns == 1_000_000_000


def test_초기_종단_지문에_노선_파일_없음(gather, monkeypatch) -> None:  # noqa: F811
    first = gather["run"]()
    assert "error" not in first, first
    assert gather["run"]("k2", False) == {"error": router.NOT_PREPARED}  # 저장본 없음 = 셈 안 함

    def broken(*args, **kwargs):
        raise RuntimeError("다시 셈")

    monkeypatch.setattr(ip, "initial_longitudinal", broken)
    route = gather["root"] / P.ROUTE
    doc = json.loads(route.read_text("utf-8"))
    doc["properties"] = {**(doc.get("properties") or {}), "saved": "노선 [저장]"}
    route.write_text(json.dumps(doc), "utf-8")  # 노선 [저장] = 노선 파일만 바뀜
    assert gather["run"]("k1", False) == first


def test_엔진_지문_다른_저장본_접근은_읽음_체인은_다시_셈(wired, monkeypatch) -> None:  # noqa: F811
    """505-1 덧 — 엔진 코드를 고쳐도(소스 지문 바뀜) 화면은 옛 저장본을 꺼냄 · 체인만 다시 셈."""
    import B05_Profile.B05_Profile_Engine_RouteInitial_Store as store

    asyncio.run(router.run_precompute(uuid4()))
    assert wired.count == 5
    monkeypatch.setattr(store, "source_fingerprint", lambda: "엔진 고침")
    for mode in DEFAULT_MODES:
        assert _ask(False, mode=mode)["stored"] is True
    assert wired.count == 5  # 접근 = 셈 0 · 안내 없음
    # 입력이 다르면(제약 바꿈 · 사용자 셈 갈래) 옛 지문을 끌어오지 않음
    assert _ask(False, mode="gentle", criteria={"min_radius_m": 15.0})["stored"] is False
    assert wired.count == 6
    asyncio.run(router.run_precompute(uuid4()))  # 체인 = 지문 견줌 · 다시 셈
    assert wired.count == 11
    sources = {
        json.loads(p.read_text("utf-8"))["source"] for p in store_dir(wired.root).glob("*.json")
    }
    assert "엔진 고침" in sources and len(sources) == 2  # 옛 판은 30일 정리 몫


def test_초기_종단_접근은_코드_지문_안_견줌(gather, monkeypatch) -> None:  # noqa: F811
    first = gather["run"]()
    assert "error" not in first, first

    def broken(*args, **kwargs):
        raise RuntimeError("다시 셈")

    monkeypatch.setattr(ip, "initial_longitudinal", broken)
    monkeypatch.setattr(ip, "source_fingerprint", lambda: "엔진 고침")
    assert gather["run"]("k1", False) == first  # 접근 = 그대로 읽음
    assert "다시 셈" in gather["run"]("k1", True)["error"]  # 체인 = 다시 셈
    path = gather["root"] / ip.INITIAL_DIR / "k1.json"
    data = json.loads(path.read_text("utf-8"))
    path.write_text(json.dumps({k: v for k, v in data.items() if k != "source"}), "utf-8")
    assert gather["run"]("k1", False) == first  # 코드 지문 없는 옛 파일도 읽음
    path.write_text(json.dumps(data), "utf-8")
    gather["state"]["stages"][0]["params"]["max_grade_pct"] = 7.5  # 입력이 바뀜
    assert gather["run"]("k1", False) == {"error": router.NOT_PREPARED}