Aislo/resources/tester/test_b05_routeedit_initial_roundtrip.py

"""초기노선 갈래 저장 왕복 — [확인]에 실은 초기값이 다시 열기 뒤 그대로인가(계획서 20).

환경 배정 storage 프로젝트(데스크탑_메인)의 **사본**에서 화면과 같은 차례로 서버 길을 부른다.
갈래(추종 · 토공 최소 · 기울기 순한 · 비중 50/25/25)마다
  ① `POST /route/initial` → 비교값이 다 있음
  ② 화면 코드로 [확인] 몸통을 만들어 `POST /route/replan` → 서버가 initial 을 적음
  ③ `GET /route/plan` → 현재 초기값 = 보낸 것 · 다시 연 노드 = 보낸 노드
  ④ [초기화] — 캐시가 있으면 그 노드, 없으면(다른 PC) 서버가 같은 갈래로 같은 노드를 냄
화면 쪽 셈은 `helper_b05_routeedit_initial_apply.cjs` 가 화면 코드 그대로 돌린다.

화면 제약 칸(종단기울기 · 최소 반지름 · 계류 이격 · 지반 고름, 계획서 20 확정)에 겹친 값도
같은 차례로 왕복 — 겹친 칸만 `current_initial.criteria` 에 남고 안 겹친 칸은 프로젝트 기본값.

- 파일: 프로젝트 폴더를 임시 폴더로 복사(원본 입력 · 점군 · 큰 파일 제외) ·
  `STORAGE_BASE_DIR` 를 그리로.
- DB: 읽기 전용 세션으로 진짜 DB 를 읽기만 함.
- 재계산 체인(`_recompute`, 배수유역부터 약 90초 · DB 에 씀)은 성공만 돌려주는 가짜로 갈음 —
  여기서 보는 것은 초기값이 적히고 다시 읽히는 길이다.
- 원본 프로젝트 · 사용자 프로젝트 · DB 는 바뀌지 않음.
"""

from __future__ import annotations

import json
import shutil
import subprocess
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any

import pytest

ROOT = Path(__file__).resolve().parents[2]
PROJECT_ID = "79da822d-921a-40f8-bb93-d8ffd90eeedf"  # OWNERS.md — desktop-main `데스크탑_메인`
SOURCE = ROOT / "storage" / "1" / "3" / PROJECT_ID
_COPY_DIRS = ("B04_PreProcess/drainage", "B05_Profile", "B06_Section", "initial_snapshot")
_COPY_MAX_BYTES = 20 * 1024 * 1024
HELPER = Path(__file__).with_name("helper_b05_routeedit_initial_apply.cjs")

pytestmark = pytest.mark.skipif(not SOURCE.is_dir(), reason="배정 storage 프로젝트가 없음")

LOGICS: list[tuple[str, dict[str, float] | None]] = [
    ("follow", None),
    ("earthwork", None),
    ("gentle", None),
    ("weighted", {"expected": 50, "earthwork": 25, "grade": 25}),
]


def _copy_project(dest: Path) -> None:
    for rel in _COPY_DIRS:
        shutil.copytree(SOURCE / rel, dest / rel)
    for rel in ("B04_PreProcess/models", "B04_PreProcess/processed"):
        (dest / rel).mkdir(parents=True, exist_ok=True)
        for item in (SOURCE / rel).iterdir():
            if item.is_file() and item.stat().st_size <= _COPY_MAX_BYTES:
                shutil.copy2(item, dest / rel / item.name)
    for item in SOURCE.iterdir():
        if item.is_file():
            shutil.copy2(item, dest / item.name)


