Aislo/resources/tester/test_initial_snapshot.py

"""초기값 스냅샷 — 촬영·복원이 작업본을 초기 상태로 되돌리는지 확인한다.

DB는 aiomysql 대신 최소 가짜 커넥션으로 대신한다. 확인하려는 것은 스냅샷 모듈의
규약(무엇을 뜨고, 무엇을 되돌리고, 무엇을 건드리지 않는가)이지 드라이버가 아니다.
"""

import asyncio
import json
from pathlib import Path

import pytest

from common_util.common_util_initial_snapshot import (
    has_initial_snapshot,
    restore_initial_snapshot,
    restore_snapshot_files,
    save_initial_snapshot,
    snapshot_dir,
)

ROUTE_ROW = {"id": 7, "project_id": "p1", "status": "CONFIRMED", "total_length_m": 123.4}


class _Cursor:
    """SELECT는 준비된 행을, INSERT는 새 id를 돌려주는 최소 커서."""

    def __init__(self, state):
        self._state = state
        self.lastrowid = 0
        self._rows = []

    async def __aenter__(self):
        return self

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

    async def execute(self, sql, params=None):
        head = sql.strip().split()[0].upper()
        if head == "INSERT":
            self._state["inserted"].append((sql, params))
            self._state["next_id"] += 1
            self.lastrowid = self._state["next_id"]
            return
        if "FROM routes" in sql:
            self._rows = [dict(ROUTE_ROW)]
        else:
            table = sql.split("FROM ")[1].split()[0]
            self._rows = list(self._state["children"].get(table, []))

    async def fetchone(self):
        return self._rows[0] if self._rows else None

    async def fetchall(self):
        return self._rows


class _Connection:
    def __init__(self, state):
        self._state = state

    def cursor(self, *_args, **_kwargs):
        return _Cursor(self._state)


@pytest.fixture
def project(tmp_path: Path):
    """정본 파일이 든 프로젝트 루트를 만든다."""
    for tree, name, text in (
        ("B05_Profile/route", "structures.json", '{"v": "initial"}'),
        ("B06_Section/longitudinal", "long.json", "initial"),
        ("B06_Section/cross_sections", "cross.json", "initial"),
        ("B04_PreProcess/drainage/edits", "pipe_points.json", '{"pipes": []}'),
    ):
        folder = tmp_path / tree
        folder.mkdir(parents=True, exist_ok=True)
        (folder / name).write_text(text, encoding="utf-8")
    return tmp_path


@pytest.fixture
def state():
    return {
        "inserted": [],
        "next_id": 100,
        "children": {
            "route_points": [{"id": 1, "route_id": 7, "chainage_m": 0.0}],
            "route_statistics": [{"id": 2, "route_id": 7, "min_slope": 1.0}],
            "longitudinal_sections": [{"id": 3, "route_id": 7, "project_id": "p1"}],
            "cross_sections": [{"id": 4, "route_id": 7, "project_id": "p1"}],
        },
    }


def test_snapshot_captures_files_and_rows(project, state):
    assert not has_initial_snapshot(project)
    asyncio.run(save_initial_snapshot(_Connection(state), project, 7))

    assert has_initial_snapshot(project)
    dump = json.loads((snapshot_dir(project) / "db.json").read_text(encoding="utf-8"))
    assert dump["routes"][0]["total_length_m"] == 123.4
    assert len(dump["cross_sections"]) == 1
    # 파일 트리 4개가 통째로 떠 있어야 한다.
    # 배수유역은 `edits/` 만이 아니라 **폴더 통째**로 뜬다(2026-09-04 사용자 확정).
    saved = snapshot_dir(project) / "B04_PreProcess__drainage" / "edits" / "pipe_points.json"
    assert saved.read_text(encoding="utf-8") == '{"pipes": []}'


def test_snapshot_is_taken_once(project, state):
    asyncio.run(save_initial_snapshot(_Connection(state), project, 7))
    # 촬영 뒤 사용자가 정본을 고쳐도 스냅샷은 그대로여야 한다(읽기 전용 층).
    (project / "B05_Profile/route/structures.json").write_text('{"v": "edited"}', encoding="utf-8")
    asyncio.run(save_initial_snapshot(_Connection(state), project, 7))

    kept = snapshot_dir(project) / "B05_Profile__route" / "structures.json"
    assert json.loads(kept.read_text(encoding="utf-8"))["v"] == "initial"


