Aislo/resources/tester/test_590_1_plan_file.py

# -*- coding: utf-8 -*-
"""590-1 예정노선 설계 내용 = 프로젝트 폴더 한 파일(요구 452 · 453) — 지킴 시험.

- 옮김 = 옛 DB 칸(지점 points · starts/ends · start/end · 빈 칸 = 연도 행 접기 · 노선 정본 ·
  옛 연도 노선) → 새 꼴 한 벌(좌표계 머리 한 곳 · 빈 연도 없음)
- 원자 쓰기 = 바꿔치기 실패에도 옛 파일 그대로 · 임시 파일 안 남음 · 새 판 파일 거절
- 연도 읽기 = 업체 줄 연도는 파일 조각에서(낡은 DB 연도 무시 · ⑤) · 확정 = 지문 같을 때만
- 잠금 = 같은 프로젝트 [저장] 둘이 겹치지 않음

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

from __future__ import annotations

import asyncio
import json
import sys
from pathlib import Path

import pytest
from fastapi import HTTPException

sys.path.insert(0, str(Path(__file__).parent))

from helper_plan_file import use_plan, year_line_plan  # noqa: E402

from common_util.common_util_crs import lonlat_to_tm  # noqa: E402
from D03_PlanAnalysis import D03_PlanAnalysis_Network as net  # noqa: E402
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store  # noqa: E402
from D04_PlanConfirm import D04_PlanConfirm_Router as d04  # noqa: E402

PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
A = {"x": 210000.0, "y": 380000.0}
B = {"x": 210300.0, "y": 380400.0}
NETWORK = {
    "epsg": 5186,
    "lines": [{"id": "main", "vertices": [[0.0, 0.0], [100.0, 0.0]], "branch": None}],
    "marks": [{"line": "main", "at_m": 40.0}],
    "pieces": [
        {"id": "p1", "line": "main", "from_m": 0.0, "to_m": 40.0, "year": 2027},
        {"id": "p2", "line": "main", "from_m": 40.0, "to_m": 100.0, "year": 2029},
    ],
}


# --- 옮김 ---


@pytest.mark.parametrize(
    "base",
    [
        {"epsg": 5186, "points": [A, B]},
        {"epsg": 5186, "starts": [A], "ends": [B]},
        {"epsg": 5186, "start": A, "end": B},
    ],
)
def test_migrate_old_point_forms(base):
    plan = store.migrate(json.dumps(base), json.dumps(NETWORK), [])
    assert plan["points"] == [A, B] and plan["epsg"] == 5186
    assert plan["lines"] == NETWORK["lines"] and plan["pieces"] == NETWORK["pieces"]
    assert set(plan) == {"version", "epsg", "points", "lines", "marks", "pieces"}


def test_migrate_folds_year_rows_and_legacy_route():
    rows = [
        {"year": 2027, "epsg": 5186, "route_json": None,
         "start_x": 1.0, "start_y": 2.0, "end_x": 3.0, "end_y": 4.0},
        {"year": 2026, "epsg": 5186, "route_json": json.dumps({"planned": [[0, 0], [30, 40]]}),
         "start_x": 0.0, "start_y": 0.0, "end_x": 1.0, "end_y": 2.0},
        {"year": 2028, "epsg": None, "route_json": None,
         "start_x": None, "start_y": None, "end_x": None, "end_y": None},  # 빈 연도 줄
    ]  # fmt: skip
    plan = store.migrate(None, None, rows)
    assert plan["points"] == [{"x": 0.0, "y": 0.0}, {"x": 3.0, "y": 4.0}]  # 첫 시점 · 끝 종점
    assert [p["year"] for p in plan["pieces"]] == [2026]  # 빈 연도는 조각이 아님
    assert store.plan_units(plan)[0]["length_m"] == 50.0
    assert store.migrate(None, None, []) == store.empty_plan()


def test_migrate_one_epsg_converts_points():
    lon_lat = (127.1, 37.5)
    x, y = lonlat_to_tm(*lon_lat, 5185)
    plan = store.migrate(
        json.dumps({"epsg": 5185, "points": [{"x": x, "y": y}]}), json.dumps(NETWORK), []
    )
    assert plan["epsg"] == 5186
    assert (plan["points"][0]["x"], plan["points"][0]["y"]) == pytest.approx(
        lonlat_to_tm(*lon_lat, 5186), abs=1e-3
    )


# --- 원자 쓰기 · 판 ---


def test_atomic_write_keeps_old_file_on_failure(tmp_path, monkeypatch):
    plan = {**store.empty_plan(5186), **NETWORK}
    store.save_plan(tmp_path, plan)
    path = store.plan_path(tmp_path)
    assert (
        path.parent.name == "D03_PlanAnalysis"
        and json.loads(path.read_text("utf-8"))["version"] == 1
    )
    assert store.read_plan(tmp_path) == plan

    import common_util.common_util_json as cj

    def broken(source, target):
        raise PermissionError("잡힘")

    monkeypatch.setattr(cj, "replace_with_retry", broken)
    with pytest.raises(PermissionError):
        store.save_plan(tmp_path, {**plan, "lines": [], "marks": [], "pieces": []})
    assert store.read_plan(tmp_path) == plan  # 옛 파일 그대로
    assert [p.name for p in path.parent.iterdir()] == ["plan_network.json"]  # 임시 파일 안 남음


def test_newer_version_refused(tmp_path):
    path = store.plan_path(tmp_path)
    path.parent.mkdir(parents=True)
    path.write_text(json.dumps({**store.empty_plan(), "version": 2}), "utf-8")
    with pytest.raises(HTTPException) as err:
        store.read_plan(tmp_path)
    assert err.value.status_code == 409
    assert store.read_plan(tmp_path / "없음") is None


# --- 연도 읽기(⑤) ---


def test_assignee_year_comes_from_file(monkeypatch):
    use_plan(monkeypatch, {**store.empty_plan(5186), **NETWORK})
    _, units = asyncio.run(d04._units(PID))
    stale = [{"id": 27, "piece_id": "p1", "year": 2028}, {"id": 28, "piece_id": "p2", "year": 2027}]
    assert [a["year"] for a in d04.with_years(stale, units)] == [2027, 2029]
    assert d04.with_years([{"id": 1, "piece_id": "없음", "year": 2027}], units)[0]["year"] is None


def test_confirmed_only_when_fingerprint_matches(monkeypatch):
    plan = year_line_plan({2026: [[0.0, 0.0], [10.0, 0.0]], 2027: [[10.0, 0.0], [10.0, 10.0]]})
    state = use_plan(monkeypatch, plan, {2026: "x"})
    years = asyncio.run(store.plan_state(PID))["years"]
    assert years[2026]["confirmed_at"] == "x" and years[2027]["confirmed_at"] is None
    moved = json.loads(json.dumps(plan))
    moved["lines"][0]["vertices"][0] = [20.0, 0.0]  # 2026 조각 모양 바뀜(길이 같음)

    async def stale(_pid):  # DB 지문 = 옛 파일 것
        return {2026: {"fingerprint": years[2026]["fingerprint"], "confirmed_at": "x"}}

    monkeypatch.setattr(store, "year_status", stale)
    state["plan"] = moved
    assert asyncio.run(store.plan_state(PID))["years"][2026]["confirmed_at"] is None


# --- 잠금 ---


def test_two_saves_do_not_interleave(monkeypatch, tmp_path):
    order: list[str] = []

    async def load(_pid):
        return tmp_path, store.read_plan(tmp_path) or store.empty_plan()

    async def project_info(_pid):
        return {"year_start": 2026, "site": None}

    async def slow_status(connection, project_id, pieces, years):
        order.append(f"start {len(pieces)}")
        await asyncio.sleep(0.05)
        order.append(f"end {len(pieces)}")

    class Conn:
        async def __aenter__(self):
            return self

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

        async def begin(self):
            pass

        async def commit(self):
            pass

        async def rollback(self):
            pass

    class Pool:
        def acquire(self):
            return Conn()

    monkeypatch.setattr(store, "load_plan", load)
    monkeypatch.setattr(net, "project_info", project_info)
    monkeypatch.setattr(net, "_store_status", slow_status)
    monkeypatch.setattr(net, "get_db_pool", lambda: Pool())
    forest = {"role": "SYSTEM_ADMIN", "email": "a@b.kr"}
    one = net.Network(**{**NETWORK, "pieces": NETWORK["pieces"][:1]})
    two = net.Network(**NETWORK)

    async def both():
        from uuid import UUID

        await asyncio.gather(
            net.put_plan_network(UUID(PID), one, forest),
            net.put_plan_network(UUID(PID), two, forest),
        )

    asyncio.run(both())
    assert order == ["start 1", "end 1", "start 2", "end 2"]
    assert len(store.read_plan(tmp_path)["pieces"]) == 2  # 뒤 저장이 이김 · 반쪽 없음