@pytest.fixture(scope="module")
def api(tmp_path_factory):
    aiomysql = pytest.importorskip("aiomysql")
    from fastapi import FastAPI
    from fastapi.testclient import TestClient

    import B05_Profile.B05_Profile_Router_Replan as replan
    import common_util.common_util_storage as storage
    import config.config_db as config_db
    from config import config_system as cfg

    storage_root = tmp_path_factory.mktemp("initial") / "storage"
    project_root = storage_root / "1" / "3" / PROJECT_ID
    _copy_project(project_root)
    patch = pytest.MonkeyPatch()
    patch.setattr(storage, "STORAGE_BASE_DIR", str(storage_root))
    recomputed: list[str] = []

    async def fake_recompute(project_id, root, stored_path):
        recomputed.append(str(root))
        return {"route_id": None, "total_length_m": None, "vertex_count": 0}

    patch.setattr(replan, "_recompute", fake_recompute)

    @asynccontextmanager
    async def lifespan(_app):
        try:
            pool = await aiomysql.create_pool(
                host=cfg.DB_HOST,
                port=cfg.DB_PORT,
                db=cfg.DB_NAME,
                user=cfg.DB_USER,
                password=cfg.DB_PASSWORD,
                minsize=1,
                maxsize=4,
                autocommit=True,
                charset="utf8mb4",
                # 읽기 전용 세션 — 새는 쓰기가 있으면 DB 를 바꾸지 않고 터짐
                init_command="SET time_zone='+00:00', SESSION tx_read_only=1",
            )
        except Exception as exc:  # DB 가 안 닿는 곳
            pytest.skip(f"DB 에 안 닿음: {exc}")
        config_db.db_pool = pool
        try:
            yield
        finally:
            config_db.db_pool = None
            pool.close()
            await pool.wait_closed()

    app = FastAPI(lifespan=lifespan)
    app.include_router(replan.router)
    try:
        with TestClient(app) as client:
            client.root = project_root  # type: ignore[attr-defined]
            client.recomputed = recomputed  # type: ignore[attr-defined]
            yield client
    finally:
        patch.undo()


def _screen(jobs: list[dict[str, Any]]) -> list[Any]:
    if shutil.which("node") is None:
        pytest.skip("node 가 없음")
    out = subprocess.run(
        ["node", str(HELPER)],
        input=json.dumps({"jobs": jobs}),
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=120,
        check=True,
    )
    return json.loads(out.stdout)


def _initial(
    api,
    logic: str,
    weights: dict[str, float] | None,
    criteria: dict[str, float] | None = None,
) -> dict[str, Any]:
    body: dict[str, Any] = {"logic": logic}
    if weights:
        body["weights"] = weights
    if criteria:
        body["criteria"] = criteria
    response = api.post(f"/api/projects/{PROJECT_ID}/route/initial", json=body)
    assert response.status_code == 200, response.text
    return response.json()


def _close(a: list[list[float]], b: list[list[float]], tol: float = 1e-3) -> bool:
    return len(a) == len(b) and all(
        abs(p[0] - q[0]) <= tol and abs(p[1] - q[1]) <= tol for p, q in zip(a, b)
    )


@pytest.fixture(scope="module")
def rounds(api):
    """갈래마다 ①~③ — 결과를 모아 두고 시험마다 나눠 본다."""
    out = []
    before = api.get(f"/api/projects/{PROJECT_ID}/route/plan").json()
    for logic, weights in LOGICS:
        answer = _initial(api, logic, weights)
        [applied] = _screen(
            [{"kind": "apply", "response": answer, "logic": logic, "weights": weights}]
        )
        confirmed = api.post(f"/api/projects/{PROJECT_ID}/route/replan", json=applied["body"])
        assert confirmed.status_code == 200, confirmed.text
        plan = api.get(f"/api/projects/{PROJECT_ID}/route/plan").json()
        [reopened] = _screen([{"kind": "reopen", "plan": plan, "cached": applied["choice"]}])
        out.append(
            {
                "logic": logic,
                "weights": weights,
                "answer": answer,
                "applied": applied,
                "confirmed": confirmed.json(),
                "plan": plan,
                "reopened": reopened,
            }
        )
    return {"before": before, "rounds": out}


def test_원본은_처음_추종(rounds):
    assert rounds["before"]["current_initial"]["logic"] == "follow"


def test_비교값이_응답에_다_있음(rounds):
    for item in rounds["rounds"]:
        metrics = item["answer"]["metrics"]
        for key in ("length_m", "max_grade_pct", "avg_grade_pct", "cut_m3", "fill_m3"):
            assert isinstance(metrics[key], (int, float)), (item["logic"], key)
        assert isinstance(metrics["over_limit"]["total"], int)
        assert isinstance(metrics["offset_avg_m"], (int, float))
        # 화면 한 줄에 빈 칸(—)이 없다.
        assert "—" not in item["applied"]["metrics"], item["applied"]["metrics"]


