Aislo/resources/tester/test_280_3_migrate_guard.py

# -*- coding: utf-8 -*-
"""요구 280-3 서버 — 옛 비정규 측점 이관 막이 · 이미 생긴 「기타 구조물」 알림 · 맞춤 실패 알림.

못박는 것(임시 폴더 · 공유 storage 안 씀)
  - 이름 빈 측점(추가 측점 정본 소관) = 이관 0 — 「기타 구조물」로 안 옮김.
  - 종단 정본이 지금 심는 자리를 다 못 세움(저장 도중 끊김 · 다른 탭 [저장] 도중) = 이관 0.
  - 정본 맞는 옛 자유 글(「대피로 2.0m」)은 종전대로 옮김.
  - 옛 이관이 만든 「기타 구조물」(이관 기록 · 이름 그대로 · 메모 없음) = `legacy_etc` · 안 지움.
  - 이관 = [저장]과 같은 프로젝트 잠금 안(앞 저장이 끝날 때까지 기다림).
  - 맞출 측점이 있는데 샘플링 조건 없음 = `StationsUnsynced` · 추가 측점 PUT `stations_synced` 거짓.
판 = `test_452_server_keys_prune_migrate.migrate_client`.
"""

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

import pytest

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

import test_452_server_keys_prune_migrate as base  # noqa: E402

import B06_Section.B06_Section_Router_Stations as stations  # noqa: E402
from common_util import common_util_save_guard as guard  # noqa: E402
from common_util.common_util_drainage_pipes import pipe_points_path_in  # noqa: E402

PROJECT_ID = base.PROJECT_ID
migrate_client = base.migrate_client  # 판 그대로
router_module = base.router_module
_write_longitudinal = base._write_longitudinal

URL = f"/api/projects/{PROJECT_ID}/route/structures/migrate"
STRUCTURES = Path("B05_Profile") / "route" / "structures.json"


def _pipe(project: Path, chainage: float) -> None:
    path = pipe_points_path_in(project)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps({"route_signature": "s", "points": [{"chainage_m": chainage, "id": "p1"}]}),
        encoding="utf-8",
    )


def _line(*irregular: tuple[float, str]) -> list[dict]:
    return [
        {"chainage_m": 0.0, "kind": "bp"},
        *({"chainage_m": at, "kind": "irregular", "structure": name} for at, name in irregular),
        {"chainage_m": 400.0, "kind": "ep"},
    ]


def test_이름_빈_측점은_이관_안_함(migrate_client):
    client, project = migrate_client
    _write_longitudinal(project, _line((376.0, "")))
    response = client.post(URL, json={"route_id": None})
    assert response.status_code == 200, response.text
    assert response.json()["migrated"] == 0
    stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
    assert stored == []


def test_낡은_종단은_이관_안_함(migrate_client):
    client, project = migrate_client
    _pipe(project, 351.0)  # 지금 관 = 351 · 종단은 옛 자리 330 만(저장 도중 끊김)
    _write_longitudinal(project, _line((330.0, "파형강관 D800"), (600.0, "맹암거(구판)")))
    body = client.post(URL, json={"route_id": None}).json()
    assert body["migrated"] == 0 and body["skipped"] == "stale_longitudinal"
    assert not (project / STRUCTURES).exists()


def test_정본_맞는_옛_자유글은_옮김(migrate_client):
    client, project = migrate_client
    _pipe(project, 351.0)
    _write_longitudinal(project, _line((300.0, "대피로 2.0m"), (351.0, "파형강관 D800")))
    body = client.post(URL, json={"route_id": None}).json()
    assert body["migrated"] == 1 and "skipped" not in body
    stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
    assert [s["type_id"] for s in stored] == ["refuge"]


def test_이미_생긴_기타_구조물은_알림만(migrate_client):
    client, project = migrate_client
    _write_longitudinal(project, _line((376.0, "")))
    path = project / STRUCTURES
    path.parent.mkdir(parents=True, exist_ok=True)
    etc = {
        "structure_id": "e1",
        "type_id": "etc",
        "placement": "point",
        "chainage_m": 376.0,
        "options": {"name": "기타 구조물"},
    }
    named = {**etc, "structure_id": "e2", "chainage_m": 200.0, "options": {"name": "돌망태"}}
    path.write_text(
        json.dumps(
            {
                "revision": 1,
                "structures": [etc, named],
                "migrated_legacy": ["etc@376.0", "etc@200.0"],
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    body = client.post(URL, json={"route_id": None}).json()
    assert body["legacy_etc"] == [376.0]  # 이름 단 것(돌망태)은 사용자 것
    stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
    assert sorted(s["structure_id"] for s in stored) == ["e1", "e2"]  # 지우지 않음


def test_이관은_저장_잠금을_기다림(monkeypatch):
    seen: list[str] = []

    async def body(project_id, payload):
        seen.append("migrate")
        return "done"

    monkeypatch.setattr(router_module, "_migrate_structures", body)

    async def run() -> None:
        lock = guard.project_lock(PROJECT_ID)
        async with lock:
            task = asyncio.create_task(
                router_module.migrate_structures(
                    UUID(PROJECT_ID), router_module.StructureMigrateRequest()
                )
            )
            await asyncio.sleep(0.05)
            assert seen == []  # 저장 도중 = 기다림
        assert await task == "done" and seen == ["migrate"]

    asyncio.run(run())


def test_샘플링_조건_없으면_맞춤_실패_알림(monkeypatch, tmp_path):
    from B05_Profile import B05_Profile_Engine_Sections as sections
    from B05_Profile import B05_Profile_Plus_Stations as plus
    from B05_Profile import B05_Profile_Router_Confirm as confirm

    async def paths(_project_id):
        return tmp_path, {"route_data_path": ""}

    async def nothing(*_args, **_kwargs):
        return None

    monkeypatch.setattr(stations, "_project_paths", paths)
    monkeypatch.setattr(stations, "_missing_marks", lambda *_: [120.0])
    monkeypatch.setattr(stations, "_stale_marks", lambda *_: 0)
    monkeypatch.setattr(stations, "_curve_marks", lambda *_: 0)
    monkeypatch.setattr(stations, "run_with_connection", nothing)
    monkeypatch.setattr(stations, "_sampling_conditions", nothing)
    monkeypatch.setattr(confirm, "_section_options_from_stored", lambda *_: None)
    with pytest.raises(stations.StationsUnsynced):
        asyncio.run(stations.ensure_structure_stations(UUID(PROJECT_ID)))

    monkeypatch.setattr(sections, "_load_route_polyline", lambda *_: [])
    monkeypatch.setattr(plus, "save_plus_stations", lambda _root, chainages, _xy: list(chainages))
    response = asyncio.run(
        stations.put_plus_stations(UUID(PROJECT_ID), stations.PlusStationsRequest(chainages=[120]))
    )
    body = json.loads(response.body)
    assert body["status"] == "success" and body["chainages"] == [120]
    assert body["stations_synced"] is False  # 저장은 그대로 · 맞춤 실패 알림

    monkeypatch.setattr(stations, "_missing_marks", lambda *_: [])
    assert (
        asyncio.run(stations.ensure_structure_stations(UUID(PROJECT_ID))) == 0
    )  # 맞출 것 없음 = 조용