Aislo/resources/tester/test_476_1_apply_initial.py

# -*- coding: utf-8 -*-
# ruff: noqa: F811 — 346-3 fixture 를 불러 씀
"""476-1 ① 추천 [적용] = 늘 새 규칙 초기 종단(요구 276) — 옛 계획고 · 편집 · L 안 옮김.

체인은 346-3 시험의 가짜 세계(`chain_world`)로 · 종단 파일은 실제 1차 계획선.
"""

from __future__ import annotations

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

import pytest

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))

from test_346_3_route_save import (  # noqa: E402, F401 — 가짜 세계 fixture
    NEW,
    _async,
    _longitudinal,
    _run,
    api_world,
    chain_world,
)

import B05_Profile.B05_Profile_Repository as b05_repo  # 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_Router as b06_router  # noqa: E402
from B05_Profile.B05_Profile_Route_Save_Parts import vertices_of  # noqa: E402

LONG = "B06_Section/longitudinal/longitudinal.json"


def _real_profile_world(world, monkeypatch):
    """옛 종단 = 손 편집 · L 15 · 새 종횡단 = 실제 1차 계획선 — `_carry_profile` 은 진짜."""
    total = float(vertices_of(NEW)[-1].chainage_m)
    old = _longitudinal(total)
    old["profile_alignment"]["edits"] = {"station_offsets": {"40.000": 0.7}, "curve_radii": {}}
    old["profile_alignment"]["policy"]["default_curve_length_m"] = 15.0
    path = world.root / LONG
    path.write_text(json.dumps(old), "utf-8")
    fresh = _longitudinal(total)

    async def regenerate(*args, **kwargs):
        world.order.append("sections")
        path.write_text(json.dumps(fresh), "utf-8")

    monkeypatch.setattr(b06_router, "regenerate_route_sections", regenerate)
    monkeypatch.setattr(save, "_carry_profile", _CARRY)
    monkeypatch.setattr(b05_repo, "update_longitudinal_grade_summary", _async(None))
    return path, fresh["profile_alignment"]


_CARRY = save._carry_profile


def test_적용_끔이면_계획고_안_옮김(chain_world):
    result = _run(chain_world, initial=True)
    assert "error" not in result, result
    assert "profile" not in chain_world.order  # 옛 코드 = 같은 노선 옛 꼭짓점 통째 옮김


def test_적용_같은_노선_끔_새_1차_계획선(chain_world, monkeypatch):
    path, fresh = _real_profile_world(chain_world, monkeypatch)
    _run(chain_world, initial=True)
    alignment = json.loads(path.read_text("utf-8"))["profile_alignment"]
    assert alignment == fresh  # 꼭짓점 · L(정책) · 편집 = 새 1차 계획선(438) 그대로
    assert not (alignment.get("edits") or {}).get("station_offsets")


def test_손_노선_저장은_계획고_옮김_그대로(chain_world, monkeypatch):
    path, fresh = _real_profile_world(chain_world, monkeypatch)
    _run(chain_world)
    alignment = json.loads(path.read_text("utf-8"))["profile_alignment"]
    assert alignment["edits"]["station_offsets"].get("40.000") == 0.7  # 앞 구간 손 편집 옮김


def test_적용_켬_균형_셈_실패도_1차_계획선(chain_world, monkeypatch):
    path, fresh = _real_profile_world(chain_world, monkeypatch)
    _run(chain_world, profile_target=None, initial=True)  # 균형 목표 None(셈 실패)
    assert json.loads(path.read_text("utf-8"))["profile_alignment"] == fresh


def test_적용은_체인에_initial(api_world, monkeypatch):
    seen: list = []

    async def recompute(*args, **kwargs):
        seen.append(args[6] if len(args) > 6 else kwargs.get("initial"))
        return {"route_id": 9, "total_length_m": 10.0}

    monkeypatch.setattr(replan, "_recompute", recompute)
    body = save.RouteSaveRequest(vertices=[{"x": 0, "y": 0}, {"x": 5, "y": 5}])
    asyncio.run(replan._save_route(uuid4(), body))  # 잠금 · 떠 두기 밖 본체(④)
    asyncio.run(replan._save_route(uuid4(), body.model_copy(update={"as_initial": True})))
    assert seen == [False, True]


def test_횡단_손_입력_넘김_한_자리():
    # 사용자 답 전 = [적용]도 넘김(475-3) — 답 뒤 이 함수 한 줄
    assert save.carries_cross_inputs(True) is True and save.carries_cross_inputs(False) is True


# ── ④ [적용] = [저장]과 같은 잠금 · 떠 두고 실패면 되돌림 ─────────────────────────


class _Files:
    def discard(self):
        pass


@pytest.fixture
def guarded(monkeypatch, tmp_path):
    from fastapi.responses import JSONResponse

    import B06_Section.B06_Section_Router_DesignSave as design_save
    import common_util.common_util_save_guard as guard

    log: list = []

    class _Pool:
        def acquire(self):
            class _A:
                async def __aenter__(self):
                    return None

                async def __aexit__(self, *exc):
                    return False

            return _A()

    async def save(project_id, body):
        rev = int(body.vertices[0].x)
        log.append(("start", rev))
        await asyncio.sleep(0.05)
        log.append(("end", rev))
        if rev == 9:
            return JSONResponse(status_code=409, content={"status": "error", "message": "끊김"})
        return {"status": "success", "route_id": 1}

    async def restore(project_id, rows, files):
        log.append(("restore", rows))
        return True

    monkeypatch.setattr(replan, "_project_paths", _async((tmp_path, "p")))
    monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool())
    monkeypatch.setattr(replan, "_save_route", save)
    monkeypatch.setattr(guard, "snapshot_rows", lambda *a, **k: log.append("rows") or _async("R")())
    monkeypatch.setattr(guard, "snapshot_storage", lambda *a: log.append("files") or _Files())
    monkeypatch.setattr(design_save, "_restore", restore)
    return log


def _apply(rev: int, as_initial: bool = True):
    body = save.RouteSaveRequest(
        vertices=[{"x": rev, "y": 0}, {"x": 5, "y": 5}], as_initial=as_initial
    )
    return replan.replan_route(_PID, body)  # 같은 프로젝트 = 같은 잠금


_PID = uuid4()


def test_적용_겹치면_기다림(guarded):
    async def both():
        return await asyncio.gather(_apply(1), _apply(2))

    first, second = asyncio.run(both())
    assert first["waited"] is False and second["waited"] is True
    starts = [entry for entry in guarded if isinstance(entry, tuple) and entry[0] != "restore"]
    assert starts == [("start", 1), ("end", 1), ("start", 2), ("end", 2)]  # 하나씩
    assert "rows" in guarded and "files" in guarded


def test_적용_실패면_되돌림(guarded):
    response = asyncio.run(_apply(9))
    assert response.status_code == 409
    assert ("restore", "R") in guarded


def test_손_노선_저장은_잠금_밖(guarded):
    result = asyncio.run(_apply(1, as_initial=False))
    assert "waited" not in result and "rows" not in guarded  # /design/save 가 잠금 · 떠 둠