def test_restore_overwrites_edited_masters(project, state):
    """편집분이 남아 초기값이 오염되던 것이 이 복원으로 사라진다."""
    asyncio.run(save_initial_snapshot(_Connection(state), project, 7))
    edited = project / "B04_PreProcess/drainage/edits/pipe_points.json"
    edited.write_text('{"pipes": [{"chainage_m": 30}]}', encoding="utf-8")
    (project / "B05_Profile/route/extra.json").write_text("사용자가 만든 것", encoding="utf-8")

    restore_snapshot_files(project)

    assert edited.read_text(encoding="utf-8") == '{"pipes": []}'
    # 스냅샷에 없던 파일은 남지 않는다(트리 통째 교체).
    assert not (project / "B05_Profile/route/extra.json").exists()


def test_restore_reinserts_rows_with_new_route_id(project, state):
    asyncio.run(save_initial_snapshot(_Connection(state), project, 7))
    connection = _Connection(state)
    new_id = asyncio.run(restore_initial_snapshot(connection, project, "p1"))

    assert new_id == 101  # routes 1행이 먼저 들어간다
    tables = [sql.split("INSERT INTO ")[1].split()[0] for sql, _ in state["inserted"]]
    assert tables == [
        "routes",
        "route_points",
        "route_statistics",
        "longitudinal_sections",
        "cross_sections",
    ]
    # 자식 행은 옛 id가 아니라 새 route_id를 달고 들어가야 한다.
    child_sql, child_params = state["inserted"][1]
    columns = child_sql.split("(")[1].split(")")[0].replace("`", "").split(", ")
    assert child_params[columns.index("route_id")] == new_id
    assert "id" not in columns  # AUTO_INCREMENT 자리는 비워 둔다


def test_restore_without_snapshot_returns_none(tmp_path, state):
    assert asyncio.run(restore_initial_snapshot(_Connection(state), tmp_path, "p1")) is None


# ── 계산 중 진입 차단 마커 · 스냅샷 무효화 · 편집 정본 제거 (2026-08-29) ──


def test_designing_marker_lifecycle(project):
    from common_util.common_util_initial_snapshot import (
        clear_designing,
        designing_lock_path,
        is_designing,
        mark_designing,
    )

    assert not is_designing(project)
    mark_designing(project)
    assert is_designing(project)
    # 마커는 스냅샷 대상 트리 밖(프로젝트 루트)에 있어야 복원에 딸려 들어가지 않는다.
    assert designing_lock_path(project).parent == project
    clear_designing(project)
    assert not is_designing(project)
    clear_designing(project)  # 두 번 내려도 터지지 않는다


def test_marker_not_captured_by_snapshot(project, state):
    from common_util.common_util_initial_snapshot import mark_designing

    mark_designing(project)
    asyncio.run(save_initial_snapshot(_Connection(state), project, 7))
    captured = [p.name for p in snapshot_dir(project).rglob("*") if p.is_file()]
    assert "initial_design.lock" not in captured


def test_discard_initial_snapshot(project, state):
    from common_util.common_util_initial_snapshot import (
        discard_initial_snapshot,
        has_initial_snapshot,
    )

    asyncio.run(save_initial_snapshot(_Connection(state), project, 7))
    assert has_initial_snapshot(project)
    assert discard_initial_snapshot(project) is True
    assert not has_initial_snapshot(project)
    # 없는 것을 지우면 False — 호출부가 로그를 남길지 판단한다.
    assert discard_initial_snapshot(project) is False


def test_wipe_edited_masters(project):
    """폴백 재계산이 진짜 초기값을 만들려면 이 둘이 먼저 사라져야 한다."""
    from common_util.common_util_initial_snapshot import wipe_edited_masters

    removed = wipe_edited_masters(project)

    assert sorted(removed) == [
        "B04_PreProcess/drainage/edits",
        "B05_Profile/route/structures.json",
    ]
    assert not (project / "B05_Profile/route/structures.json").exists()
    assert not (project / "B04_PreProcess/drainage/edits").exists()
    # 다른 정본은 건드리지 않는다.
    assert (project / "B06_Section/longitudinal/long.json").exists()
    assert wipe_edited_masters(project) == []  # 두 번 불러도 조용히 끝난다