Aislo/resources/tester/test_472_6_design_save.py

"""PLAN 472-6 6-A — [저장] · [확정] 한 본문 창구 · 전부 아니면 없음.

  ① storage 떠 두기 · 되돌림 — 바뀐 · 새로 생긴 · 지운 파일만 되씀(안 바뀐 파일 그대로).
  ② 실제 창구(검증_sub8 · DB · storage) — 중간 차례 실패 → DB 7 표 행 · storage 해시 시작 그대로 ·
     차례 이름 · 까닭 · restored. 같은 본문 정상 = 옛 차례(안쪽 창구 차례로 부름)와 행 · 파일 같음.
검증 프로젝트가 없는 자리(다른 PC · DB 없음)는 ② 건너뜀.
② 는 살아 있는 검증_sub8 DB · storage 는 임시 사본(492 · `helper_temp_storage`) · 노선 바뀌어도 됨
(최신 노선 id · 추가 측점 자리를 그때 읽음).

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

from __future__ import annotations

import asyncio
import hashlib
import json
import sys
from pathlib import Path
from uuid import UUID

import pytest

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(HERE))

from common_util import common_util_save_guard as guard  # noqa: E402

PROJECT = "9872e9a7-e483-427b-a6b2-72cdbc063bc3"  # 검증_sub8(OWNERS.md)
VOLATILE = ("created_at", "updated_at", "confirmed_at", "started_at", "completed_at")


def test_storage_되돌림은_바뀐_것만(tmp_path) -> None:
    (tmp_path / "B06_Section" / "a").mkdir(parents=True)
    keep = tmp_path / "B06_Section" / "a" / "keep.json"
    edit = tmp_path / "B06_Section" / "edit.json"
    gone = tmp_path / "workflow.json"
    for path, text in ((keep, "k"), (edit, "e"), (gone, "w")):
        path.write_text(text, encoding="utf-8")
    outside = tmp_path / "B03_FileInput.txt"
    outside.write_text("o", encoding="utf-8")
    snap = guard.snapshot_storage(tmp_path, ("B06_Section", "workflow.json"))
    kept_stamp = keep.stat().st_mtime_ns
    edit.write_text("changed", encoding="utf-8")
    gone.unlink()
    (tmp_path / "B06_Section" / "new.json").write_text("n", encoding="utf-8")
    outside.write_text("o2", encoding="utf-8")  # 범위 밖은 안 건드림
    assert snap.changed() == ["B06_Section/edit.json", "B06_Section/new.json", "workflow.json"]
    assert snap.restore() == 3
    assert edit.read_text(encoding="utf-8") == "e" and gone.read_text(encoding="utf-8") == "w"
    assert not (tmp_path / "B06_Section" / "new.json").exists()
    assert keep.stat().st_mtime_ns == kept_stamp and outside.read_text(encoding="utf-8") == "o2"
    snap.discard()
    assert not snap.copy_dir.exists()


# ── 실제 창구 ──────────────────────────────────────────────────────────────


def _db_ready() -> Path | None:
    try:
        from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
        from common_util.common_util_storage import resolve_stored_project_path
        from config.config_db import close_db_pool, init_db_pool

        async def find():
            pool = await init_db_pool()
            try:
                async with pool.acquire() as connection:
                    return await get_project_storage_relative_path(connection, UUID(PROJECT))
            finally:
                await close_db_pool()

        root = Path(resolve_stored_project_path(asyncio.run(find())))
        return root if (root / "B06_Section" / "longitudinal").is_dir() else None
    except Exception:
        return None


SHARED_ROOT = _db_ready()
#: 시험이 쓰는 뿌리 = 공유 storage 의 임시 사본(492 · `sandbox` 가 세움 · Synology 로 안 번지게)
ROOT_DIR: Path | None = None


@pytest.fixture(scope="module")
def sandbox():
    global ROOT_DIR
    from helper_temp_storage import temp_project_storage

    with temp_project_storage(SHARED_ROOT) as root:
        ROOT_DIR = root
        yield root
    ROOT_DIR = None


def live(test):
    """실DB · 사본 storage 시험 — 검증 프로젝트 없으면 건너뜀."""
    return pytest.mark.skipif(SHARED_ROOT is None, reason="검증_sub8 · DB 없음")(
        pytest.mark.usefixtures("sandbox")(test)
    )


def _state(root: Path) -> tuple[dict, dict]:
    """(행 — 시각 칸 뺌, 파일 내용 해시) — 저장 범위 전부."""
    from config.config_db import close_db_pool, init_db_pool

    async def rows():
        pool = await init_db_pool()
        try:
            async with pool.acquire() as connection:
                return await guard.snapshot_rows(connection, PROJECT)
        finally:
            await close_db_pool()

    raw = asyncio.run(rows())
    clean = {
        table: [{k: str(v) for k, v in row.items() if k not in VOLATILE} for row in values]
        for table, values in raw.items()
    }
    scope = guard.STORAGE_SCOPE + guard.PIPE_SCOPE
    files = {
        relative: hashlib.sha1(path.read_bytes()).hexdigest()
        for relative, path in guard._files(root, scope).items()
        if not relative.startswith("_derived/")
    }
    return clean, files


def _latest_route() -> tuple[int, float]:
    """(최신 노선 id — 저장 창구가 견주는 그 값, 추가 측점 자리 — 가운데 두 측점 사이)."""
    from B06_Section.B06_Section_Repository import get_workflow_route_context
    from config.config_db import close_db_pool, init_db_pool

    async def run():
        pool = await init_db_pool()
        try:
            async with pool.acquire() as connection:
                return await get_workflow_route_context(connection, UUID(PROJECT))
        finally:
            await close_db_pool()

    route_id = int(asyncio.run(run())["route_id"])
    longitudinal = ROOT_DIR / "B06_Section" / "longitudinal" / "longitudinal.json"
    stations = json.loads(longitudinal.read_text(encoding="utf-8"))["stations"]
    middle = len(stations) // 2
    a, b = (float(stations[i]["chainage_m"]) for i in (middle - 1, middle))
    return route_id, round((a + b) / 2, 2)


def _without_ids(rows: dict) -> dict:
    """행 자동 번호(`id`) 뺀 비교 꼴 — 새 추가 측점 행은 두 길이 다른 번호를 받음."""
    return {
        table: sorted(
            json.dumps({k: v for k, v in row.items() if k != "id"}, sort_keys=True)
            for row in values
        )
        for table, values in rows.items()
    }


def _request(**extra):
    from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest

    route_id, plus_at = _latest_route()
    body = {
        "route_id": route_id,
        "settings": {"design_speed_kph": 20, "terrain_type": "special", "max_grade_pct": 13.0},
        "plus_stations": {"chainages": [plus_at]},
        "cross": {},
        **extra,
    }
    return DesignSaveRequest.model_validate(body)


def _save(request, monkeypatch=None):
    from B06_Section.B06_Section_Router_DesignSave import save_design
    from config.config_db import close_db_pool, init_db_pool

    async def run():
        await init_db_pool()
        try:
            return await save_design(UUID(PROJECT), request, None)
        finally:
            await close_db_pool()

    response = asyncio.run(run())
    return response.status_code, json.loads(bytes(response.body).decode("utf-8"))


@live
def test_중간_차례_실패면_DB_storage_그대로(monkeypatch) -> None:
    from B06_Section import B06_Section_Router_Stations as stations

    before = _state(ROOT_DIR)

    async def broken(project_id, request):
        (ROOT_DIR / "B06_Section" / "half_written.json").write_text("{}", encoding="utf-8")
        raise RuntimeError("일부러 실패")

    monkeypatch.setattr(stations, "put_plus_stations", broken)
    status, body = _save(_request())
    assert status == 409 and body["step"] == "plusStations" and body["restored"] is True
    assert "일부러 실패" in body["message"]
    after = _state(ROOT_DIR)
    assert after[0] == before[0]  # 앞 차례(페이지 설정)가 쓴 단계 params 도 되돌림
    assert after[1] == before[1]


@live
def test_정상_한_본문은_옛_차례와_같음() -> None:
    from B05_Profile.B05_Profile_Router_Lifecycle import save_page_settings
    from B06_Section.B06_Section_Router_Confirm import save_sections
    from B06_Section.B06_Section_Router_DesignSave import _save_stages
    from B06_Section.B06_Section_Router_Stations import put_plus_stations
    from config.config_db import close_db_pool, get_db_pool, init_db_pool

    start = _state(ROOT_DIR)
    request = _request()

    async def old_way():
        await init_db_pool()
        try:
            async with get_db_pool().acquire() as connection:
                rows = await guard.snapshot_rows(connection, PROJECT)
            files = guard.snapshot_storage(ROOT_DIR, guard.STORAGE_SCOPE)
            await save_page_settings(UUID(PROJECT), request.settings)
            await put_plus_stations(UUID(PROJECT), request.plus_stations)
            await save_sections(UUID(PROJECT), request.route_id, request.cross)
            await _save_stages(PROJECT)  # 298-2 [저장] = 단계 2 · 3 진행 중(B1)
            return rows, files
        finally:
            await close_db_pool()

    rows, files = asyncio.run(old_way())
    old = _state(ROOT_DIR)

    async def undo():
        await init_db_pool()
        try:
            files.restore()
            async with get_db_pool().acquire() as connection:
                await guard.restore_rows(connection, PROJECT, rows)
        finally:
            await close_db_pool()

    asyncio.run(undo())
    assert _state(ROOT_DIR) == start
    try:
        status, body = _save(request)
        assert status == 200 and body["status"] == "success", body
        assert set(body["steps"]) == {"settings", "plusStations", "cross"}
        new = _state(ROOT_DIR)
        assert _without_ids(new[0]) == _without_ids(old[0])
        assert new[1] == old[1]
    finally:
        asyncio.run(undo())
        files.discard()
    assert _state(ROOT_DIR) == start


@live
def test_되돌림마저_실패면_복구_확인_필요(monkeypatch) -> None:
    from B05_Profile import B05_Profile_Router_Lifecycle as lifecycle

    async def broken(project_id, request):
        raise RuntimeError("설정 실패")

    def no_restore(self):
        raise OSError("디스크 잠김")

    monkeypatch.setattr(lifecycle, "save_page_settings", broken)
    monkeypatch.setattr(guard.StorageSnapshot, "restore", no_restore)
    status, body = _save(_request())
    assert status == 500 and body["restored"] is False and body["step"] == "settings"
    assert "복구 확인 필요" in body["message"]


@live
def test_같은_프로젝트_저장이_겹치면_뒤_요청이_기다림() -> None:
    from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest, save_design
    from config.config_db import close_db_pool, init_db_pool

    route_id = _latest_route()[0]

    async def run():
        await init_db_pool()
        try:
            lock = guard.project_lock(PROJECT)
            await lock.acquire()
            task = asyncio.create_task(
                save_design(UUID(PROJECT), DesignSaveRequest(route_id=route_id), None)
            )
            await asyncio.sleep(0.3)
            assert not task.done()  # 앞 저장(잠금)이 끝날 때까지 기다림
            lock.release()
            return await task
        finally:
            await close_db_pool()

    response = asyncio.run(run())
    body = json.loads(bytes(response.body).decode("utf-8"))
    assert body["status"] == "success" and body["waited"] is True and body["steps"] == {}


@live
def test_시험_갈래_환경_변수면_그_차례_뒤_실패_되돌림(monkeypatch) -> None:
    before = _state(ROOT_DIR)
    monkeypatch.setenv("AISLO_DESIGN_SAVE_FAIL_AFTER", "settings")
    status, body = _save(_request())
    assert status == 409 and body["step"] == "settings" and body["restored"] is True
    assert _state(ROOT_DIR) == before