def test_제약_못_지킨_구간이_목록과_화면_줄에(rounds):
    for item in rounds["rounds"]:
        answer, applied = item["answer"], item["applied"]
        violations = answer["metrics"]["violations"]
        assert answer["metrics"]["over_limit"]["total"] == len(violations)
        for entry in violations:
            assert entry["kind"] in ("grade", "radius", "stream")
            assert entry["station"].startswith("No.") and entry["reason"]
            assert isinstance(entry["value"], (int, float))
        head = (
            f"제약 못 지킨 구간 {len(violations)}곳: " if violations else "제약 못 지킨 구간 없음"
        )
        assert applied["violations"].startswith(head), applied["violations"]
        if answer.get("relaxed"):
            bound = f"{answer['terrain_bound_pct']:.1f}%"
            assert applied["relaxed"].endswith(f"상한을 {bound} 로 풂"), applied["relaxed"]
        else:
            assert applied["relaxed"] == ""
        smooth = answer["criteria"]["grade_smooth_m"]
        assert f"기울기 잴 지반 {smooth:.0f}m 고름" in applied["criteria"]


def test_확인하면_서버가_초기값을_적고_다시_열면_그대로(rounds, api):
    assert len(api.recomputed) == len(LOGICS)
    for item in rounds["rounds"]:
        sent = item["applied"]["body"]["initial"]
        assert sent["mode"] == item["logic"]
        assert item["confirmed"]["current_initial"]["logic"] == item["logic"]
        current = item["plan"]["current_initial"]
        assert current["logic"] == item["logic"]
        if item["weights"]:
            assert {k: round(v) for k, v in current["weights"].items()} == item["weights"]
        reopened = item["reopened"]
        assert reopened["base"]["logic"] == item["logic"]
        assert reopened["label"] == item["applied"]["label"]
        assert reopened["sameAsCache"] is True  # [초기화]가 캐시 노드를 씀
    saved = json.loads(
        (api.root / "B05_Profile" / "route" / "initial_choice.json").read_text("utf-8")
    )
    assert saved["logic"] == "weighted"


def test_다시_연_노선은_확인한_노선(rounds):
    for item in rounds["rounds"]:
        sent = [[v["x"], v["y"]] for v in item["applied"]["body"]["vertices"]]
        assert _close(item["reopened"]["planned"], sent), item["logic"]


def test_초기화는_마지막_초기값으로(rounds, api):
    last = rounds["rounds"][-1]
    cached = last["applied"]["choice"]["snapshot"]["planned"]
    # 캐시가 있으면 그 노드 — 마지막 적용(비중 50/25/25).
    assert _close(cached, [[v["x"], v["y"]] for v in last["applied"]["body"]["vertices"]])
    # 캐시가 없으면(다른 PC) 서버가 적어 둔 갈래 · 비중으로 다시 물음 → 같은 노드.
    base = last["reopened"]["base"]
    again = _initial(api, base["logic"], base["weights"])
    [redo] = _screen(
        [{"kind": "apply", "response": again, "logic": base["logic"], "weights": base["weights"]}]
    )
    assert _close(redo["choice"]["snapshot"]["planned"], cached)
    # 다른 갈래들과는 다른 노선이다(되돌림이 실제로 갈래를 가림).
    others = [r["applied"]["choice"]["snapshot"]["planned"] for r in rounds["rounds"][:-1]]
    assert all(not _close(o, cached) for o in others)


def test_제약_겹침도_확인_뒤_그대로(api, rounds):
    """화면 제약 칸(계류 이격 30m)만 겹쳐 물으면 답에 그 값이 실리고, [확인] → 다시 열기에도
    그 한 칸만 남는다(나머지는 기본값이라 안 담김 — 계획서 20 확정값이 프로젝트 기본값)."""
    criteria = {"stream_offset_m": 30.0}
    answer = _initial(api, "earthwork", None, criteria)
    assert answer["criteria"]["stream_offset_m"] == 30.0
    [applied] = _screen(
        [
            {
                "kind": "apply",
                "response": answer,
                "logic": "earthwork",
                "weights": None,
                "criteria": criteria,
            }
        ]
    )
    assert applied["body"]["initial"]["criteria"] == criteria
    confirmed = api.post(f"/api/projects/{PROJECT_ID}/route/replan", json=applied["body"])
    assert confirmed.status_code == 200, confirmed.text
    assert confirmed.json()["current_initial"]["criteria"] == criteria
    plan = api.get(f"/api/projects/{PROJECT_ID}/route/plan").json()
    assert plan["current_initial"]["criteria"] == criteria
    [reopened] = _screen([{"kind": "reopen", "plan": plan, "cached": applied["choice"]}])
    assert reopened["base"]["criteria"] == criteria
    assert reopened["sameAsCache"] is True  # 캐시와 서버 겹침이 같으니 [초기화]가 캐시를 쓴다